Some refactoring of the 0-10 codepath (being migrated to final spec) that primarily colocates the current session and execution layers to facilitate implementing the new session layer that will now encompass this behaviour.

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

1 files changed, 13 insertions, 18 deletions

diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index 7b4035604f..c6c0ae5440 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -19,7 +19,7 @@
  *
  */
 
-#include "SessionState.h"
+#include "SessionContext.h"
 #include "BrokerAdapter.h"
 #include "Queue.h"
 #include "Connection.h"
@@ -56,7 +56,7 @@ using namespace qpid::framing;
 using namespace qpid::sys;
 using namespace qpid::ptr_map;
 
-SemanticState::SemanticState(DeliveryAdapter& da, SessionState& ss)
+SemanticState::SemanticState(DeliveryAdapter& da, SessionContext& ss)
     : session(ss),
       deliveryAdapter(da),
       prefetchSize(0),
@@ -263,21 +263,16 @@ SemanticState::ConsumerImpl::ConsumerImpl(SemanticState* _parent,
 
 bool SemanticState::ConsumerImpl::deliver(QueuedMessage& msg)
 {
-    if (parent->getSession().isAttached() && accept(msg.payload)) {
-        allocateCredit(msg.payload);
-        DeliveryId deliveryTag =
-            parent->deliveryAdapter.deliver(msg, token);
-        if (windowing || ackExpected) {
-            parent->record(DeliveryRecord(msg, queue, name, token, deliveryTag, acquire, !ackExpected));
-        } 
-        if (acquire && !ackExpected) {
-            queue->dequeue(0, msg.payload);
-        }
-        return true;
-    } else {
-        QPID_LOG(debug, "Failed to deliver message to '" << name << "' on " << parent);
-        return false;
+    allocateCredit(msg.payload);
+    DeliveryId deliveryTag =
+        parent->deliveryAdapter.deliver(msg, token);
+    if (windowing || ackExpected) {
+        parent->record(DeliveryRecord(msg, queue, name, token, deliveryTag, acquire, !ackExpected));
+    } 
+    if (acquire && !ackExpected) {
+        queue->dequeue(0, msg.payload);
     }
+    return true;
 }
 
 bool SemanticState::ConsumerImpl::filter(intrusive_ptr<Message> msg)
@@ -331,7 +326,7 @@ void SemanticState::cancel(ConsumerImpl& c)
     if(queue) {
         queue->cancel(c);
         if (queue->canAutoDelete() && !queue->hasExclusiveOwner()) {            
-            Queue::tryAutoDelete(getSession().getBroker(), queue);
+            Queue::tryAutoDelete(getSession().getConnection().getBroker(), queue);
         }
     }
 }
@@ -584,7 +579,7 @@ Queue::shared_ptr SemanticState::getQueue(const string& name) const {
     if (name.empty()) {
         throw NotAllowedException(QPID_MSG("No queue name specified."));
     } else {
-        queue = session.getBroker().getQueues().find(name);
+        queue = session.getConnection().getBroker().getQueues().find(name);
         if (!queue)
             throw NotFoundException(QPID_MSG("Queue not found: "<<name));
     }