From 9e8a7c77a94a92c6cf92cf60be508817f0778040 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Thu, 18 Jan 2007 06:27:50 +0000 Subject: There are a ton of FIXMES and request/response IDs are not yet working fully but all tests are passing. * broker::Broker: Removed requester/responder from broker. * framing::BodyHandler: added Requester/Responder to BodyHandler, becomes the base class for channel adapters in broker and client. * broker::BrokerAdapter: Inherit BodyHandler, wraps a broker::Channel. Hide private *HandlerImpl detail classes in BodyHandler.cpp. * broker::Connection: Requester/Responder/Adapter now per-channel. Connection channel map replaced with adapter map of BrokerAdapters. handle* functions moved to BrokerAdapter. All methods now handled by a BrokerAdapter for the relevant channel. ChannelHandlerImpl is repsonsible for checking that - No method on a non-0 channel is processed before open() - Channel 0 methods only happen on channel 0 and similar for non-zero methods Checks are not yet complete (see FIXMES) * client::ResponseHandler: fix for client hang if broker crashs. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@497319 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/lib/common/framing/BodyHandler.cpp | 52 ++++++++++++++++++------- cpp/lib/common/framing/BodyHandler.h | 71 ++++++++++++++++++++++------------ 2 files changed, 84 insertions(+), 39 deletions(-) (limited to 'cpp/lib/common/framing') diff --git a/cpp/lib/common/framing/BodyHandler.cpp b/cpp/lib/common/framing/BodyHandler.cpp index 1f761bf3f1..72ba82468f 100644 --- a/cpp/lib/common/framing/BodyHandler.cpp +++ b/cpp/lib/common/framing/BodyHandler.cpp @@ -18,38 +18,62 @@ * under the License. * */ -#include -#include +#include "QpidError.h" +#include "BodyHandler.h" +#include +#include +#include +#include +#include +#include using namespace qpid::framing; using namespace boost; BodyHandler::~BodyHandler() {} -void BodyHandler::handleBody(const AMQBody::shared_ptr& body){ - +void BodyHandler::handleBody(shared_ptr body) { switch(body->type()) { - case METHOD_BODY: case REQUEST_BODY: + handleRequest(shared_polymorphic_cast(body)); + break; case RESPONSE_BODY: - handleMethod(dynamic_pointer_cast(body)); + handleResponse(shared_polymorphic_cast(body)); + break; + case METHOD_BODY: + handleMethod(shared_polymorphic_cast(body)); break; - case HEADER_BODY: - handleHeader(dynamic_pointer_cast(body)); + handleHeader(shared_polymorphic_cast(body)); break; - case CONTENT_BODY: - handleContent(dynamic_pointer_cast(body)); + handleContent(shared_polymorphic_cast(body)); break; - case HEARTBEAT_BODY: - handleHeartbeat(dynamic_pointer_cast(body)); + handleHeartbeat(shared_polymorphic_cast(body)); break; - default: - throw UnknownBodyType(body->type()); + QPID_ERROR(PROTOCOL_ERROR, "Unknown frame type "+body->type()); } +} + +void BodyHandler::handleRequest(AMQRequestBody::shared_ptr request) { + responder.received(request->getData()); + handleMethod(request); +} + +void BodyHandler::handleResponse(AMQResponseBody::shared_ptr response) { + handleMethod(response); + requester.processed(response->getData()); +} + +void BodyHandler::assertChannelZero(u_int16_t id) { + if (id != 0) + throw ConnectionException(504, "Invalid channel id, not 0"); +} +void BodyHandler::assertChannelNonZero(u_int16_t id) { + if (id == 0) + throw ConnectionException(504, "Invalid channel id 0"); } diff --git a/cpp/lib/common/framing/BodyHandler.h b/cpp/lib/common/framing/BodyHandler.h index 0260a35e88..c9c74e2b3f 100644 --- a/cpp/lib/common/framing/BodyHandler.h +++ b/cpp/lib/common/framing/BodyHandler.h @@ -1,3 +1,6 @@ +#ifndef _BodyHandler_ +#define _BodyHandler_ + /* * * Licensed to the Apache Software Foundation (ASF) under one @@ -18,37 +21,55 @@ * under the License. * */ -#include -#ifndef _BodyHandler_ -#define _BodyHandler_ +#include -#include -#include -#include -#include +#include "Requester.h" +#include "Responder.h" namespace qpid { namespace framing { - class BodyHandler{ - public: - virtual ~BodyHandler(); - virtual void handleMethod(AMQMethodBody::shared_ptr body) = 0; - virtual void handleHeader(AMQHeaderBody::shared_ptr body) = 0; - virtual void handleContent(AMQContentBody::shared_ptr body) = 0; - virtual void handleHeartbeat(AMQHeartbeatBody::shared_ptr body) = 0; - - void handleBody(const AMQBody::shared_ptr& body); - }; - - class UnknownBodyType{ - public: - const u_int16_t type; - inline UnknownBodyType(u_int16_t _type) : type(_type){} - }; -} -} +class AMQRequestBody; +class AMQResponseBody; +class AMQMethodBody; +class AMQHeaderBody; +class AMQContentBody; +class AMQHeartbeatBody; + +/** + * Base class for client and broker channel handlers. + * + * Handles request/response id management common to client and broker. + * Derived classes provide remaining client/broker specific handling. + */ +class BodyHandler { + public: + virtual ~BodyHandler(); + + void handleBody(boost::shared_ptr body); + + protected: + virtual void handleRequest(boost::shared_ptr); + virtual void handleResponse(boost::shared_ptr); + + virtual void handleMethod(boost::shared_ptr) = 0; + virtual void handleHeader(boost::shared_ptr) = 0; + virtual void handleContent(boost::shared_ptr) = 0; + virtual void handleHeartbeat(boost::shared_ptr) = 0; + + protected: + /** Throw protocol exception if this is not channel 0. */ + static void assertChannelZero(u_int16_t id); + /** Throw protocol exception if this is channel 0. */ + static void assertChannelNonZero(u_int16_t id); + + private: + Requester requester; + Responder responder; +}; + +}} #endif -- cgit v1.2.1