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

2 files changed, 84 insertions, 39 deletions

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 <boost/shared_ptr.hpp>
-#include <BodyHandler.h>
+#include "QpidError.h"
+#include "BodyHandler.h"
+#include <AMQRequestBody.h>
+#include <AMQResponseBody.h>
+#include <AMQMethodBody.h>
+#include <AMQHeaderBody.h>
+#include <AMQContentBody.h>
+#include <AMQHeartbeatBody.h>
 
 using namespace qpid::framing;
 using namespace boost;
 
 BodyHandler::~BodyHandler() {}
 
-void BodyHandler::handleBody(const AMQBody::shared_ptr& body){
-
+void BodyHandler::handleBody(shared_ptr<AMQBody> body) {
     switch(body->type())
     {
-      case METHOD_BODY:
       case REQUEST_BODY:
+        handleRequest(shared_polymorphic_cast<AMQRequestBody>(body));
+        break; 
       case RESPONSE_BODY:
-	handleMethod(dynamic_pointer_cast<AMQMethodBody, AMQBody>(body));
+        handleResponse(shared_polymorphic_cast<AMQResponseBody>(body));
+        break;
+      case METHOD_BODY:
+	handleMethod(shared_polymorphic_cast<AMQMethodBody>(body));
 	break;
- 
       case HEADER_BODY:
-	handleHeader(dynamic_pointer_cast<AMQHeaderBody, AMQBody>(body));
+	handleHeader(shared_polymorphic_cast<AMQHeaderBody>(body));
 	break;
-
       case CONTENT_BODY:
-	handleContent(dynamic_pointer_cast<AMQContentBody, AMQBody>(body));
+	handleContent(shared_polymorphic_cast<AMQContentBody>(body));
 	break;
-
       case HEARTBEAT_BODY:
-	handleHeartbeat(dynamic_pointer_cast<AMQHeartbeatBody, AMQBody>(body));
+	handleHeartbeat(shared_polymorphic_cast<AMQHeartbeatBody>(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 <string>
 
-#ifndef _BodyHandler_
-#define _BodyHandler_
+#include <boost/shared_ptr.hpp>
 
-#include <AMQMethodBody.h>
-#include <AMQHeaderBody.h>
-#include <AMQContentBody.h>
-#include <AMQHeartbeatBody.h>
+#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<AMQBody> body);
+
+  protected:
+    virtual void handleRequest(boost::shared_ptr<AMQRequestBody>);
+    virtual void handleResponse(boost::shared_ptr<AMQResponseBody>);
+
+    virtual void handleMethod(boost::shared_ptr<AMQMethodBody>) = 0;
+    virtual void handleHeader(boost::shared_ptr<AMQHeaderBody>) = 0;
+    virtual void handleContent(boost::shared_ptr<AMQContentBody>) = 0;
+    virtual void handleHeartbeat(boost::shared_ptr<AMQHeartbeatBody>) = 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