7 files changed, 230 insertions, 22 deletions
diff --git a/cpp/lib/common/Makefile.am b/cpp/lib/common/Makefile.am
index 941b30f8e5..541145ac97 100644
--- a/cpp/lib/common/Makefile.am
+++ b/cpp/lib/common/Makefile.am
@@ -72,6 +72,8 @@ libqpidcommon_la_SOURCES =			\
   $(framing)/ProtocolInitiation.cpp		\
   $(framing)/ProtocolVersion.cpp		\
   $(framing)/ProtocolVersionException.cpp	\
+  $(framing)/Requester.cpp			\
+  $(framing)/Responder.cpp			\
   $(framing)/Value.cpp				\
   $(gen)/AMQP_ClientProxy.cpp			\
   $(gen)/AMQP_HighestVersion.h			\
diff --git a/cpp/lib/common/framing/AMQRequestBody.h b/cpp/lib/common/framing/AMQRequestBody.h
index 1836a7d49d..74aa398606 100644
--- a/cpp/lib/common/framing/AMQRequestBody.h
+++ b/cpp/lib/common/framing/AMQRequestBody.h
@@ -31,7 +31,17 @@ class AMQRequestBody : public AMQMethodBody
 {
  public:
     typedef boost::shared_ptr<AMQRequestBody> shared_ptr;
-    
+
+    struct Data {
+        Data(RequestId id=0, ResponseId mark=0)
+            : requestId(id), responseMark(mark) {}
+        void encode(Buffer&) const;
+        void decode(Buffer&);
+
+        RequestId requestId;
+        ResponseId responseMark;
+    };
+
     static shared_ptr create(
         AMQP_MethodVersionMap& versionMap, ProtocolVersion version,
         Buffer& buffer);
@@ -51,16 +61,6 @@ class AMQRequestBody : public AMQMethodBody
     static const u_int32_t baseSize() { return AMQMethodBody::baseSize()+16; }
     
   private:
-    struct Data {
-        Data(RequestId id=0, ResponseId mark=0)
-            : requestId(id), responseMark(mark) {}
-        void encode(Buffer&) const;
-        void decode(Buffer&);
-
-        RequestId requestId;
-        ResponseId responseMark;
-    };
-
     Data data;
 };
 
diff --git a/cpp/lib/common/framing/AMQResponseBody.h b/cpp/lib/common/framing/AMQResponseBody.h
index bfd6cb2b9a..d7095d3da0 100644
--- a/cpp/lib/common/framing/AMQResponseBody.h
+++ b/cpp/lib/common/framing/AMQResponseBody.h
@@ -35,6 +35,17 @@ class AMQResponseBody : public AMQMethodBody
   public:
     typedef boost::shared_ptr<AMQResponseBody> shared_ptr;
     
+    struct Data {
+        Data(ResponseId id=0, RequestId req=0, BatchOffset off=0)
+            : responseId(id), requestId(req), batchOffset(off) {}
+        void encode(Buffer&) const;
+        void decode(Buffer&);
+
+        u_int64_t responseId;
+        u_int64_t requestId;
+        u_int32_t batchOffset;
+    };
+
     static shared_ptr create(
         AMQP_MethodVersionMap& versionMap, ProtocolVersion version,
         Buffer& buffer);
@@ -53,17 +64,6 @@ class AMQResponseBody : public AMQMethodBody
   protected:
     static const u_int32_t baseSize() { return AMQMethodBody::baseSize()+20; }
   private:
-    struct Data {
-        Data(ResponseId id=0, RequestId req=0, BatchOffset off=0)
-            : responseId(id), requestId(req), batchOffset(off) {}
-        void encode(Buffer&) const;
-        void decode(Buffer&);
-
-        u_int64_t responseId;
-        u_int64_t requestId;
-        u_int32_t batchOffset;
-    };
-
     Data data;
 };
 
