Support for AMQP 0-10 sessions in C++ broker.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@658246 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 37 insertions, 35 deletions

diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
index 3fb2579e8c..6d43dd1789 100644
--- a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
+++ b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
@@ -33,10 +33,8 @@ namespace amqp_0_10 {
 using namespace framing;
 using namespace std;
 
-SessionHandler::SessionHandler() : peer(channel), ignoring(), sendReady(), receiveReady() {}
-
-SessionHandler::SessionHandler(FrameHandler& out, ChannelId ch)
-    : channel(ch, &out), peer(channel), ignoring(false) {}
+SessionHandler::SessionHandler(FrameHandler* out, ChannelId ch)
+    : channel(ch, out), peer(channel), ignoring(false), sendReady(), receiveReady() {}
 
 SessionHandler::~SessionHandler() {}
 
@@ -75,7 +73,7 @@ void SessionHandler::handleIn(AMQFrame& f) {
             checkAttached();
             if (!receiveReady)
                 throw IllegalStateException(QPID_MSG(getState()->getId() << ": Not ready to receive data"));
-            if (!getState()->receiver.record(f))
+            if (!getState()->receiverRecord(f))
                 return; // Ignore duplicates.
             getInHandler()->handle(f);
         }
@@ -100,10 +98,10 @@ void SessionHandler::handleOut(AMQFrame& f) {
     checkAttached();
     if (!sendReady)
         throw IllegalStateException(QPID_MSG(getState()->getId() << ": Not ready to send data"));
-    getState()->sender.record(f); 
-    if (getState()->sender.needFlush()) {
+    getState()->senderRecord(f); 
+    if (getState()->senderNeedFlush()) {
         peer.flush(false, true, true); 
-        getState()->sender.recordFlush();
+        getState()->senderRecordFlush();
     }
     channel.handle(f);
 }
@@ -128,7 +126,7 @@ void SessionHandler::attach(const std::string& name, bool force) {
     if (getState()->hasState())
         peer.flush(true, true, true);
     else
-        sendCommandPoint();
+        sendCommandPoint(getState()->senderGetCommandPoint());
 }
 
 void SessionHandler::attached(const std::string& name) {
@@ -168,7 +166,7 @@ void SessionHandler::timeout(uint32_t t) {
 
 void SessionHandler::commandPoint(const SequenceNumber& id, uint64_t offset) {
     checkAttached();
-    getState()->receiver.setCommandPoint(SessionPoint(id, offset));
+    getState()->receiverSetCommandPoint(SessionPoint(id, offset));
     if (!receiveReady) {
         receiveReady = true;
         readyToReceive();
@@ -177,32 +175,37 @@ void SessionHandler::commandPoint(const SequenceNumber& id, uint64_t offset) {
 
 void SessionHandler::expected(const SequenceSet& commands, const Array& /*fragments*/) {
     checkAttached();
-    if (commands.empty() && getState()->hasState())
-        throw IllegalStateException(
+    if (getState()->hasState()) { // Replay
+        if (commands.empty()) throw IllegalStateException(
             QPID_MSG(getState()->getId() << ": has state but client is attaching as new session."));        
-    getState()->sender.expected(commands.empty() ? SequenceNumber(0) : commands.front());
-    if (!sendReady)  // send command point if not already sent
-        sendCommandPoint();
+        // TODO aconway 2008-05-12: support replay of partial commands.
+        // Here we always round down to the last command boundary.
+        SessionPoint expectedPoint = commands.empty() ? SequenceNumber(0) : SessionPoint(commands.front(),0);
+        SessionState::ReplayRange replay = getState()->senderExpected(expectedPoint);
+        sendCommandPoint(expectedPoint);
+        std::for_each(replay.begin(), replay.end(), out); // replay
+    }
+    else
+        sendCommandPoint(getState()->senderGetCommandPoint());
 }
 
 void SessionHandler::confirmed(const SequenceSet& commands, const Array& /*fragments*/) {
     checkAttached();
     // Ignore non-contiguous confirmations.
-    if (!commands.empty() && commands.front() >= getState()->sender.getReplayPoint()) {
-        getState()->sender.confirmed(commands.rangesBegin()->last());
-        }
+    if (!commands.empty() && commands.front() >= getState()->senderGetReplayPoint()) 
+        getState()->senderConfirmed(commands.rangesBegin()->last());
 }
 
 void SessionHandler::completed(const SequenceSet& commands, bool /*timelyReply*/) {
     checkAttached();
-    getState()->sender.completed(commands);
+    getState()->senderCompleted(commands);
     if (!commands.empty())
         peer.knownCompleted(commands);    // Always send a timely reply
 }
 
 void SessionHandler::knownCompleted(const SequenceSet& commands) {
     checkAttached();
-    getState()->receiver.knownCompleted(commands);
+    getState()->receiverKnownCompleted(commands);
 }
 
 void SessionHandler::flush(bool expected, bool confirmed, bool completed) {
@@ -210,18 +213,18 @@ void SessionHandler::flush(bool expected, bool confirmed, bool completed) {
     if (expected)  {
         SequenceSet expectSet;
         if (getState()->hasState())
-            expectSet.add(getState()->receiver.getExpected().command);
+            expectSet.add(getState()->receiverGetExpected().command);
         peer.expected(expectSet, Array());
     }
     if (confirmed) {
         SequenceSet confirmSet;
-        if (!getState()->receiver.getUnknownComplete().empty()) 
-            confirmSet.add(getState()->receiver.getUnknownComplete().front(),
-                           getState()->receiver.getReceived().command);
+        if (!getState()->receiverGetUnknownComplete().empty()) 
+            confirmSet.add(getState()->receiverGetUnknownComplete().front(),
+                           getState()->receiverGetReceived().command);
         peer.confirmed(confirmSet, Array());
     }
     if (completed)
-        peer.completed(getState()->receiver.getUnknownComplete(), true);
+        peer.completed(getState()->receiverGetUnknownComplete(), true);
 }
 
 void SessionHandler::gap(const SequenceSet& /*commands*/) {
@@ -237,20 +240,20 @@ void SessionHandler::sendDetach()
 
 void SessionHandler::sendCompletion() {
     checkAttached();
-    peer.completed(getState()->receiver.getUnknownComplete(), true);
+    peer.completed(getState()->receiverGetUnknownComplete(), true);
 }
 
 void SessionHandler::sendAttach(bool force) {
     checkAttached();
+    QPID_LOG(debug, "SessionHandler::sendAttach attach id=" << getState()->getId());
     peer.attach(getState()->getId().getName(), force);
     if (getState()->hasState())
         peer.flush(true, true, true);
     else
-        sendCommandPoint();
+        sendCommandPoint(getState()->senderGetCommandPoint());
 }
 
-void SessionHandler::sendCommandPoint()  {
-    SessionPoint point(getState()->sender.getCommandPoint());
+void SessionHandler::sendCommandPoint(const SessionPoint& point)  {
     peer.commandPoint(point.command, point.offset);
     if (!sendReady) {
         sendReady = true;
diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.h b/cpp/src/qpid/amqp_0_10/SessionHandler.h
index 85577ebafc..ccbe597bfc 100644
--- a/cpp/src/qpid/amqp_0_10/SessionHandler.h
+++ b/cpp/src/qpid/amqp_0_10/SessionHandler.h
@@ -25,10 +25,10 @@
 #include "qpid/framing/ChannelHandler.h"
 #include "qpid/framing/AMQP_AllProxy.h"
 #include "qpid/framing/AMQP_AllOperations.h"
+#include "qpid/SessionState.h"
 
 namespace qpid {
 
-class SessionState;
 
 namespace amqp_0_10 {
 
@@ -45,8 +45,7 @@ class SessionHandler : public framing::AMQP_AllOperations::SessionHandler,
   public:
     typedef framing::AMQP_AllProxy::Session Peer;
 
-    SessionHandler();
-    SessionHandler(framing::FrameHandler& out, uint16_t channel);
+    SessionHandler(framing::FrameHandler* out=0, uint16_t channel=0);
     ~SessionHandler();
 
     void setChannel(uint16_t ch) { channel = ch; }
@@ -63,7 +62,6 @@ class SessionHandler : public framing::AMQP_AllOperations::SessionHandler,
     void sendAttach(bool force);
     void sendTimeout(uint32_t t);
     void sendFlush();
-    void sendCommandPoint();
 
     /** True if the handler is ready to send and receive */
     bool ready() const;
@@ -96,8 +94,8 @@ class SessionHandler : public framing::AMQP_AllOperations::SessionHandler,
     // Notification of events
     virtual void readyToSend() {}
     virtual void readyToReceive() {}
-    virtual void handleDetach();
     
+    virtual void handleDetach();
     virtual void handleIn(framing::AMQFrame&);
     virtual void handleOut(framing::AMQFrame&);
 
@@ -108,8 +106,9 @@ class SessionHandler : public framing::AMQP_AllOperations::SessionHandler,
     Peer peer;
     bool ignoring;
     bool sendReady, receiveReady;
-    // FIXME aconway 2008-05-07: move handler-related functions from SessionState.
 
+  private:
+    void sendCommandPoint(const SessionPoint&);
 };
 }} // namespace qpid::amqp_0_10