Author: pmoravec
Date: Thu Jul 3 12:56:51 2014
New Revision: 1607628
URL: http://svn.apache.org/r1607628
Log:
[QPID-5866]: [C++ client] AMQP 1.0 closing session without closing receiver
first marks further messages as redelivered (previous commit not complete)
Modified:
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.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=1607628&r1=1607627&r2=1607628&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Thu Jul
3 12:56:51 2014
@@ -140,13 +140,9 @@ void ConnectionContext::endSession(boost
qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
if (pn_session_state(ssn->session) & PN_REMOTE_ACTIVE) {
//explicitly release messages that have yet to be fetched
- for (boost::shared_ptr<ReceiverContext> lnk = ssn->nextReceiver(); lnk
!= boost::shared_ptr<ReceiverContext>(); lnk = ssn->nextReceiver()) {
- for (pn_delivery_t* d = pn_link_current(lnk->receiver); d; d =
pn_link_current(lnk->receiver)) {
- pn_link_advance(lnk->receiver);
- pn_delivery_update(d, PN_RELEASED);
- pn_delivery_settle(d);
- }
- }
+ for (SessionContext::ReceiverMap::iterator i = ssn->receivers.begin();
i != ssn->receivers.end(); ++i) {
+ drain_and_release_messages(ssn, i->second);
+ }
//wait for outstanding sends to settle
while (!ssn->settled()) {
QPID_LOG(debug, "Waiting for sends to settle before closing");
@@ -338,9 +334,8 @@ void ConnectionContext::detach(boost::sh
ssn->removeSender(lnk->getName());
}
-void ConnectionContext::detach(boost::shared_ptr<SessionContext> ssn,
boost::shared_ptr<ReceiverContext> lnk)
+void
ConnectionContext::drain_and_release_messages(boost::shared_ptr<SessionContext>
ssn, boost::shared_ptr<ReceiverContext> lnk)
{
- qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
pn_link_drain(lnk->receiver, 0);
wakeupDriver();
//Not all implementations handle drain correctly, so limit the
@@ -356,6 +351,12 @@ void ConnectionContext::detach(boost::sh
pn_delivery_update(d, PN_RELEASED);
pn_delivery_settle(d);
}
+}
+
+void ConnectionContext::detach(boost::shared_ptr<SessionContext> ssn,
boost::shared_ptr<ReceiverContext> lnk)
+{
+ qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
+ drain_and_release_messages(ssn, lnk);
if (pn_link_state(lnk->receiver) & PN_LOCAL_ACTIVE) {
lnk->close();
}
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h?rev=1607628&r1=1607627&r2=1607628&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.h Thu Jul 3
12:56:51 2014
@@ -78,6 +78,7 @@ class ConnectionContext : public qpid::s
void attach(boost::shared_ptr<SessionContext>,
boost::shared_ptr<ReceiverContext>);
void detach(boost::shared_ptr<SessionContext>,
boost::shared_ptr<SenderContext>);
void detach(boost::shared_ptr<SessionContext>,
boost::shared_ptr<ReceiverContext>);
+ void drain_and_release_messages(boost::shared_ptr<SessionContext>,
boost::shared_ptr<ReceiverContext>);
bool isClosed(boost::shared_ptr<SessionContext>,
boost::shared_ptr<ReceiverContext>);
void send(boost::shared_ptr<SessionContext>,
boost::shared_ptr<SenderContext> ctxt, const qpid::messaging::Message& message,
bool sync);
bool fetch(boost::shared_ptr<SessionContext> ssn,
boost::shared_ptr<ReceiverContext> lnk, qpid::messaging::Message& message,
qpid::messaging::Duration timeout);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
