Fixed error handling session-busy condition on broker.

Added accessors to iterate over broker::SemanticState consumers.


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

2 files changed, 12 insertions, 0 deletions

diff --git a/cpp/src/qpid/broker/SemanticState.h b/cpp/src/qpid/broker/SemanticState.h
index 94bd929adc..0c56885f8f 100644
--- a/cpp/src/qpid/broker/SemanticState.h
+++ b/cpp/src/qpid/broker/SemanticState.h
@@ -45,6 +45,7 @@
 #include <vector>
 
 #include <boost/intrusive_ptr.hpp>
+#include <boost/cast.hpp>
 
 namespace qpid {
 namespace broker {
@@ -58,6 +59,7 @@ class SessionContext;
 class SemanticState : public sys::OutputTask,
                       private boost::noncopyable
 {
+  public:
     class ConsumerImpl : public Consumer, public sys::OutputTask,
                          public boost::enable_shared_from_this<ConsumerImpl>
     {
@@ -106,8 +108,11 @@ class SemanticState : public sys::OutputTask,
 
         bool hasOutput();
         bool doOutput();
+
+        std::string getName() const { return name; }
     };
 
+  private:
     typedef std::map<std::string, ConsumerImpl::shared_ptr> ConsumerImplMap;
     typedef std::map<std::string, DtxBuffer::shared_ptr> DtxBufferMap;
 
@@ -190,6 +195,11 @@ class SemanticState : public sys::OutputTask,
 
     void attached();
     void detached();
+
+    template <class F> void eachConsumer(const F& f) {
+        outputTasks.eachOutput(
+            boost::bind(f, boost::bind(&boost::polymorphic_downcast<ConsumerImpl*, OutputTask>, _1)));
+    }
 };
 
 }} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h
index 5dd57d2299..bdef894f9f 100644
--- a/cpp/src/qpid/broker/SessionState.h
+++ b/cpp/src/qpid/broker/SessionState.h
@@ -100,6 +100,8 @@ class SessionState : public qpid::SessionState,
 
     void readyToSend();
 
+    template <class F> void eachConsumer(const F& f) { semanticState.eachConsumer(f); }
+    
   private:
 
     void handleCommand(framing::AMQMethodBody* method, const framing::SequenceNumber& id);