activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r711882 - in /activemq/activemq-cpp/trunk/src/test/activemq/transport: IOTransportTest.cpp IOTransportTest.h
Date Thu, 06 Nov 2008 15:40:33 GMT
Author: tabish
Date: Thu Nov  6 07:40:23 2008
New Revision: 711882

URL: http://svn.apache.org/viewvc?rev=711882&view=rev
Log:
https://issues.apache.org/activemq/browse/AMQCPP-200

Add another IOTransport test to stress start/stop/destory for the transport.

Modified:
    activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.cpp
    activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.h

Modified: activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.cpp?rev=711882&r1=711881&r2=711882&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.cpp (original)
+++ activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.cpp Thu Nov  6
07:40:23 2008
@@ -17,11 +17,232 @@
 
 #include "IOTransportTest.h"
 
+#include <activemq/transport/IOTransport.h>
+#include <activemq/transport/CommandListener.h>
+#include <activemq/transport/CommandReader.h>
+#include <activemq/transport/CommandWriter.h>
+#include <activemq/transport/Command.h>
+#include <activemq/transport/TransportExceptionListener.h>
+#include <decaf/util/concurrent/Concurrent.h>
+#include <decaf/util/concurrent/CountDownLatch.h>
+#include <decaf/util/concurrent/Mutex.h>
+#include <decaf/io/BufferedInputStream.h>
+#include <decaf/io/BufferedOutputStream.h>
+#include <decaf/io/BlockingByteArrayInputStream.h>
+#include <decaf/io/ByteArrayOutputStream.h>
+#include <decaf/lang/Thread.h>
+#include <decaf/lang/Exception.h>
+#include <decaf/util/Random.h>
+
 using namespace activemq;
 using namespace activemq::transport;
 using namespace activemq::exceptions;
 
 ////////////////////////////////////////////////////////////////////////////////
