activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nmitt...@apache.org
Subject svn commit: r519611 - in /activemq/activemq-cpp/trunk/activemq-cpp/src: main/activemq/core/ test-integration/integration/connector/openwire/ test-integration/integration/connector/stomp/
Date Sun, 18 Mar 2007 13:38:34 GMT
Author: nmittler
Date: Sun Mar 18 06:38:33 2007
New Revision: 519611

URL: http://svn.apache.org/viewvc?view=rev&rev=519611
Log:
AMQCPP-81 - fixed distribution of messages already in queue

Modified:
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSessionExecutor.cpp
    activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.cpp
    activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.h
    activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.cpp
    activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.h

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp?view=diff&rev=519611&r1=519610&r2=519611
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQConnection.cpp
Sun Mar 18 06:38:33 2007
@@ -217,10 +217,17 @@
             fire( ex );
 
             return;
-        }
+        }        
 
-        // When not started we drop incomming messages
-        if( !started )
+        // Look up the dispatcher.
+        Dispatcher* dispatcher = NULL;
+        synchronized( &dispatchers )
+        {
+            dispatcher = dispatchers.getValue(consumer->getConsumerId());
+        }
+        
+        // If we have no registered dispatcher, this is bad!! (should never happen)
+        if( dispatcher == NULL )
         {
             // Indicate to Broker that we received the message, but it
             // was not consumed.
@@ -233,21 +240,12 @@
             // Delete the message here
             delete message;
 
-            return;
-        }
-
-        // Look up the dispatcher.
-        Dispatcher* dispatcher = NULL;
-        synchronized( &dispatchers )
-        {
-            dispatcher = dispatchers.getValue(consumer->getConsumerId());
+            throw ActiveMQException(__FILE__, __LINE__, "no dispatcher registered for consumer"
);
         }
 
         // Dispatch the message.
-        if( dispatcher != NULL ) {
-            DispatchData data( consumer, message );
-            dispatcher->dispatch( data );
-        }
+        DispatchData data( consumer, message );
+        dispatcher->dispatch( data );
     }
     catch( exceptions::ActiveMQException& ex )
     {

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSessionExecutor.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSessionExecutor.cpp?view=diff&rev=519611&r1=519610&r2=519611
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSessionExecutor.cpp
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/core/ActiveMQSessionExecutor.cpp
Sun Mar 18 06:38:33 2007
@@ -133,7 +133,9 @@
         consumer = consumers.getValue( data.getConsumer()->getConsumerId() );
     }
     
-    if( consumer != NULL ) {
+    if( consumer == NULL ) {
+        execute( data );
+    } else {
         consumer->dispatch( data );
     }
         

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.cpp?view=diff&rev=519611&r1=519610&r2=519611
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.cpp
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.cpp
Sun Mar 18 06:38:33 2007
@@ -408,3 +408,60 @@
     }
     AMQ_CATCH_RETHROW( ActiveMQException )
 }
+
+void OpenwireSimpleTest::testReceiveAlreadyInQueue() {
+    
+        try
+    {
+        
+        if( IntegrationCommon::debug ) {
+            cout << "Starting activemqcms test (sending "
+                 << IntegrationCommon::defaultMsgCount
+                 << " messages per type and sleeping "
+                 << IntegrationCommon::defaultDelay 
+                 << " milli-seconds) ...\n"
+                 << endl;
+        }
+        
+        // Create CMS Object for Comms
+        cms::ConnectionFactory* factory = new ActiveMQConnectionFactory("tcp://localhost:61616?wireFormat=openwire");
+        cms::Connection* connection = factory->createConnection();
+
+        cms::Session* session = connection->createSession();
+        
+        cms::Topic* topic = session->createTopic(Guid::createGUIDString());
+        
+        cms::MessageConsumer* consumer = session->createConsumer( topic );
+        
+        cms::MessageProducer* producer = session->createProducer( topic );
+
+        cms::TextMessage* textMsg = session->createTextMessage();
+        
+        // Send some text messages
+        producer->send( textMsg );
+        
+        delete textMsg;
+        
+        Thread::sleep( 100 );
+        
+        connection->start();
+
+        cms::Message* message = consumer->receive(1000);
+        CPPUNIT_ASSERT( message != NULL );
+        delete message;
+
+        if( IntegrationCommon::debug ) {
+            printf("Shutting Down\n" );
+        }
+        
+        connection->close();
+        
+        delete producer;                      
+        delete consumer;
+        delete topic;
+        delete session;
+        delete connection;
+        delete factory;
+    }
+    AMQ_CATCH_RETHROW( ActiveMQException )
+}

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.h?view=diff&rev=519611&r1=519610&r2=519611
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.h
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/openwire/OpenwireSimpleTest.h
Sun Mar 18 06:38:33 2007
@@ -36,6 +36,7 @@
         CPPUNIT_TEST( testSyncReceive );
         CPPUNIT_TEST( testMultipleConnections );
         CPPUNIT_TEST( testMultipleSessions );
