svn commit: r481559 - in /incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration: ./ ExpirationTest.cpp ExpirationTest.h

[nm...@apache.org]

Author: nmittler
Date: Sat Dec  2 05:39:33 2006
New Revision: 481559

URL: http://svn.apache.org/viewvc?view=rev&rev=481559
Log:
[AMQCPP-14] Adding integration test for expiration

Added:
    incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/
    incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.cpp
    incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.h

Added: incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.cpp
URL: http://svn.apache.org/viewvc/incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.cpp?view=auto&rev=481559
==============================================================================
--- incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.cpp (added)
+++ incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.cpp Sat Dec  2 05:39:33 2006
@@ -0,0 +1,321 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "ExpirationTest.h"
+
+#include <integration/common/IntegrationCommon.h>
+
+//CPPUNIT_TEST_SUITE_REGISTRATION( integration::expiration::ExpirationTest );
+
+#include <sstream>
+
+#include <activemq/core/ActiveMQConnectionFactory.h>
+#include <activemq/exceptions/ActiveMQException.h>
+#include <activemq/concurrent/Thread.h>
+#include <activemq/connector/stomp/StompConnector.h>
+#include <activemq/util/SimpleProperties.h>
+#include <activemq/transport/TransportFactory.h>
+#include <activemq/util/Guid.h>
+#include <activemq/util/SimpleProperties.h>
+#include <activemq/util/StringTokenizer.h>
+#include <activemq/connector/ConnectorFactoryMap.h>
+#include <activemq/network/SocketFactory.h>
+#include <activemq/transport/TransportFactory.h>
+#include <activemq/network/Socket.h>
+#include <activemq/exceptions/NullPointerException.h>
+#include <activemq/core/ActiveMQConnection.h>
+#include <activemq/core/ActiveMQConsumer.h>
+#include <activemq/core/ActiveMQProducer.h>
+#include <activemq/util/StringTokenizer.h>
+#include <activemq/util/Boolean.h>
+
+#include <cms/Connection.h>
+#include <cms/MessageConsumer.h>
+#include <cms/MessageProducer.h>
+#include <cms/MessageListener.h>
+#include <cms/Startable.h>
+#include <cms/Closeable.h>
+#include <cms/MessageListener.h>
+#include <cms/ExceptionListener.h>
+#include <cms/Topic.h>
+#include <cms/Queue.h>
+#include <cms/TemporaryTopic.h>
+#include <cms/TemporaryQueue.h>
+#include <cms/Session.h>
+#include <cms/BytesMessage.h>
+#include <cms/TextMessage.h>
+#include <cms/MapMessage.h>
+#include <cms/Session.h>
+
+using namespace activemq::connector::stomp;
+using namespace activemq::transport;
+using namespace activemq::util;
+using namespace std;
+using namespace cms;
+using namespace activemq;
+using namespace activemq::core;
+using namespace activemq::util;
+using namespace activemq::connector;
+using namespace activemq::exceptions;
+using namespace activemq::network;
+using namespace activemq::transport;
+using namespace activemq::concurrent;
+
+using namespace std;
+using namespace integration;
+using namespace integration::expiration;
+using namespace integration::common;
+
+class Producer : public Runnable {
+private:
+
+    Connection* connection;
+    Session* session;
+    Topic* destination;
+    MessageProducer* producer;
+    int numMessages;
+    long long timeToLive;
+
+public:
+
+    Producer( int numMessages, long long timeToLive ){
+        connection = NULL;
+        session = NULL;
+        destination = NULL;
+        producer = NULL;
+        this->numMessages = numMessages;
+        this->timeToLive = timeToLive;
+    }
+
+    virtual ~Producer(){
+        cleanup();
+    }
+
+    virtual void run() {
+        try {
+            // Create a ConnectionFactory
+            ActiveMQConnectionFactory* connectionFactory = new
+                ActiveMQConnectionFactory("tcp://localhost:61613");
+
+            // Create a Connection
+            connection = connectionFactory->createConnection();
+            connection->start();
+
+            string sss=connection->getClientId();
+            cout << sss << endl;
+
+            session = connection->createSession( Session::AUTO_ACKNOWLEDGE);
+            destination = session->createTopic( "mytopic" );
+
+            producer = session->createProducer( destination );
+            producer->setDeliveryMode( DeliveryMode::PERSISTANT );
+
+            //unsigned long ttt=getcurt();
+            producer->setTimeToLive( 1);
+
+            // Create the Thread Id String
+            string threadIdStr = Integer::toString( Thread::getId() );
+
+            // Create a messages
+            string text = (string)"Hello world! from thread " + threadIdStr;
+
+            for( int ix=0; ix<numMessages; ++ix ){
+                TextMessage* message = session->createTextMessage( text );
+                producer->send( message );
+                delete message;
+           }
+
+       }catch ( CMSException& e ) {
+           e.printStackTrace();
+       }
+   }
+
+private:
+
+    void cleanup(){
+
+        // Destroy resources.
+        try{
+            if( destination != NULL ) delete destination;
+        }catch ( CMSException& e ) {}
+        destination = NULL;
+
+        try{
+            if( producer != NULL ) delete producer;
+        }catch ( CMSException& e ) {}
+        producer = NULL;
+
+        // Close open resources.
+        try{
+            if( session != NULL ) session->close();
+            if( connection != NULL ) connection->close();
+        }catch ( CMSException& e ) {}
+
+        try{
+            if( session != NULL ) delete session;
+        }catch ( CMSException& e ) {}
+        session = NULL;
+
+        try{
+            if( connection != NULL ) delete connection;
+        }catch ( CMSException& e ) {}
+        connection = NULL;
+    }
+};
+
+class Consumer : public MessageListener, public Runnable {
+
+private:
+
+    Connection* connection;
+    Session* session;
+    Topic* destination;
+    MessageConsumer* consumer;
+    long waitMillis;
+    int numReceived;
+
+public:
+
+    Consumer( long waitMillis ){
+        connection = NULL;
+        session = NULL;
+        destination = NULL;
+        consumer = NULL;
+        this->waitMillis = waitMillis;
+        numReceived = 0;
+    }
+    
+    virtual ~Consumer(){
+        cleanup();
+    }
+
+    virtual int getNumReceived() const{
+        return numReceived;
+    }
+    
+    virtual void run() {
+
+        try {
+
+            string user,passwd,sID;
+            user="default";
+            passwd="";
+            sID="lsgID";
+            
+            // Create a ConnectionFactory
+            ActiveMQConnectionFactory* connectionFactory =
+               new ActiveMQConnectionFactory("tcp://localhost:61613",user,passwd,sID);
+
+            // Create a Connection
+            connection = connectionFactory->createConnection();
+            delete connectionFactory;
+            connection->start();
+
+            // Create a Session
+            session = connection->createSession( Session::AUTO_ACKNOWLEDGE);
+
+            // Create the destination (Topic or Queue)
+            destination = session->createTopic( "mytopic?consumer.retroactive=true");
+
+            consumer = session->createDurableConsumer( destination , user ,"",false);
+
+            consumer->setMessageListener( this );
+
+            // Sleep while asynchronous messages come in.
+            Thread::sleep( waitMillis );
+
+        } catch (CMSException& e) {
+            e.printStackTrace();
+        }
+    }
+
+    virtual void onMessage( const Message* message ){
+
+        try
+        {
+            const TextMessage* textMessage =
+                dynamic_cast< const TextMessage* >( message );
+            string text = textMessage->getText();
+            numReceived++;
+        } catch (CMSException& e) {
+            e.printStackTrace();
+        }
+    }
+
+private:
+
+    void cleanup(){
+
+        // Destroy resources.
+        try{
+            if( destination != NULL ) delete destination;
+        }catch (CMSException& e) {}
+        destination = NULL;
+
+        try{
+            if( consumer != NULL ) delete consumer;
+        }catch (CMSException& e) {}
+        consumer = NULL;
+
+        // Close open resources.
+        try{
+            if( session != NULL ) session->close();
+            if( connection != NULL ) connection->close();
+        }catch (CMSException& e) {}
+
+        try{
+            if( session != NULL ) delete session;
+        }catch (CMSException& e) {}
+        session = NULL;
+
+        try{
+            if( connection != NULL ) delete connection;
+        }catch (CMSException& e) {}
+        connection = NULL;
+    }
+};
+
+void ExpirationTest::testExpired()
+{
+    Producer producer( 2, 1 );
+    Thread producerThread( &producer );
+    producerThread.start();
+    producerThread.join();
+    
+    Consumer consumer( 2000 );
+    Thread consumerThread( &consumer );
+    consumerThread.start();
+    consumerThread.join();
+    
+    CPPUNIT_ASSERT( consumer.getNumReceived() == 0 );
+}
+
+void ExpirationTest::testNotExpired()
+{
+    Producer producer( 2, 2000 );
+    Thread producerThread( &producer );
+    producerThread.start();
+    producerThread.join();
+    
+    Consumer consumer( 2000 );
+    Thread consumerThread( &consumer );
+    consumerThread.start();
+    consumerThread.join();
+    
+    CPPUNIT_ASSERT( consumer.getNumReceived() == 2 );
+}
+

