Cluster code fixed for changes in codebase.

 - Using SessionManager::Observer 
 - Better ais test setup, only need to be member of ais group.
 - Update cluster_client
 - SessionState holds handler chains.
 - Cluster frames include next handler ptr.


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

3 files changed, 22 insertions, 31 deletions

diff --git a/cpp/src/qpid/broker/SessionManager.cpp b/cpp/src/qpid/broker/SessionManager.cpp
index aadb2b9004..571d3365db 100644
--- a/cpp/src/qpid/broker/SessionManager.cpp
+++ b/cpp/src/qpid/broker/SessionManager.cpp
@@ -43,12 +43,13 @@ SessionManager::SessionManager(uint32_t a) : ack(a) {}
 
 SessionManager::~SessionManager() {}
 
+// FIXME aconway 2008-02-01: pass handler*, allow open  unattached.
 std::auto_ptr<SessionState>  SessionManager::open(
     SessionHandler& h, uint32_t timeout_)
 {
     Mutex::ScopedLock l(lock);
     std::auto_ptr<SessionState> session(
-        new SessionState(*this, h, timeout_, ack));
+        new SessionState(this, &h, timeout_, ack));
     active.insert(session->getId());
     for_each(observers.begin(), observers.end(),
              boost::bind(&Observer::opened, _1,boost::ref(*session)));
diff --git a/cpp/src/qpid/broker/SessionState.cpp b/cpp/src/qpid/broker/SessionState.cpp
index a75b32cbb5..1021cca1b1 100644
--- a/cpp/src/qpid/broker/SessionState.cpp
+++ b/cpp/src/qpid/broker/SessionState.cpp
@@ -36,23 +36,17 @@ using qpid::management::ManagementObject;
 using qpid::management::Manageable;
 using qpid::management::Args;
 
-void SessionState::handleIn(AMQFrame& f) { semanticHandler->handle(f); }
-
-void SessionState::handleOut(AMQFrame& f) {
-    assert(handler);
-    handler->out.handle(f);
-}
-
 SessionState::SessionState(
-    SessionManager& f, SessionHandler& h, uint32_t timeout_, uint32_t ack) 
+    SessionManager* f, SessionHandler* h, uint32_t timeout_, uint32_t ack) 
     : framing::SessionState(ack, timeout_ > 0),
-      factory(f), handler(&h), id(true), timeout(timeout_),
-      broker(h.getConnection().broker),
-      version(h.getConnection().getVersion()),
+      factory(f), handler(h), id(true), timeout(timeout_),
+      broker(h->getConnection().broker),
+      version(h->getConnection().getVersion()),
       semanticHandler(new SemanticHandler(*this))
 {
-    // TODO aconway 2007-09-20: SessionManager may add plugin
-    // handlers to the chain.
+    in.next = semanticHandler.get();
+    out.next = &handler->out;
+
     getConnection().outputTasks.addOutputTask(&semanticHandler->getSemanticState());
 
     Manageable* parent = broker.GetVhostObject ();
@@ -66,8 +60,8 @@ SessionState::SessionState(
             mgmtObject = management::Session::shared_ptr
                 (new management::Session (this, parent, id.str ()));
             mgmtObject->set_attached (1);
-            mgmtObject->set_clientRef (h.getConnection().GetManagementObject()->getObjectId());
-            mgmtObject->set_channelId (h.getChannel());
+            mgmtObject->set_clientRef (h->getConnection().GetManagementObject()->getObjectId());
+            mgmtObject->set_channelId (h->getChannel());
             mgmtObject->set_detachedLifespan (getTimeout());
             agent->addObject (mgmtObject);
         }
@@ -76,12 +70,10 @@ SessionState::SessionState(
 
 SessionState::~SessionState() {
     // Remove ID from active session list.
-    factory.erase(getId());
-
+    if (factory)
+        factory->erase(getId());
     if (mgmtObject.get () != 0)
-    {
         mgmtObject->resourceDestroy ();
-    }
 }
 
 SessionHandler* SessionState::getHandler() {
@@ -101,7 +93,7 @@ Connection& SessionState::getConnection() {
 void SessionState::detach() {
     getConnection().outputTasks.removeOutputTask(&semanticHandler->getSemanticState());
     Mutex::ScopedLock l(lock);
-    handler = 0;
+    handler = 0; out.next = 0; 
     if (mgmtObject.get() != 0)
     {
         mgmtObject->set_attached  (0);
@@ -112,6 +104,7 @@ void SessionState::attach(SessionHandler& h) {
     {
         Mutex::ScopedLock l(lock);
         handler = &h;
+        out.next = &handler->out;
         if (mgmtObject.get() != 0)
         {
             mgmtObject->set_attached (1);
diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h
index c8c32a046d..bc1b974eaa 100644
--- a/cpp/src/qpid/broker/SessionState.h
+++ b/cpp/src/qpid/broker/SessionState.h
@@ -58,7 +58,7 @@ class Connection;
  * themselves have state. 
  */
 class SessionState : public framing::SessionState,
-    public framing::FrameHandler::InOutHandler,
+    public framing::FrameHandler::Chains,
     public sys::OutputControl,
     public management::Manageable
 {
@@ -90,18 +90,15 @@ class SessionState : public framing::SessionState,
     management::Manageable::status_t
         ManagementMethod (uint32_t methodId, management::Args& args);
 
-  protected:
-    void handleIn(framing::AMQFrame&);
-    void handleOut(framing::AMQFrame&);
-    
-  private:
-    // SessionManager creates sessions.
-    SessionState(SessionManager&,
-                 SessionHandler& out,
+    // Normally SessionManager creates sessions.
+    SessionState(SessionManager*,
+                 SessionHandler* out,
                  uint32_t timeout,
                  uint32_t ackInterval);
     
-    SessionManager& factory;
+
+  private:
+    SessionManager* factory;
     SessionHandler* handler;    
     framing::Uuid id;
     uint32_t timeout;