+        CPPUNIT_TEST( testReceiveAlreadyInQueue );
         CPPUNIT_TEST_SUITE_END();
         
     public:
@@ -49,6 +50,7 @@
         virtual void testSyncReceive();
         virtual void testMultipleConnections();
         virtual void testMultipleSessions();
+        virtual void testReceiveAlreadyInQueue();
     };
 
 }}}

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.cpp?view=diff&rev=519611&r1=519610&r2=519611
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.cpp
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.cpp
Sun Mar 18 06:38:33 2007
@@ -409,3 +409,59 @@
     AMQ_CATCH_RETHROW( ActiveMQException )
 }
 
+void SimpleTest::testReceiveAlreadyInQueue() {
+    
+        try
+    {
+        
+        if( IntegrationCommon::debug ) {
+            cout << "Starting activemqcms test (sending "
+                 << IntegrationCommon::defaultMsgCount
+                 << " messages per type and sleeping "
+                 << IntegrationCommon::defaultDelay 
+                 << " milli-seconds) ...\n"
+                 << endl;
+        }
+        
+        // Create CMS Object for Comms
+        cms::ConnectionFactory* factory = new ActiveMQConnectionFactory("tcp://localhost:61613?wireFormat=stomp");
+        cms::Connection* connection = factory->createConnection();
+
+        cms::Session* session = connection->createSession();
+        
+        cms::Topic* topic = session->createTopic(Guid::createGUIDString());
+        
+        cms::MessageConsumer* consumer = session->createConsumer( topic );
+        
+        cms::MessageProducer* producer = session->createProducer( topic );
+
+        cms::TextMessage* textMsg = session->createTextMessage();
+        
+        // Send some text messages
+        producer->send( textMsg );
+        
+        delete textMsg;
+        
+        Thread::sleep( 100 );
+        
+        connection->start();
+
+        cms::Message* message = consumer->receive(1000);
+        CPPUNIT_ASSERT( message != NULL );
+        delete message;
+
+        if( IntegrationCommon::debug ) {
+            printf("Shutting Down\n" );
+        }
+        
+        connection->close();
+        
+        delete producer;                      
+        delete consumer;
+        delete topic;
+        delete session;
+        delete connection;
+        delete factory;
+    }
+    AMQ_CATCH_RETHROW( ActiveMQException )
+}

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.h?view=diff&rev=519611&r1=519610&r2=519611
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.h
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/connector/stomp/SimpleTest.h
Sun Mar 18 06:38:33 2007
@@ -36,6 +36,7 @@
         CPPUNIT_TEST( testSyncReceive );
         CPPUNIT_TEST( testMultipleConnections );
         CPPUNIT_TEST( testMultipleSessions );
+        CPPUNIT_TEST( testReceiveAlreadyInQueue );
         CPPUNIT_TEST_SUITE_END();
         
     public:
@@ -49,6 +50,7 @@
         virtual void testSyncReceive();
         virtual void testMultipleConnections();
         virtual void testMultipleSessions();
+        virtual void testReceiveAlreadyInQueue();
 
     };
 



Mime
View raw message