svn commit: r679748 - in /incubator/qpid/branches/qpid.0-10/cpp/src: qpid/client/SubscriptionManager.cpp tests/ClientSessionTest.cpp

[gs...@apache.org]

Author: gsim
Date: Fri Jul 25 03:24:53 2008
New Revision: 679748

URL: http://svn.apache.org/viewvc?rev=679748&view=rev
Log:
Merged fix to SubscriptionManager (was r679739)


Modified:
    incubator/qpid/branches/qpid.0-10/cpp/src/qpid/client/SubscriptionManager.cpp
    incubator/qpid/branches/qpid.0-10/cpp/src/tests/ClientSessionTest.cpp

Modified: incubator/qpid/branches/qpid.0-10/cpp/src/qpid/client/SubscriptionManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-10/cpp/src/qpid/client/SubscriptionManager.cpp?rev=679748&r1=679747&r2=679748&view=diff
==============================================================================
--- incubator/qpid/branches/qpid.0-10/cpp/src/qpid/client/SubscriptionManager.cpp (original)
+++ incubator/qpid/branches/qpid.0-10/cpp/src/qpid/client/SubscriptionManager.cpp Fri Jul 25 03:24:53 2008
@@ -134,8 +134,11 @@
     std::string unique = framing::Uuid(true).str();
     subscribe(lq, queue, FlowControl::messageCredit(1), unique);
     AutoCancel ac(*this, unique);
+    //first wait for message to be delivered if a timeout has been specified
+    if (timeout && lq.get(result, timeout)) return true;
+    //make sure message is not on queue before final check
     sync(session).messageFlush(unique);
-    return lq.get(result, timeout);
+    return lq.get(result, 0);
 }
 
 }} // namespace qpid::client

Modified: incubator/qpid/branches/qpid.0-10/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-10/cpp/src/tests/ClientSessionTest.cpp?rev=679748&r1=679747&r2=679748&view=diff
==============================================================================
--- incubator/qpid/branches/qpid.0-10/cpp/src/tests/ClientSessionTest.cpp (original)
+++ incubator/qpid/branches/qpid.0-10/cpp/src/tests/ClientSessionTest.cpp Fri Jul 25 03:24:53 2008
@@ -41,6 +41,7 @@
 using namespace qpid::framing;
 using namespace qpid;
 using qpid::sys::Monitor;
+using qpid::sys::Thread;
 using qpid::sys::TIME_SEC;
 using std::string;
 using std::cout;
@@ -238,6 +239,19 @@
     BOOST_CHECK_EQUAL("foo2", lq.pop().getData());    
 }
 
+struct DelayedTransfer : sys::Runnable
+{
+    ClientSessionFixture& fixture;
+
+    DelayedTransfer(ClientSessionFixture& f) : fixture(f) {}
+
+    void run()
+    {
+        sleep(1);
+        fixture.session.messageTransfer(content=Message("foo2", "getq"));
+    }
+};
+
 QPID_AUTO_TEST_CASE(testGet) {
     ClientSessionFixture fix;
     fix.session.queueDeclare(queue="getq", exclusive=true, autoDelete=true);
@@ -249,6 +263,12 @@
     BOOST_CHECK(fix.subs.get(got, "getq", TIME_SEC));
     BOOST_CHECK_EQUAL("foo1", got.getData());
     BOOST_CHECK(!fix.subs.get(got, "getq"));
+    DelayedTransfer sender(fix);
+    Thread t(sender);
+    //test timed get where message shows up after a short delay
+    BOOST_CHECK(fix.subs.get(got, "getq", 5*TIME_SEC));
+    BOOST_CHECK_EQUAL("foo2", got.getData());    
+    t.join();
 }
 
 QPID_AUTO_TEST_SUITE_END()