+class MyCommand : public Command{
+public:
+    MyCommand(){ c = 0; }
+    virtual ~MyCommand(){}
+
+    char c;
+
+    virtual void setCommandId( int id AMQCPP_UNUSED){}
+    virtual int getCommandId() const{ return 0; }
+
+    virtual void setResponseRequired( const bool required AMQCPP_UNUSED){}
+    virtual bool isResponseRequired() const{ return false; }
+    virtual std::string toString() const{ return ""; }
+    virtual Command* cloneCommand() const{
+        MyCommand* command = new MyCommand;
+        command->c = c;
+        return command;
+    }
+};
+
+////////////////////////////////////////////////////////////////////////////////
+class MyCommandListener : public CommandListener{
+private:
+
+    decaf::util::concurrent::CountDownLatch latch;
+
+public:
+
+    MyCommandListener() : latch(1) {}
+    MyCommandListener( unsigned int num ) : latch( num ) {}
+
+    virtual ~MyCommandListener(){}
+
+    virtual void await() {
+        latch.await();
+    }
+
+    std::string str;
+    virtual void onCommand( Command* command ){
+        const MyCommand* cmd = dynamic_cast<const MyCommand*>(command);
+        str += cmd->c;
+        delete command;
+        latch.countDown();
+    }
+};
+
+////////////////////////////////////////////////////////////////////////////////
+class MyCommandReader : public CommandReader{
+private:
+
+    /**
+     * The target input stream.
+     */
+    decaf::io::InputStream* inputStream;
+
+public:
+
+    MyCommandReader(){ throwException = false; }
+    virtual ~MyCommandReader(){}
+
+    bool throwException;
+
+    virtual void setInputStream(decaf::io::InputStream* is){
+        inputStream = is;
+    }
+
+    virtual decaf::io::InputStream* getInputStream(void){
+        return inputStream;
+    }
+
+    virtual Command* readCommand( void ) throw ( CommandIOException ){
+
+        try{
+            if( throwException ){
+                throw CommandIOException();
+            }
+
+            decaf::util::Random randGen;
+
+            synchronized( inputStream ){
+                MyCommand* command = new MyCommand();
+                try{
+
+                    // Throw a little uncertainty into the test.
+                    unsigned int randWait = randGen.nextInt() % 50;
+                    decaf::lang::Thread::sleep( randWait );
+
+                    command->c = inputStream->read();
+
+                } catch( decaf::lang::Exception& ex ){
+
+                    // Free the memory.
+                    delete command;
+
+                    ex.setMark( __FILE__, __LINE__ );
+                    throw CommandIOException();
+                } catch( ... ) {
+                    // Free the memory.
+                    delete command;
+
+                    throw CommandIOException( __FILE__, __LINE__, "Catch all" );
+                }
+
+                return command;
+            }
+
+            assert(false);
+            return NULL;
+
+        }catch( decaf::lang::Exception& ex ){
+            CommandIOException cx;
+            cx.setMark( __FILE__, __LINE__ );
+            throw cx;
+        }
+        catch( ... ){
+            CommandIOException cx;
+            cx.setMark( __FILE__, __LINE__ );
+            throw cx;
+        }
+    }
+
+    virtual std::size_t read(unsigned char* buffer AMQCPP_UNUSED,
+                     std::size_t count AMQCPP_UNUSED)
+        throw( decaf::io::IOException ) {
+        return 0;
+    }
+
+    virtual unsigned char readByte() throw(decaf::io::IOException) {
+        return 0;
+    }
+};
+
+////////////////////////////////////////////////////////////////////////////////
+class MyCommandWriter : public CommandWriter{
+private:
+
+    /**
+     * Target output stream.
+     */
+    decaf::io::OutputStream* outputStream;
+
+public:
+
+    virtual ~MyCommandWriter(){}
+
+    virtual void setOutputStream(decaf::io::OutputStream* os){
+        outputStream = os;
+    }
+
+    virtual decaf::io::OutputStream* getOutputStream(void){
+        return outputStream;
+    }
+
+    virtual void writeCommand( Command* command )
+        throw (CommandIOException)
+    {
+        try{
+            synchronized( outputStream ){
+
+                const MyCommand* m =
+                    dynamic_cast<const MyCommand*>(command);
+                outputStream->write( m->c );
+            }
+        }catch( decaf::lang::Exception& ex ){
+            ex.setMark( __FILE__, __LINE__ );
+            throw CommandIOException( ex );
+        }
+        catch(...){
+            throw CommandIOException( __FILE__, __LINE__, "writeCommand");
+        }
+    }
+
+    virtual void write( const unsigned char* buffer AMQCPP_UNUSED,
+                        std::size_t count AMQCPP_UNUSED )
+        throw(decaf::io::IOException) {}
+
+    virtual void writeByte( unsigned char v AMQCPP_UNUSED )
+        throw( decaf::io::IOException ) {}
+};
+
+////////////////////////////////////////////////////////////////////////////////
+class MyExceptionListener : public TransportExceptionListener{
+public:
+
+    Transport* transport;
+    decaf::util::concurrent::Mutex mutex;
+
+    MyExceptionListener(){
+        transport = NULL;
+    }
+    virtual ~MyExceptionListener(){}
+
+    virtual void onTransportException( Transport* source,
+                const decaf::lang::Exception& ex AMQCPP_UNUSED){
+        transport = source;
+
+        synchronized( &mutex )
+        {
+           mutex.notify();
+        }
+    }
+};
+
+////////////////////////////////////////////////////////////////////////////////
 // This will just test that we can start and stop the
 // transport without any exceptions.
 void IOTransportTest::testStartClose(){
@@ -46,11 +267,47 @@
 }
 
 ////////////////////////////////////////////////////////////////////////////////
