QPID-1413: disable output before sending detach

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

6 files changed, 17 insertions, 3 deletions

diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
index 02d72ffd11..8f48e72c2d 100644
--- a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
+++ b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
@@ -88,6 +88,7 @@ void SessionHandler::handleIn(AMQFrame& f) {
         SequenceNumber commandId;
         if (getState()) commandId =  getState()->receiverGetCurrent();
         execution.exception(e.code, commandId, m ? m->amqpClassId() : 0, m ? m->amqpMethodId() : 0, 0, e.what(), FieldTable());
+        detaching();
         sendDetach();
     }
     catch(const ChannelException& e){
diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.h b/cpp/src/qpid/amqp_0_10/SessionHandler.h
index 53ce12b47f..7628cc88ae 100644
--- a/cpp/src/qpid/amqp_0_10/SessionHandler.h
+++ b/cpp/src/qpid/amqp_0_10/SessionHandler.h
@@ -87,7 +87,7 @@ class SessionHandler : public framing::AMQP_AllOperations::SessionHandler,
     virtual void setState(const std::string& sessionName, bool force) = 0;
     virtual void channelException(uint16_t code, const std::string& msg) = 0;
     virtual void connectionException(uint16_t code, const std::string& msg) = 0;
-
+    virtual void detaching() = 0;
 
     // Notification of events
     virtual void readyToSend() {}
diff --git a/cpp/src/qpid/broker/SessionHandler.cpp b/cpp/src/qpid/broker/SessionHandler.cpp
index c752f6315b..0f67c1e0c5 100644
--- a/cpp/src/qpid/broker/SessionHandler.cpp
+++ b/cpp/src/qpid/broker/SessionHandler.cpp
@@ -71,6 +71,12 @@ void SessionHandler::setState(const std::string& name, bool force) {
     session = connection.broker.getSessionManager().attach(*this, id, force);
 }
 
+void SessionHandler::detaching() 
+{
+    assert(session.get());
+    session->disableOutput();
+}
+
 FrameHandler* SessionHandler::getInHandler() { return session.get() ? &session->in : 0; }
 qpid::SessionState* SessionHandler::getState() { return session.get(); }
 
diff --git a/cpp/src/qpid/broker/SessionHandler.h b/cpp/src/qpid/broker/SessionHandler.h
index 1aa3137fdf..a8f741bc1b 100644
--- a/cpp/src/qpid/broker/SessionHandler.h
+++ b/cpp/src/qpid/broker/SessionHandler.h
@@ -65,6 +65,7 @@ class SessionHandler : public amqp_0_10::SessionHandler {
     virtual framing::FrameHandler* getInHandler();
     virtual void channelException(uint16_t code, const std::string& msg);
     virtual void connectionException(uint16_t code, const std::string& msg);
+    virtual void detaching();
     virtual void readyToSend();
 
   private:
diff --git a/cpp/src/qpid/broker/SessionState.cpp b/cpp/src/qpid/broker/SessionState.cpp
index 7910ef3f0c..fe10345499 100644
--- a/cpp/src/qpid/broker/SessionState.cpp
+++ b/cpp/src/qpid/broker/SessionState.cpp
@@ -93,13 +93,18 @@ bool SessionState::isLocal(const ConnectionToken* t) const
 
 void SessionState::detach() {
     QPID_LOG(debug, getId() << ": detached on broker.");
-    semanticState.detached();//prevents further activateOutput calls until reattached
-    getConnection().outputTasks.removeOutputTask(&semanticState);
+    disableOutput();
     handler = 0;
     if (mgmtObject != 0)
         mgmtObject->set_attached  (0);
 }
 
+void SessionState::disableOutput() 
+{
+    semanticState.detached();//prevents further activateOutput calls until reattached
+    getConnection().outputTasks.removeOutputTask(&semanticState);
+}
+
 void SessionState::attach(SessionHandler& h) {
     QPID_LOG(debug, getId() << ": attached on broker.");
     handler = &h;
diff --git a/cpp/src/qpid/broker/SessionState.h b/cpp/src/qpid/broker/SessionState.h
index 97807425dd..f3b85bbd05 100644
--- a/cpp/src/qpid/broker/SessionState.h
+++ b/cpp/src/qpid/broker/SessionState.h
@@ -73,6 +73,7 @@ class SessionState : public qpid::SessionState,
 
     void detach();
     void attach(SessionHandler& handler);
+    void disableOutput();
 
     /** @pre isAttached() */
     framing::AMQP_ClientProxy& getProxy();