diff --git a/cpp/lib/common/framing/Requester.cpp b/cpp/lib/common/framing/Requester.cpp
new file mode 100644
index 0000000000..1dd3cd4ce9
--- /dev/null
+++ b/cpp/lib/common/framing/Requester.cpp
@@ -0,0 +1,46 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "Requester.h"
+#include "QpidError.h"
+
+namespace qpid {
+namespace framing {
+
+Requester::Requester() : lastId(0), responseMark(0) {}
+
+void Requester::sending(AMQRequestBody::Data& request) {
+    request.requestId = ++lastId;
+    request.responseMark = responseMark;
+    requests.insert(request.requestId);
+}
+
+void Requester::processed(const AMQResponseBody::Data& response) {
+    responseMark = response.responseId;
+    RequestId id = response.requestId;
+    RequestId end = id + response.batchOffset;
+    for ( ; id < end; ++id) {
+        std::set<RequestId>::iterator i = requests.find(id);
+        if (i == requests.end())
+            // TODO aconway 2007-01-12: Verify this is the right exception.
+            THROW_QPID_ERROR(PROTOCOL_ERROR, "Invalid response.");
+        requests.erase(i);
+    }
+}
+
+}} // namespace qpid::framing
diff --git a/cpp/lib/common/framing/Requester.h b/cpp/lib/common/framing/Requester.h
new file mode 100644
index 0000000000..e24848f98a
--- /dev/null
+++ b/cpp/lib/common/framing/Requester.h
@@ -0,0 +1,59 @@
+#ifndef _framing_Requester_h
+#define _framing_Requester_h
+
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include <set>
+#include "AMQRequestBody.h"
+#include "AMQResponseBody.h"
+
+namespace qpid {
+namespace framing {
+
+class AMQRequestBody;
+class AMQResponseBody;
+
+/**
+ * Manage request IDs and the response mark for locally initiated requests.
+ *
+ * THREAD UNSAFE: This class is called as frames are sent or received
+ * sequentially on a connection, so it does not need to be thread safe.
+ */
+class Requester
+{
+  public:
+    Requester();
+
+    /** Called before sending a request to set request data. */
+    void sending(AMQRequestBody::Data&);
+
+    /** Called after processing a response. */
+    void processed(const AMQResponseBody::Data&);
+    
+  private:
+    std::set<RequestId> requests; /** Sent but not responded to */
+    RequestId lastId;
+    ResponseId responseMark;
+};
+
+}} // namespace qpid::framing
+
+
+
+#endif  /*!_framing_Requester_h*/
diff --git a/cpp/lib/common/framing/Responder.cpp b/cpp/lib/common/framing/Responder.cpp
new file mode 100644
index 0000000000..efe3609c7b
--- /dev/null
+++ b/cpp/lib/common/framing/Responder.cpp
@@ -0,0 +1,40 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "Responder.h"
+#include "QpidError.h"
+
+namespace qpid {
+namespace framing {
+
+Responder::Responder() : lastId(0), responseMark(0) {}
+
+void Responder::received(const AMQRequestBody::Data& request) {
+    if (request.responseMark < responseMark || request.responseMark > lastId)
+        THROW_QPID_ERROR(PROTOCOL_ERROR, "Invalid resposne mark");
+    responseMark = request.responseMark;
+}
+
+void Responder::sending(AMQResponseBody::Data& response, RequestId toRequest) {
+    response.responseId = ++lastId;
+    response.requestId = toRequest;
+    response.batchOffset = 0;
+}
+
+}} // namespace qpid::framing
+
diff --git a/cpp/lib/common/framing/Responder.h b/cpp/lib/common/framing/Responder.h
new file mode 100644
index 0000000000..a11967acc1
--- /dev/null
+++ b/cpp/lib/common/framing/Responder.h
@@ -0,0 +1,61 @@
+#ifndef _framing_Responder_h
+#define _framing_Responder_h
+
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include "AMQRequestBody.h"
+#include "AMQResponseBody.h"
+
+namespace qpid {
+namespace framing {
+
+/**
+ * Manage response ids and response mark remotely initianted requests.
+ *
+ * THREAD UNSAFE: This class is called as frames are sent or received
+ * sequentially on a connection, so it does not need to be thread safe.
+ */
+class Responder
+{
+  public:
+    Responder();
+
+    /** Called after receiving a request. */
+    void received(const AMQRequestBody::Data& request);
+
+    /** Called before sending a response to set respose data.  */
+    void sending(AMQResponseBody::Data& response, RequestId toRequest);
+
+    /** Get the ID of the highest response acknowledged by the peer. */
+    ResponseId getResponseMark() { return responseMark; }
+
+    // TODO aconway 2007-01-14: Batching support - store unsent
+    // Response for equality comparison with subsequent responses.
+    // 
+
+  private:
+    ResponseId lastId;
+    ResponseId responseMark;
+};
+
+}} // namespace qpid::framing
+
+
+
+#endif  /*!_framing_Responder_h*/