You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2010/05/14 22:55:51 UTC
svn commit: r944461 - in /qpid/trunk/qpid/cpp/src:
qpid/client/amqp0_10/IncomingMessages.cpp
qpid/client/amqp0_10/ReceiverImpl.cpp tests/MessagingThreadTests.cpp
Author: aconway
Date: Fri May 14 20:55:51 2010
New Revision: 944461
URL: http://svn.apache.org/viewvc?rev=944461&view=rev
Log:
Fix errors when new API Receiver is closed while in use.
Modified:
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp
qpid/trunk/qpid/cpp/src/tests/MessagingThreadTests.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp?rev=944461&r1=944460&r2=944461&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/IncomingMessages.cpp Fri May 14 20:55:51 2010
@@ -186,22 +186,25 @@ bool IncomingMessages::process(Handler*
{
AbsTime deadline(AbsTime::now(), duration);
FrameSet::shared_ptr content;
- for (Duration timeout = duration; incoming->pop(content, timeout); timeout = Duration(AbsTime::now(), deadline)) {
- if (content->isA<MessageTransferBody>()) {
- MessageTransfer transfer(content, *this);
- if (handler && handler->accept(transfer)) {
- QPID_LOG(debug, "Delivered " << *content->getMethod());
- return true;
+ try {
+ for (Duration timeout = duration; incoming->pop(content, timeout); timeout = Duration(AbsTime::now(), deadline)) {
+ if (content->isA<MessageTransferBody>()) {
+ MessageTransfer transfer(content, *this);
+ if (handler && handler->accept(transfer)) {
+ QPID_LOG(debug, "Delivered " << *content->getMethod());
+ return true;
+ } else {
+ //received message for another destination, keep for later
+ QPID_LOG(debug, "Pushed " << *content->getMethod() << " to received queue");
+ sys::Mutex::ScopedLock l(lock);
+ received.push_back(content);
+ }
} else {
- //received message for another destination, keep for later
- QPID_LOG(debug, "Pushed " << *content->getMethod() << " to received queue");
- sys::Mutex::ScopedLock l(lock);
- received.push_back(content);
+ //TODO: handle other types of commands (e.g. message-accept, message-flow etc)
}
- } else {
- //TODO: handle other types of commands (e.g. message-accept, message-flow etc)
}
}
+ catch (const qpid::ClosedException&) {} // Just return false if queue closed.
return false;
}
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp?rev=944461&r1=944460&r2=944461&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ReceiverImpl.cpp Fri May 14 20:55:51 2010
@@ -112,21 +112,16 @@ void ReceiverImpl::init(qpid::client::As
{
sys::Mutex::ScopedLock l(lock);
session = s;
+ if (state == CANCELLED) return;
if (state == UNRESOLVED) {
source = resolver.resolveSource(session, address);
assert(source.get());
state = STARTED;
}
- if (state == CANCELLED) {
- source->cancel(session, destination);
- parent->receiverCancelled(destination);
- } else {
- source->subscribe(session, destination);
- startFlow(l);
- }
+ source->subscribe(session, destination);
+ startFlow(l);
}
-
const std::string& ReceiverImpl::getName() const {
sys::Mutex::ScopedLock l(lock);
return destination;
@@ -156,6 +151,10 @@ ReceiverImpl::ReceiverImpl(SessionImpl&
bool ReceiverImpl::getImpl(qpid::messaging::Message& message, qpid::messaging::Duration timeout)
{
+ {
+ sys::Mutex::ScopedLock l(lock);
+ if (state == CANCELLED) return false;
+ }
return parent->get(*this, message, timeout);
}
@@ -163,7 +162,7 @@ bool ReceiverImpl::fetchImpl(qpid::messa
{
{
sys::Mutex::ScopedLock l(lock);
- if (state == CANCELLED) return false;//TODO: or should this be an error?
+ if (state == CANCELLED) return false;
if (capacity == 0 || state != STARTED) {
session.messageSetFlowMode(destination, FLOW_MODE_CREDIT);
@@ -174,6 +173,7 @@ bool ReceiverImpl::fetchImpl(qpid::messa
if (getImpl(message, timeout)) {
return true;
} else {
+ if (state == CANCELLED) return false; // Might have been closed during get.
sync(session).messageFlush(destination);
{
sys::Mutex::ScopedLock l(lock);
@@ -198,8 +198,6 @@ bool ReceiverImpl::isClosed() const {
return state == CANCELLED;
}
-
-
void ReceiverImpl::setCapacityImpl(uint32_t c)
{
sys::Mutex::ScopedLock l(lock);
Modified: qpid/trunk/qpid/cpp/src/tests/MessagingThreadTests.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/MessagingThreadTests.cpp?rev=944461&r1=944460&r2=944461&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessagingThreadTests.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessagingThreadTests.cpp Fri May 14 20:55:51 2010
@@ -54,13 +54,14 @@ struct ReceiveThread : public sys::Runna
}
};
+
QPID_AUTO_TEST_CASE(testConcurrentSendReceive) {
- QueueFixture fix;
- Sender s = fix.session.createSender(fix.queue);
- Receiver r = fix.session.createReceiver(fix.queue+";{link:{reliability:unreliable}}");
+ MessagingFixture fix;
+ Sender s = fix.session.createSender("concurrent;{create:always}");
+ Receiver r = fix.session.createReceiver("concurrent;{create:always,link:{reliability:unreliable}}");
ReceiveThread rt(r);
sys::Thread thread(rt);
- const size_t COUNT=1000;
+ const size_t COUNT=100;
for (size_t i = 0; i < COUNT; ++i) {
s.send(Message());
}
@@ -71,28 +72,35 @@ QPID_AUTO_TEST_CASE(testConcurrentSendRe
}
QPID_AUTO_TEST_CASE(testCloseBusyReceiver) {
- QueueFixture fix;
- Receiver r = fix.session.createReceiver(fix.queue);
+ MessagingFixture fix;
+ Receiver r = fix.session.createReceiver("closeReceiver;{create:always}");
ReceiveThread rt(r);
sys::Thread thread(rt);
+ sys::usleep(1000); // Give the receive thread time to block.
r.close();
thread.join();
BOOST_CHECK_EQUAL(rt.error, string());
- // Check that using a closed receiver gives the right result.
+ // Fetching on closed receiver should fail.
Message m;
BOOST_CHECK(!r.fetch(m, Duration(0)));
BOOST_CHECK_THROW(r.fetch(Duration(0)), NoMessageAvailable);
}
QPID_AUTO_TEST_CASE(testCloseSessionBusyReceiver) {
- QueueFixture fix;
- Receiver r = fix.session.createReceiver(fix.queue);
+ MessagingFixture fix;
+ Receiver r = fix.session.createReceiver("closeSession;{create:always}");
ReceiveThread rt(r);
sys::Thread thread(rt);
+ sys::usleep(1000); // Give the receive thread time to block.
fix.session.close();
thread.join();
BOOST_CHECK_EQUAL(rt.error, string());
+
+ // Fetching on closed receiver should fail.
+ Message m;
+ BOOST_CHECK(!r.fetch(m, Duration(0)));
+ BOOST_CHECK_THROW(r.fetch(Duration(0)), NoMessageAvailable);
}
QPID_AUTO_TEST_SUITE_END()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org