Added: incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.h
URL: http://svn.apache.org/viewvc/incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.h?view=auto&rev=481559
==============================================================================
--- incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.h (added)
+++ incubator/activemq/activemq-cpp/trunk/activemq-cpp/src/test-integration/integration/expiration/ExpirationTest.h Sat Dec  2 05:39:33 2006
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef _INTEGRATION_EXPIRATION_EXPIRATIONTEST_H_
+#define _INTEGRATION_EXPIRATION_EXPIRATIONTEST_H_
+
+#include <cppunit/TestFixture.h>
+#include <cppunit/extensions/HelperMacros.h>
+
+#include <activemq/concurrent/Mutex.h>
+
+#include <cms/MessageListener.h>
+#include <cms/ExceptionListener.h>
+#include <cms/ConnectionFactory.h>
+#include <cms/Connection.h>
+#include <cms/Session.h>
+#include <cms/MessageProducer.h>
+
+namespace integration{
+namespace expiration{
+
+    class ExpirationTest : public CppUnit::TestFixture   
+    {
+        CPPUNIT_TEST_SUITE( ExpirationTest );
+        //CPPUNIT_TEST( testExpired );
+        //CPPUNIT_TEST( testNotExpired );
+        CPPUNIT_TEST_SUITE_END();
+
+    public:
+    
+        ExpirationTest(){}
+        virtual ~ExpirationTest(){}
+        
+        virtual void testExpired();
+        virtual void testNotExpired();
+    };
+
+}}
+
+#endif /*_INTEGRATION_EXPIRATION_EXPIRATIONTEST_H_*/