qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject svn commit: r1412241 - in /qpid/trunk/qpid/cpp/src/qpid/messaging/amqp: ConnectionContext.cpp SenderContext.cpp SenderContext.h SessionContext.cpp SessionContext.h
Date Wed, 21 Nov 2012 19:31:21 GMT
Author: gsim
Date: Wed Nov 21 19:31:20 2012
New Revision: 1412241

URL: http://svn.apache.org/viewvc?rev=1412241&view=rev
Log:
QPID-4451: wait for outgoing messages to settle when closing

Modified:
    qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
    qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
    qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
    qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
    qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h

Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp?rev=1412241&r1=1412240&r2=1412241&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Wed Nov 21 19:31:20
2012
@@ -149,7 +149,14 @@ void ConnectionContext::close()
     qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
     if (state != CONNECTED) return;
     if (!(pn_connection_state(connection) & PN_LOCAL_CLOSED)) {
-        for (SessionMap::iterator i = sessions.begin(); i != sessions.end(); ++i){
+        for (SessionMap::iterator i = sessions.begin(); i != sessions.end(); ++i) {
+            //wait for outstanding sends to settle
+            while (!i->second->settled()) {
+                QPID_LOG(debug, "Waiting for sends to settle before closing");
+                wait();//wait until message has been confirmed
+            }
+
+
             if (!(pn_session_state(i->second->session) & PN_LOCAL_CLOSED)) {
                 pn_session_close(i->second->session);
             }

Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp?rev=1412241&r1=1412240&r2=1412241&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp Wed Nov 21 19:31:20 2012
@@ -350,4 +350,10 @@ void SenderContext::configure(pn_terminu
         helper.setNodeProperties(target);
     }
 }
+
+bool SenderContext::settled()
+{
+    return processUnsettled() == 0;
+}
+
 }}} // namespace qpid::messaging::amqp

Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h?rev=1412241&r1=1412240&r2=1412241&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h Wed Nov 21 19:31:20 2012
@@ -69,6 +69,7 @@ class SenderContext
     const std::string& getTarget() const;
     Delivery* send(const qpid::messaging::Message& message);
     void configure() const;
+    bool settled();
   private:
     friend class ConnectionContext;
     typedef std::deque<Delivery> Deliveries;

Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp?rev=1412241&r1=1412240&r2=1412241&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp Wed Nov 21 19:31:20 2012
@@ -144,4 +144,13 @@ void SessionContext::acknowledge(const q
     }
 }
 
+bool SessionContext::settled()
+{
+    bool result = true;
+    for (SenderMap::iterator i = senders.begin(); i != senders.end(); ++i) {
+        if (!i->second->settled()) result = false;
+    }
+    return result;
+}
+
 }}} // namespace qpid::messaging::amqp

Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h?rev=1412241&r1=1412240&r2=1412241&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h Wed Nov 21 19:31:20 2012
@@ -59,6 +59,7 @@ class SessionContext
     boost::shared_ptr<ReceiverContext> nextReceiver(qpid::messaging::Duration timeout);
     uint32_t getReceivable();
     uint32_t getUnsettledAcks();
+    bool settled();
   private:
     friend class ConnectionContext;
     typedef std::map<std::string, boost::shared_ptr<SenderContext> > SenderMap;



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message