+void IOTransportTest::testStressTransportStartClose(){
+
+    decaf::io::BlockingByteArrayInputStream is;
+    decaf::io::ByteArrayOutputStream os;
+    decaf::io::BufferedInputStream bis( &is );
+    decaf::io::BufferedOutputStream bos( &os );
+
+    for( int i = 0; i < 50; ++i ) {
+        MyCommandListener listener;
+        MyCommandReader reader;
+        MyCommandWriter writer;
+        MyExceptionListener exListener;
+
+        IOTransport transport;
+        transport.setCommandListener( &listener );
+        transport.setCommandReader( &reader );
+        transport.setCommandWriter( &writer );
+        transport.setInputStream( &bis );
+        transport.setOutputStream( &bos );
+        transport.setTransportExceptionListener( &exListener );
+
+        transport.start();
+        unsigned char buffer[10] = { '1', '2', '3', '4', '5', '6', '7', '8', '9', '0' };
+        try{
+            synchronized( &is ){
+                is.setByteArray( buffer, 10 );
+            }
+        }catch( decaf::lang::Exception& ex ){
+            ex.setMark( __FILE__, __LINE__ );
+        }
+        decaf::lang::Thread::yield();
+        transport.close();
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
 void IOTransportTest::testRead(){
 
     decaf::io::BlockingByteArrayInputStream is;
     decaf::io::ByteArrayOutputStream os;
-    MyCommandListener listener;
+    MyCommandListener listener(10);
     MyCommandReader reader;
     MyCommandWriter writer;
     MyExceptionListener exListener;
@@ -75,7 +332,7 @@
         ex.setMark( __FILE__, __LINE__ );
     }
 
-    decaf::lang::Thread::sleep( 100 );
+    listener.await();
 
     CPPUNIT_ASSERT( listener.str == "1234567890" );
 

Modified: activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.h?rev=711882&r1=711881&r2=711882&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.h (original)
+++ activemq/activemq-cpp/trunk/src/test/activemq/transport/IOTransportTest.h Thu Nov  6 07:40:23
2008
@@ -20,19 +20,6 @@
 
 #include <cppunit/TestFixture.h>
 #include <cppunit/extensions/HelperMacros.h>
-
-#include <activemq/transport/IOTransport.h>
-#include <activemq/transport/CommandListener.h>
-#include <activemq/transport/CommandReader.h>
-#include <activemq/transport/CommandWriter.h>
-#include <activemq/transport/Command.h>
-#include <activemq/transport/TransportExceptionListener.h>
-#include <decaf/util/concurrent/Concurrent.h>
-#include <decaf/io/BlockingByteArrayInputStream.h>
-#include <decaf/io/ByteArrayOutputStream.h>
-#include <decaf/lang/Thread.h>
-#include <decaf/lang/Exception.h>
-#include <decaf/util/concurrent/Mutex.h>
 #include <activemq/util/Config.h>
 
 namespace activemq{
@@ -42,6 +29,7 @@
 
         CPPUNIT_TEST_SUITE( IOTransportTest );
         CPPUNIT_TEST( testStartClose );
+        CPPUNIT_TEST( testStressTransportStartClose );
         CPPUNIT_TEST( testRead );
         CPPUNIT_TEST( testWrite );
         CPPUNIT_TEST( testException );
@@ -49,190 +37,13 @@
 
     public:
 
-        class MyCommand : public Command{
-        public:
-            MyCommand(){ c = 0; }
-            virtual ~MyCommand(){}
-
-            char c;
-
-            virtual void setCommandId( int id AMQCPP_UNUSED){}
-            virtual int getCommandId() const{ return 0; }
-
-            virtual void setResponseRequired( const bool required AMQCPP_UNUSED){}
-            virtual bool isResponseRequired() const{ return false; }
-            virtual std::string toString() const{ return ""; }
-            virtual Command* cloneCommand() const{
-                MyCommand* command = new MyCommand;
-                command->c = c;
-                return command;
-            }
-        };
-
-        class MyCommandListener : public CommandListener{
-        public:
-            MyCommandListener(){}
-            virtual ~MyCommandListener(){}
-
-            std::string str;
-            virtual void onCommand( Command* command ){
-                const MyCommand* cmd = dynamic_cast<const MyCommand*>(command);
-                str += cmd->c;
-                delete command;
-            }
-        };
-
-        class MyCommandReader : public CommandReader{
-        private:
-
-            /**
-             * The target input stream.
-             */
-            decaf::io::InputStream* inputStream;
-
-        public:
-            MyCommandReader(){ throwException = false; }
-            virtual ~MyCommandReader(){}
-
-            bool throwException;
-
-            virtual void setInputStream(decaf::io::InputStream* is){
-                inputStream = is;
-            }
-
-            virtual decaf::io::InputStream* getInputStream(void){
-                return inputStream;
-            }
-
-            virtual Command* readCommand( void ) throw ( CommandIOException ){
-
-                try{
-                    if( throwException ){
-                        throw CommandIOException();
-                    }
-
-                    synchronized( inputStream ){
-                        MyCommand* command = new MyCommand();
-                        try{
-                            command->c = inputStream->read();
-                        } catch( decaf::lang::Exception& ex ){
-
-                            // Free the memory.
-                            delete command;
-
-                            ex.setMark( __FILE__, __LINE__ );
-                            throw CommandIOException();
-                        } catch( ... ) {
-                            // Free the memory.
-                            delete command;
-
-                            throw CommandIOException( __FILE__, __LINE__, "Catch all" );
-                        }
-
-                        return command;
-                    }
-
-                    assert(false);
-                    return NULL;
-                }catch( decaf::lang::Exception& ex ){
-                    CommandIOException cx;
-                    cx.setMark( __FILE__, __LINE__ );
-                    throw cx;
-                }
-                catch( ... ){
-                    CommandIOException cx;
-                    cx.setMark( __FILE__, __LINE__ );
-                    throw cx;
-                }
-            }
-
-            virtual std::size_t read(unsigned char* buffer AMQCPP_UNUSED,
-                             std::size_t count AMQCPP_UNUSED)
-                throw( decaf::io::IOException ) {
-                return 0;
-            }
-
-            virtual unsigned char readByte() throw(decaf::io::IOException) {
-                return 0;
-            }
-        };
-
-        class MyCommandWriter : public CommandWriter{
-        private:
-
-            /**
-             * Target output stream.
-             */
-            decaf::io::OutputStream* outputStream;
-
-        public:
-            virtual ~MyCommandWriter(){}
-
-            virtual void setOutputStream(decaf::io::OutputStream* os){
-                outputStream = os;
-            }
-
-            virtual decaf::io::OutputStream* getOutputStream(void){
-                return outputStream;
-            }
-
-            virtual void writeCommand( Command* command )
-                throw (CommandIOException)
-            {
-                try{
-                    synchronized( outputStream ){
-
-                        const MyCommand* m =
-                            dynamic_cast<const MyCommand*>(command);
-                        outputStream->write( m->c );
-                    }
-                }catch( decaf::lang::Exception& ex ){
-                    ex.setMark( __FILE__, __LINE__ );
-                    throw CommandIOException( ex );
-                }
-                catch( ... ){
-                    throw CommandIOException( __FILE__, __LINE__, "writeCommand");
-                }
-            }
-
-            virtual void write( const unsigned char* buffer AMQCPP_UNUSED,
-                                std::size_t count AMQCPP_UNUSED)
-                throw(decaf::io::IOException) {}
-
-            virtual void writeByte(unsigned char v AMQCPP_UNUSED)
-                throw( decaf::io::IOException ) {}
-        };
-
-        class MyExceptionListener : public TransportExceptionListener{
-        public:
-
-            Transport* transport;
-            decaf::util::concurrent::Mutex mutex;
-
-            MyExceptionListener(){
-                transport = NULL;
-            }
-            virtual ~MyExceptionListener(){}
-
-            virtual void onTransportException( Transport* source,
-                        const decaf::lang::Exception& ex AMQCPP_UNUSED){
-                transport = source;
-
-                synchronized(&mutex)
-                {
-                   mutex.notify();
-                }
-            }
-        };
-
-    public:
-
         virtual ~IOTransportTest(){}
 
         void testException();
         void testWrite();
         void testRead();
         void testStartClose();
+        void testStressTransportStartClose();
 
     };
 



Mime
View raw message