qpid::SessionState: Added error checking for invalid frame sequences.

client: Fix client crash on error during connection shutdown.


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

5 files changed, 29 insertions, 14 deletions

diff --git a/cpp/src/qpid/client/Connection.cpp b/cpp/src/qpid/client/Connection.cpp
index c11f155afb..4089ad79ce 100644
--- a/cpp/src/qpid/client/Connection.cpp
+++ b/cpp/src/qpid/client/Connection.cpp
@@ -68,7 +68,8 @@ void Connection::open(const ConnectionSettings& settings)
     if (impl)
         throw Exception(QPID_MSG("Connection::open() was already called"));
 
-    impl = boost::shared_ptr<ConnectionImpl>(new ConnectionImpl(version, settings));
+    impl = shared_ptr<ConnectionImpl>(new ConnectionImpl(version, settings));
+    impl->open(settings.host, settings.port);
     max_frame_size = impl->getNegotiatedSettings().maxFrameSize;
 }
 
diff --git a/cpp/src/qpid/client/ConnectionImpl.cpp b/cpp/src/qpid/client/ConnectionImpl.cpp
index 67ae2293f1..bdafa795c2 100644
--- a/cpp/src/qpid/client/ConnectionImpl.cpp
+++ b/cpp/src/qpid/client/ConnectionImpl.cpp
@@ -52,7 +52,6 @@ ConnectionImpl::ConnectionImpl(framing::ProtocolVersion v, const ConnectionSetti
     connector.setTimeoutHandler(this);
     connector.setShutdownHandler(this);
 
-    open(settings.host, settings.port);
     //only set error handler once  open
     handler.onError = boost::bind(&ConnectionImpl::closed, this, _1, _2);
 }
@@ -135,11 +134,13 @@ ConnectionImpl::SessionVector ConnectionImpl::closeInternal(const Mutex::ScopedL
 }
 
 void ConnectionImpl::closed(uint16_t code, const std::string& text) 
-{
-    Mutex::ScopedLock l(lock);
-    if (isClosed) return;
-    SessionVector save(closeInternal(l));
-    Mutex::ScopedUnlock u(lock);
+{ 
+    SessionVector save;
+    {
+        Mutex::ScopedLock l(lock);
+        if (isClosed) return;
+        save = closeInternal(l);
+    }
     std::for_each(save.begin(), save.end(), boost::bind(&SessionImpl::connectionClosed, _1, code, text));
 }
 
diff --git a/cpp/src/qpid/client/ConnectionImpl.h b/cpp/src/qpid/client/ConnectionImpl.h
index 986b044b49..ac28a0f695 100644
--- a/cpp/src/qpid/client/ConnectionImpl.h
+++ b/cpp/src/qpid/client/ConnectionImpl.h
@@ -33,6 +33,7 @@
 #include <map>
 #include <boost/shared_ptr.hpp>
 #include <boost/weak_ptr.hpp>
+#include <boost/enable_shared_from_this.hpp>
 
 namespace qpid {
 namespace client {
@@ -43,7 +44,8 @@ class SessionImpl;
 class ConnectionImpl : public Bounds,
                        public framing::FrameHandler,
                        public sys::TimeoutHandler, 
-                       public sys::ShutdownHandler
+                       public sys::ShutdownHandler,
+                       public boost::enable_shared_from_this<ConnectionImpl>
 
 {
     typedef std::map<uint16_t, boost::weak_ptr<SessionImpl> > SessionMap;
@@ -59,8 +61,6 @@ class ConnectionImpl : public Bounds,
 
     template <class F> void detachAll(const F&);
 
-    void open(const std::string& host, int port);
-
     SessionVector closeInternal(const sys::Mutex::ScopedLock&);
     void incoming(framing::AMQFrame& frame);    
     void closed(uint16_t, const std::string&);
@@ -73,6 +73,8 @@ class ConnectionImpl : public Bounds,
     ConnectionImpl(framing::ProtocolVersion version, const ConnectionSettings& settings);
     ~ConnectionImpl();
     
+    void open(const std::string& host, int port);
+
     void addSession(const boost::shared_ptr<SessionImpl>&);
         
     void close();
diff --git a/cpp/src/qpid/client/Connector.cpp b/cpp/src/qpid/client/Connector.cpp
index c9c55c50e8..793809fc7c 100644
--- a/cpp/src/qpid/client/Connector.cpp
+++ b/cpp/src/qpid/client/Connector.cpp
@@ -21,6 +21,7 @@
 #include "Connector.h"
 
 #include "Bounds.h"
+#include "ConnectionImpl.h"
 #include "ConnectionSettings.h"
 #include "qpid/log/Statement.h"
 #include "qpid/sys/Time.h"
@@ -42,7 +43,9 @@ using namespace qpid::framing;
 using boost::format;
 using boost::str;
 
-Connector::Connector(ProtocolVersion ver, const ConnectionSettings& settings, Bounds* bounds)
+Connector::Connector(ProtocolVersion ver,
+                     const ConnectionSettings& settings,
+                     ConnectionImpl* cimpl)
     : maxFrameSize(settings.maxFrameSize),
       version(ver), 
       initiated(false),
@@ -52,8 +55,9 @@ Connector::Connector(ProtocolVersion ver, const ConnectionSettings& settings, Bo
       idleIn(0), idleOut(0), 
       timeoutHandler(0),
       shutdownHandler(0),
-      writer(maxFrameSize, bounds),
-      aio(0)
+      writer(maxFrameSize, cimpl),
+      aio(0),
+      impl(cimpl)
 {
     QPID_LOG(debug, "Connector created for " << version);
     socket.configure(settings);
@@ -294,6 +298,9 @@ void Connector::eof(AsynchIO&) {
 // TODO: astitcher 20070908 This version of the code can never time out, so the idle processing
 // will never be called
 void Connector::run(){
+    // Keep the connection impl in memory until run() completes.
+    boost::shared_ptr<ConnectionImpl> protect = impl->shared_from_this();
+    assert(protect);
     try {
         Dispatcher d(poller);
 	
diff --git a/cpp/src/qpid/client/Connector.h b/cpp/src/qpid/client/Connector.h
index 9b13e1d519..ce2b23a32e 100644
--- a/cpp/src/qpid/client/Connector.h
+++ b/cpp/src/qpid/client/Connector.h
@@ -37,6 +37,7 @@
 #include "qpid/sys/AsynchIO.h"
 
 #include <queue>
+#include <boost/weak_ptr.hpp>
 #include <boost/shared_ptr.hpp>
 
 namespace qpid {
@@ -45,6 +46,7 @@ namespace client {
 
 class Bounds;
 class ConnectionSettings;
+class ConnectionImpl;
 
 class Connector : public framing::OutputHandler, 
                   private sys::Runnable
@@ -121,13 +123,15 @@ class Connector : public framing::OutputHandler,
     void eof(qpid::sys::AsynchIO&);
 
     std::string identifier;
+
+    ConnectionImpl* impl;
     
   friend class Channel;
 
   public:
     Connector(framing::ProtocolVersion pVersion,
               const ConnectionSettings&, 
-              Bounds* bounds = 0);
+              ConnectionImpl*);
     virtual ~Connector();
     virtual void connect(const std::string& host, int port);
     virtual void init();