activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Timothy Bish <tabish...@gmail.com>
Subject Re: AMQCPP: unexpected exception
Date Tue, 01 Mar 2011 23:31:16 GMT
On Tue, 2011-03-01 at 20:22 -0300, Henrique Magarotto wrote:
> Hi everybody,
> 
> First, sorry for long mail and thanks for great job in ActiveMQ and
> ActiveMQ-CPP.
> 
> I'm testing some failure cases with ActiveMQ and ActiveMQ-CPP.
> 
> In my test, an unexpected exception is thrown if broker is stopped
> while consumer is in transaction.
> 
> Environment:
> ActiveMQ 5.4.2
> ActiveMQ-CPP 3.2.4
> Ubuntu 10.04
> JRE 1.6.0_20-b02
> 

> I get the following message:
> 
> terminate called after throwing an instance of 'cms::CMSException'
>   what():  Failover timeout of 1000 ms reached.
> Aborted (core dumped)
> 

Sounds like a bug, can you open a new issue in are Jira bug tracker and
attach your test case along with instructions for reproducing the error?

https://issues.apache.org/jira/browse/AMQCPP

Regards
Tim.

> core backtrace:
> #0  0x001ef422 in __kernel_vsyscall ()
> #1  0x00c5a651 in *__GI_raise (sig=6) at
> ../nptl/sysdeps/unix/sysv/linux/raise.c:64
> #2  0x00c5da82 in *__GI_abort () at abort.c:92
> #3  0x00bf952f in __gnu_cxx::__verbose_terminate_handler() () from
> /usr/lib/libstdc++.so.6
> #4  0x00bf7465 in ?? () from /usr/lib/libstdc++.so.6
> #5  0x00bf74a2 in std::terminate() () from /usr/lib/libstdc++.so.6
> #6  0x00bf74c5 in ?? () from /usr/lib/libstdc++.so.6
> #7  0x00bf6915 in __cxa_call_unexpected () from /usr/lib/libstdc++.so.6
> #8  0x0052f8ae in
> activemq::core::TransactionSynhcronization::beforeEnd (this=0x93ac548)
> at activemq/core/ActiveMQConsumer.cpp:84
> #9  0x00550588 in
> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x93b3e48)
> at activemq/core/ActiveMQTransactionContext.cpp:192
> #10 0x00550d63 in activemq::core::ActiveMQTransactionContext::commit
> (this=0x93b3e48) at activemq/core/ActiveMQTransactionContext.cpp:127
> #11 0x0053e460 in activemq::core::ActiveMQSession::commit
> (this=0x93b3c10) at activemq/core/ActiveMQSession.cpp:189
> #12 0x0042c1a0 in activemq::cmsutil::PooledSession::commit
> (this=0x93b3fd8) at activemq/cmsutil/PooledSession.h:87
> #13 0x0804c0fb in Consumer::onMessage (this=0x93a8c30,
> message=0x93b49b0) at main.cpp:455
> #14 0x0804bdc8 in Consumer::consumeLoop (this=0x93a8c30) at main.cpp:415
> #15 0x0804bb56 in Consumer::run (this=0x93a8c30) at main.cpp:381
> #16 0x00823cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x93abff0) at decaf/lang/Thread.cpp:135
> #17 0x00822847 in threadWorker (arg=0x93abff0) at decaf/lang/Thread.cpp:188
> #18 0x0015a96e in start_thread (arg=0xb6f8bb70) at pthread_create.c:300
> #19 0x00cfda4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> 
> Exception specification for method
> 'activemq::core::TransactionSynhcronization::beforeEnd' says: 'throw(
> exceptions::ActiveMQException )' but 'ActiveMQConsumer::acknowledge'
> throws 'cms::CMSException', so 'std::unexpected' is called and 'abort'
> occurs. (http://www.linuxprogrammingblog.com/cpp-exception-specifications-are-evil)
> 
> 
> When 'MessageListener' is used, the commit freezes, probability, in thread 5:
> 
> Thread 1:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00ddcb5d in pthread_join (threadid=3061869424,
> thread_return=0xbf964bdc) at pthread_join.c:89
> #2  0x00a923c0 in decaf::lang::Thread::join (this=0x8eba574) at
> decaf/lang/Thread.cpp:421
> #3  0x0804c9f1 in AppTest::joinEndPoins (this=0xbf964c64) at main.cpp:586
> #4  0x0804c678 in AppTest::run (this=0xbf964c64) at main.cpp:555
> #5  0x0804a33c in main (argc=4, argv=0xbf964d84) at main.cpp:633
> 
> Thread 2:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eb0e38,
> mutex=0x8eb0e08) at forward.c:139
> #3  0x00a606a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x8eb0e38) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
> (this=0x8eb0d6c) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x007fc9c5 in activemq::threads::CompositeTaskRunner::run
> (this=0x8eb0d48) at activemq/threads/CompositeTaskRunner.cpp:118
> #6  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8eb0e90) at decaf/lang/Thread.cpp:135
> #7  0x00a92847 in threadWorker (arg=0x8eb0e90) at decaf/lang/Thread.cpp:188
> #8  0x00ddb96e in start_thread (arg=0xb7808b70) at pthread_create.c:300
> #9  0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> Thread 3:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de2af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8ebb3a8) at
> pthread_mutex_lock.c:61
> #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8ebb3a8) at forward.c:182
> #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x8ebb3a8) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
> (this=0x8ebb2bc) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x007c5770 in
> decaf::util::StlQueue<decaf::lang::Pointer<activemq::commands::MessageDispatch,
> decaf::util::concurrent::atomic::AtomicRefCounter> >::lock
> (this=0x8ebb2a8)
>     at ./decaf/util/StlQueue.h:253
> #8  activemq::core::MessageDispatchChannel::lock (this=0x8ebb2a8) at
> activemq/core/MessageDispatchChannel.h:153
> #9  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb7006fe4) at decaf/util/concurrent/Lock.cpp:54
> #10 0x00ad0c08 in Lock (this=0xfffffe00, object=0x8ebb3a8,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #11 0x0078e102 in
> activemq::core::ActiveMQConsumer::clearMessagesInProgress
> (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:1112
> #12 0x007af15c in
> activemq::core::ActiveMQSession::clearMessagesInProgress
> (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:239
> #13 0x00778543 in
> activemq::core::ActiveMQConnection::transportInterrupted
> (this=0x8eb1210) at activemq/core/ActiveMQConnection.cpp:704
> #14 0x00803b34 in
> activemq::transport::TransportFilter::transportInterrupted
> (this=0x8eb11c0) at activemq/transport/TransportFilter.cpp:67
> #15 0x008187ae in
> activemq::transport::failover::FailoverTransport::handleTransportFailure
> (this=0x8eb0a38, error=...) at
> activemq/transport/failover/FailoverTransport.cpp:476
> #16 0x0082406c in
> activemq::transport::failover::FailoverTransportListener::onException
> (this=0x8eb0878, ex=...) at
> activemq/transport/failover/FailoverTransportListener.cpp:97
> #17 0x00803c0b in activemq::transport::TransportFilter::fire
> (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:49
> #18 0x00803c64 in activemq::transport::TransportFilter::onException
> (this=0x8eb2d80, ex=...) at activemq/transport/TransportFilter.cpp:41
> #19 0x00803c0b in activemq::transport::TransportFilter::fire
> (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:49
> #20 0x00803c64 in activemq::transport::TransportFilter::onException
> (this=0x8eb27f0, ex=...) at activemq/transport/TransportFilter.cpp:41
> #21 0x00801b13 in activemq::transport::IOTransport::fire
> (this=0x8eb27a0, ex=...) at activemq/transport/IOTransport.cpp:73
> #22 0x008023bf in activemq::transport::IOTransport::run
> (this=0x8eb27a0) at activemq/transport/IOTransport.cpp:246
> #23 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8eb2ea0) at decaf/lang/Thread.cpp:135
> #24 0x00a92847 in threadWorker (arg=0x8eb2ea0) at decaf/lang/Thread.cpp:188
> #25 0x00ddb96e in start_thread (arg=0xb7007b70) at pthread_create.c:300
> #26 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> Thread 4:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de0015 in pthread_cond_wait@@GLIBC_2.3.2 () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/pthread_cond_wait.S:122
> #2  0x0034f9dd in __pthread_cond_wait (cond=0x8eba6c8,
> mutex=0x8eba698) at forward.c:139
> #3  0x00a606a9 in
> decaf::internal::util::concurrent::ConditionImpl::wait
> (condition=0x8eba6c8) at
> decaf/internal/util/concurrent/unix/ConditionImpl.cpp:94
> #4  0x00ad1353 in decaf::util::concurrent::Mutex::wait
> (this=0x8eba584) at decaf/util/concurrent/Mutex.cpp:95
> #5  0x00ad029a in decaf::util::concurrent::CountDownLatch::await
> (this=0x8eba57c) at decaf/util/concurrent/CountDownLatch.cpp:56
> #6  0x0804af84 in JMSEndPointThread::stopCheck (this=0x8eba574,
> timeOut=-1) at main.cpp:226
> #7  0x0804bb46 in Consumer::run (this=0x8eba570) at main.cpp:379
> #8  0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8eba5b0) at decaf/lang/Thread.cpp:135
> #9  0x00a92847 in threadWorker (arg=0x8eba5b0) at decaf/lang/Thread.cpp:188
> #10 0x00ddb96e in start_thread (arg=0xb6806b70) at pthread_create.c:300
> #11 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> Thread 5:
> #0  0x00cfd422 in __kernel_vsyscall ()
> #1  0x00de2af9 in __lll_lock_wait () at
> ../nptl/sysdeps/unix/sysv/linux/i386/i686/../i486/lowlevellock.S:142
> #2  0x00dde13b in _L_lock_748 () from /lib/tls/i686/cmov/libpthread.so.0
> #3  0x00dddf61 in __pthread_mutex_lock (mutex=0x8eb05e0) at
> pthread_mutex_lock.c:61
> #4  0x0034fba6 in pthread_mutex_lock (mutex=0x8eb05e0) at forward.c:182
> #5  0x00a60aa1 in decaf::internal::util::concurrent::MutexImpl::lock
> (handle=0x8eb05e0) at
> decaf/internal/util/concurrent/unix/MutexImpl.cpp:71
> #6  0x00ad1412 in decaf::util::concurrent::Mutex::lock
> (this=0x8eb0a88) at decaf/util/concurrent/Mutex.cpp:75
> #7  0x00ad0ab5 in decaf::util::concurrent::Lock::lock
> (this=0xb6004ce0) at decaf/util/concurrent/Lock.cpp:54
> #8  0x00ad0c08 in Lock (this=0xfffffe00, object=0x8eb05e0,
> intiallyLocked=true) at decaf/util/concurrent/Lock.cpp:32
> #9  0x0081b8fd in
> activemq::transport::failover::FailoverTransport::oneway
> (this=0x8eb0a38, command=...) at
> activemq/transport/failover/FailoverTransport.cpp:186
> #10 0x00807f9f in
> activemq::transport::correlator::ResponseCorrelator::oneway
> (this=0x8eb11c0, command=...) at
> activemq/transport/correlator/ResponseCorrelator.cpp:82
> #11 0x007713be in activemq::core::ActiveMQConnection::oneway
> (this=0x8eb1210, command=...) at
> activemq/core/ActiveMQConnection.cpp:741
> #12 0x007b0a4f in activemq::core::ActiveMQSession::oneway
> (this=0x8ebab90, command=...) at activemq/core/ActiveMQSession.cpp:903
> #13 0x00795a1c in activemq::core::ActiveMQConsumer::acknowledge
> (this=0x8ebb270) at activemq/core/ActiveMQConsumer.cpp:860
> #14 0x0079f885 in
> activemq::core::TransactionSynhcronization::beforeEnd (this=0x8ebb9e8)
> at activemq/core/ActiveMQConsumer.cpp:85
> #15 0x007c0588 in
> activemq::core::ActiveMQTransactionContext::beforeEnd (this=0x8ebadc8)
> at activemq/core/ActiveMQTransactionContext.cpp:192
> #16 0x007c0d63 in activemq::core::ActiveMQTransactionContext::commit
> (this=0x8ebadc8) at activemq/core/ActiveMQTransactionContext.cpp:127
> #17 0x007ae460 in activemq::core::ActiveMQSession::commit
> (this=0x8ebab90) at activemq/core/ActiveMQSession.cpp:189
> #18 0x0069c1a0 in activemq::cmsutil::PooledSession::commit
> (this=0x8ebaf38) at activemq/cmsutil/PooledSession.h:87
> #19 0x0804c0fb in Consumer::onMessage (this=0x8eba570,
> message=0x8ebbec0) at main.cpp:455
> #20 0x0079972f in activemq::core::ActiveMQConsumer::dispatch
> (this=0x8ebb270, dispatch=...) at
> activemq/core/ActiveMQConsumer.cpp:1018
> #21 0x007bd5c3 in activemq::core::ActiveMQSessionExecutor::dispatch
> (this=0x8ebae78, dispatch=...) at
> activemq/core/ActiveMQSessionExecutor.cpp:129
> #22 0x007bd993 in activemq::core::ActiveMQSessionExecutor::iterate
> (this=0x8ebae78) at activemq/core/ActiveMQSessionExecutor.cpp:166
> #23 0x008007f3 in activemq::threads::DedicatedTaskRunner::run
> (this=0x8ebb550) at activemq/threads/DedicatedTaskRunner.cpp:111
> #24 0x00a93cf1 in decaf::lang::ThreadProperties::runCallback
> (properties=0x8ebaad8) at decaf/lang/Thread.cpp:135
> #25 0x00a92847 in threadWorker (arg=0x8ebaad8) at decaf/lang/Thread.cpp:188
> #26 0x00ddb96e in start_thread (arg=0xb6005b70) at pthread_create.c:300
> #27 0x00342a4e in clone () at ../sysdeps/unix/sysv/linux/i386/clone.S:130
> 
> 
> 
> I tried to resolve this issue changing configurations and code, but i
> always get an error or client freezes.
> Is it a bug? Any suggestions?
> 
> Thanks.
> Henrique
> 
> How to reproduce:
> 1. Start ActiveMQ
> 2. Send a massage './activemqTest -p1 -n1'
> 3. Run consumer './activemqTest -c1 -d10000'
> 4. Stop activemq when 'Starting delay...' message appears.
> 5. After activemq is down and delay is finish, consumer try commit
> (message 'Try commit') and unexpected exception occurs.
> 
> Following test code, based on ActiveMQ-CPP sample
> 'activemq-cpp-library-3.2.4/src/examples/main.cpp':
> 
> 
> /*
>  * 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.
>  */
> 
> // START SNIPPET: demo
> 
> #include <activemq/library/ActiveMQCPP.h>
> #include <activemq/cmsutil/SessionPool.h>
> #include <activemq/cmsutil/PooledSession.h>
> #include <activemq/cmsutil/ResourceLifecycleManager.h>
> #include <activemq/exceptions/ActiveMQException.h>
> #include <decaf/lang/Thread.h>
> #include <decaf/lang/Runnable.h>
> #include <decaf/util/concurrent/CountDownLatch.h>
> #include <decaf/util/concurrent/Mutex.h>
> #include <decaf/lang/Integer.h>
> #include <decaf/lang/Long.h>
> #include <decaf/lang/System.h>
> #include <activemq/core/ActiveMQConnectionFactory.h>
> #include <activemq/util/Config.h>
> #include <cms/Connection.h>
> #include <cms/Session.h>
> #include <cms/TextMessage.h>
> #include <cms/BytesMessage.h>
> #include <cms/MapMessage.h>
> #include <cms/ExceptionListener.h>
> #include <cms/MessageListener.h>
> #include <iostream>
> #include <memory>
> #include <map>
> #include <vector>
> #include <unistd.h>
> 
> using namespace activemq::core;
> using namespace activemq::cmsutil;
> using namespace decaf::util::concurrent;
> using namespace decaf::util;
> using namespace decaf::lang;
> using namespace cms;
> 
> 
> class ConnPoolManager : protected decaf::util::concurrent::Mutex,
>                         public ExceptionListener {
> private:
>     typedef std::map< int, activemq::cmsutil::SessionPool* > tSessionPoolMap;
> 
>     std::string _brokerURI;
>     ConnectionFactory* _connectionFactory;
>     Connection* _connection;
>     ResourceLifecycleManager* _lifecycleManager;
>     tSessionPoolMap _pools;
>     bool _started;
> 
> public:
>     ConnPoolManager() :
>             _brokerURI("tcp://localhost:61616"),
>             _connectionFactory(NULL),
>             _connection(NULL),
>             _lifecycleManager(NULL),
>             _started(false)
>     {
>     }
> 
>     virtual ~ConnPoolManager() {
>         cleanup();
>     }
> 
>     void setBrokerURI(const std::string& uri) {
>         _brokerURI = uri;
>     }
> 
>     bool startPools() {
>         synchronized(this) {
>             if(_started) return true;
>             try {
>                 createConnection();
>                 createPools();
>                 _started=true;
>                 return true;
>             } catch(CMSException& e) {
>                 std::cout<<"(ConnPoolManager::startPools)
> CMSException: " << e.getMessage() << std::endl;
>             } catch(...) {
>                 std::cout<<"(ConnPoolManager::startPools) Unknown
> exception!!!" << std::endl;
>             }
>         }
>         cleanup();
>         return false;
>     }
> 
>     PooledSession* getSession(cms::Session::AcknowledgeMode ack) {
>         if(_started) {
>             return _pools[ack]->takeSession();
>         } else if(startPools()) {
>             return _pools[ack]->takeSession();
>         } else {
>             return NULL;
>         }
>     }
> 
> private:
> 
>     void createConnection(){
>         // Create a ConnectionFactory
>         _connectionFactory =
> ConnectionFactory::createCMSConnectionFactory( _brokerURI );
> 
>         // Create a Connection
>         _connection = _connectionFactory->createConnection();
>         _connection->start();
>         _connection->setExceptionListener(this);
>     }
> 
>     void createPools(){
>         // Create lifecycle manager
>         _lifecycleManager = new activemq::cmsutil::ResourceLifecycleManager();
> 
>         // Create pools, one for each acknowledge type
>         for(    int ack = cms::Session::AUTO_ACKNOWLEDGE;
>                 ack <= cms::Session::INDIVIDUAL_ACKNOWLEDGE;
>                 ack++)
>         {
>             _pools[ack] =  new activemq::cmsutil::SessionPool(
>                     _connection,
>                     static_cast<cms::Session::AcknowledgeMode>(ack),
>                     _lifecycleManager);
>         }
>     }
> 
>     void cleanup() {
>         synchronized(this) {
>             releasePools();
>             if(_lifecycleManager!=NULL) {
>                 try { delete _lifecycleManager;
>                 } catch(...) {}
>             } _lifecycleManager = NULL;
> 
>             if(_connection!=NULL) {
>                 try { delete _connection;
>                 } catch(...) {}
>             } _connection = NULL;
>             if(_connectionFactory!=NULL) {
>                 try { delete _connectionFactory;
>                 } catch(...) {}
>             } _connectionFactory = NULL;
>             _started=false;
>         }
>     }
> 
>     void releasePools() {
>         for(tSessionPoolMap::iterator it=_pools.begin();
> it!=_pools.end(); it++) {
>             try {
>                 delete it->second;
>             } catch(...) { }
>         }
>         _pools.clear();
>     }
> 
>     // If something bad happens you see it here as this class is also been
>     // registered as an ExceptionListener with the connection.
>     virtual void onException( const CMSException& ex AMQCPP_UNUSED) {
>         std::cout<<"(ConnPoolManager::onException) CMS Exception
> occurred.  Shutting down client."<<std::endl;
>         ex.printStackTrace();
>         exit(1);
>     }
> 
> 
> 
> };
> 
> class JMSEndPointThread  : public Thread {
> private:
>     CountDownLatch _stopNow;
>     ConnPoolManager* _pools;
>     std::string _destinationString;
>     PooledSession* _session;
>     Destination* _destination;
>     bool _sessionTransacted;
> 
> protected:
> 
>     JMSEndPointThread(  ConnPoolManager* pools,
>                         const std::string& dest,
>                         bool sessionTransacted) :
>             _stopNow(1),
>             _pools(pools),
>             _destinationString(dest),
>             _session(NULL),
>             _destination(NULL),
>             _sessionTransacted(sessionTransacted)
>     {
>     }
> 
>     virtual bool prepareSession() {
>         cleanup();
>         // Create a Session
>         if( _sessionTransacted ) {
>             _session = _pools->getSession(Session::SESSION_TRANSACTED);
>         } else {
>             _session = _pools->getSession(Session::AUTO_ACKNOWLEDGE);
>         }
>         if(_session==NULL) return false;
> 
>         // Create the Queue destination
>         _destination = _session->createQueue( _destinationString );
>         return true;
>     }
> 
>     PooledSession* getSession() { return _session; }
>     Destination* getDestination() { return _destination; }
>     bool getTransacted() { return _sessionTransacted; }
> 
>     bool stopCheck(int timeOut=0) {
>         if(timeOut>=0) return _stopNow.await(timeOut);
>         else _stopNow.await();
>         return true;
>     }
> 
>     virtual void cleanup() {
>         // Destroy resources.
>         try{
>             if( _destination != NULL ) delete _destination;
>         } catch(...) { }
>         _destination = NULL;
> 
>         // Back session to pool
>         try{
>             if( _session != NULL ) _session->close();
>         } catch(...) { }
>         _session = NULL;
>     }
> 
> public:
> 
>     typedef std::vector<JMSEndPointThread*> List;
> 
>     virtual ~JMSEndPointThread() {
>         cleanup();
>     }
> 
>     void stopNow() {
>         _stopNow.countDown();
>     }
> 
> };
> 
> class Producer : public JMSEndPointThread {
> private:
> 
>     MessageProducer* _producer;
>     int _delay;
>     int _numMessages;
> 
> public:
> 
>     Producer(   ConnPoolManager* pools,
>                 const std::string& dest,
>                 int delay,
>                 int numMessages,
>                 bool sessionTransacted = false ) :
>         JMSEndPointThread(pools,dest,sessionTransacted)
>     {
>         _producer = NULL;
>         _delay = delay;
>         _numMessages = numMessages;
>     }
> 
>     virtual ~Producer() {
>         cleanup();
>     }
> 
>     virtual void run() {
>         std::cout<<"Producer started!!!! Thread: "<<Thread::getId()<<std::endl;
>         while(!stopCheck() && _numMessages) {
>             try {
>                 if(prepareProducer()) {
>                     sendMessages();
>                     continue;
>                 }
>             } catch(CMSException& e) {
>                 std::cout<<"(Producer::run) CMSException: " <<
> e.getMessage() << std::endl;
>             } catch(...) {
>                 std::cout<<"(Producer::run) Unknown exception!!!" << std::endl;
>             }
>             cleanup();
>             stopCheck(5000); // reconnect delay
>         }
>         std::cout<<"Producer end!!!! Thread: "<<Thread::getId()<<std::endl;
>         cleanup();
>     }
> 
> private:
> 
>     bool prepareProducer() {
>         // Get Session
>         if(!prepareSession()) return false;
>         // Create a MessageProducer from the Session to the Queue
>         _producer = getSession()->createProducer( getDestination() );
>         _producer->setDeliveryMode( DeliveryMode::PERSISTENT );
>         return true;
>     }
>     void sendMessages() {
>         // Create a messages
>         std::stringstream text;
>         text<<"Hello world! from thread " << Thread::getId();
> 
>         while(!stopCheck() && _numMessages) {
> 
>             TextMessage* message = getSession()->createTextMessage(
> text.str() );
>             message->setIntProperty( "Integer", _numMessages );
> 
>             // Tell the producer to send the message
>             std::cout<<"Sent message #"<<_numMessages<<" from thread
> "<<Thread::getId()<<std::endl;
>             _producer->send( message );
>             delete message;
>             --_numMessages;
> 
>             stopCheck(_delay);
>             if(getTransacted()) getSession()->commit();
>         }
>     }
> 
>     virtual void cleanup() {
>         try{
>             if( _producer != NULL ) delete _producer;
>         } catch(...) { }
>         _producer = NULL;
> 
>         JMSEndPointThread::cleanup();
>     }
> 
> };
> 
> 
> class Consumer :    public MessageListener,
>                     public JMSEndPointThread {
> 
> private:
> 
>     MessageConsumer* _consumer;
>     int _delay;
>     bool _useListener;
> 
> public:
> 
>     Consumer(   ConnPoolManager* pools,
>                 const std::string& dest,
>                 int delay,
>                 bool useListener = true,
>                 bool sessionTransacted = true ) :
>         JMSEndPointThread(pools,dest,sessionTransacted)
>     {
>         _consumer = NULL;
>         _delay = delay;
>         _useListener = useListener;
>     }
>     virtual ~Consumer(){
>         cleanup();
>     }
> 
>     virtual void run() {
>         std::cout<<"Consumer started!!!! Thread: "<<Thread::getId()<<std::endl;
>         while(!stopCheck()) {
>             try {
>                 if(prepareConsumer()) {
>                     // Wait while asynchronous messages come in.
>                     if(_useListener) {
>                         stopCheck(-1);
>                     } else {
>                         consumeLoop();
>                     }
>                     continue;
>                 }
>             } catch (CMSException& e) {
>                 std::cout<<"(Consumer::run) CMSException: " <<
> e.getMessage() << std::endl;
>             } catch(...) {
>                 std::cout<<"(Consumer::run) Unknown exception!!!" << std::endl;
>             }
>             cleanup();
>             stopCheck(5000); // reconnect delay
>         }
>         std::cout<<"Consumer end!!!! Thread: "<<Thread::getId()<<std::endl;
>         cleanup();
>     }
> 
> 
> private:
> 
>     bool prepareConsumer() {
>         // Get Session
>         if(!prepareSession()) return false;
>         // Create a MessageConsumer from the Session to the Queue
>         _consumer = getSession()->createConsumer( getDestination() );
>         if(_useListener)
>             _consumer->setMessageListener( this );
> 
>         return true;
>     }
> 
>     void consumeLoop() {
>         while(!stopCheck()) {
>             Message *msg = _consumer->receive(500);
>             if(msg!=NULL) {
>                 onMessage(msg);
>                 delete msg;
>             }
>         }
>     }
> 
>     // Called from the consumer since this class is a registered
> MessageListener.
>     virtual void onMessage( const Message* message ) {
> 
>         static int count = 0;
> 
>         try
>         {
>             count++;
>             const TextMessage* textMessage =
>                 dynamic_cast< const TextMessage* >( message );
>             std::string text = "";
> 
>             if( textMessage != NULL ) {
>                 text = textMessage->getText();
>             } else {
>                 text = "NOT A TEXTMESSAGE!";
>             }
> 
>             std::cout<<"Message #"<<count<<" Received: "<<text<<std::endl;
>             if(_delay) {
>                 std::cout<<"Starting delay..."<<std::endl;
>                 stopCheck(_delay);
>             }
> 
>         } catch (CMSException& e) {
>             std::cout<<"(Consumer::onMessage) CMSException: " <<
> e.getMessage() << std::endl;
>         } catch(...) {
>             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
> std::endl;
>         }
> 
>         try
>         {   // Commit all messages.
>             if( getTransacted() ) {
>                 std::cout<<"Try commit....."<<std::endl;
>                 getSession()->commit();
>                 std::cout<<"Commit OK!!!"<<std::endl;
>             }
>         } catch (CMSException& e) {
>             std::cout<<"(Consumer::onMessage) CMSException: " <<
> e.getMessage() << std::endl;
>         } catch(...) {
>             std::cout<<"(Consumer::onMessage) Unknown exception!!!" <<
> std::endl;
>         }
> 
>     }
> 
>     virtual void cleanup() {
>         try{
>             if( _consumer != NULL ) delete _consumer;
>         } catch (...) { }
>         _consumer = NULL;
>         JMSEndPointThread::cleanup();
>     }
> };
> 
> 
> class ScopedActiveMQLibrary {
> public:
>     ScopedActiveMQLibrary() {
>         activemq::library::ActiveMQCPP::initializeLibrary();
>     }
>     virtual ~ScopedActiveMQLibrary() {
>         activemq::library::ActiveMQCPP::shutdownLibrary();
>     }
> };
> 
> class AppTest : private ScopedActiveMQLibrary {
> private:
>     int _consumer;
>     int _producer;
>     int _numMessages;
>     bool _useListener;
>     int _delay;
>     ConnPoolManager _poolManager;
>     JMSEndPointThread::List _endPointList;
> 
>     static AppTest* _app;
>     static CountDownLatch _terminationRequest;
> 
> public:
>     AppTest(int argc, char** argv) :
>             _consumer(0),
>             _producer(0),
>             _numMessages(1),
>             _useListener(false),
>             _delay(0)
>     {
>         _app=this;
>         int c;
>         while ((c = getopt (argc, argv, "lc:p:n:d:")) != -1) {
>             switch(c) {
>             case 'l': _useListener=true; break;
>             case 'c': _consumer=atoi(optarg); break;
>             case 'p': _producer=atoi(optarg); break;
>             case 'n': _numMessages=atoi(optarg); break;
>             case 'd': _delay=atoi(optarg); break;
>             }
>         }
>         if(_consumer<0) _consumer=0;
>         if(_producer<0) _producer=0;
>         _poolManager.setBrokerURI(
>                 "failover://("
>                 "tcp://localhost:61616"
>                 "?transport.useInactivityMonitor=false"
>                 ")?timeout=1000"
>                 "&cms.RedeliveryPolicy.maximumRedeliveries=-1"
>                 );
>         _poolManager.startPools();
>     }
> 
>     virtual ~AppTest() {
>     }
> 
>     int run() {
> 
>         // Install signal handler
>         if(!installSigAction())
>             return 1;
> 
>         // Create consumer/producer objects
>         loadEndPoins();
> 
>         // Start the producer/consumer thread.
>         startEndPoins();
> 
>         // start another tasks
>         // ...
> 
>         // Wait for termination request
>         /*if(_consumer) {
>             _terminationRequest.await();
>             stopEndPoins();
>         }*/
> 
>         // join threads
>         joinEndPoins();
> 
>         return 0;
>     }
> 
> private:
> 
>     void loadEndPoins() {
>         for(int i=0; i<_consumer;i++) {
>             _endPointList.push_back( new
> Consumer(&_poolManager,"TEST.FOO",_delay,_useListener) );
>         }
>         for(int i=0; i<_producer;i++) {
>             _endPointList.push_back( new
> Producer(&_poolManager,"TEST.FOO",_delay,_numMessages) );
>         }
>     }
> 
>     void startEndPoins() {
>         for(JMSEndPointThread::List::iterator it=_endPointList.begin();
>             it!=_endPointList.end(); it++ )
>             (**it).start();
>     }
> 
>     void stopEndPoins() {
>         for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
>             it!=_endPointList.rend(); it++ )
>             (**it).stopNow();
>     }
> 
>     void joinEndPoins() {
>         for(JMSEndPointThread::List::reverse_iterator it=_endPointList.rbegin();
>             it!=_endPointList.rend(); it++ ) {
>             (**it).join();
>             delete *it;
>             *it = NULL;
>         }
>         _endPointList.clear();
>     }
> 
>     static void signalHandler(int sig) {
>         _app->_terminationRequest.countDown();
>         _app->stopEndPoins();
>     }
> 
>     bool installSigAction() {
>         struct sigaction action;
>         memset(&action, 0, sizeof(action));
>         action.sa_handler = signalHandler;
>         if( sigaction(SIGTERM, &action, NULL)<0 ||
>             sigaction(SIGQUIT, &action, NULL)<0 ||
>             sigaction(SIGINT,  &action, NULL)<0 )
>         { return false; }
>         return true;
>     }
> 
>     int termWait() {
>         sigset_t sset;
>         sigemptyset(&sset);
>         sigaddset(&sset, SIGINT);
>         sigaddset(&sset, SIGQUIT);
>         sigaddset(&sset, SIGTERM);
>         sigprocmask(SIG_BLOCK, &sset, NULL);
>         int sig;
>         sigwait(&sset, &sig);
>         return sig;
>     }
> 
> };
> AppTest* AppTest::_app = NULL;
> CountDownLatch AppTest::_terminationRequest(1);
> 
> 
> void my_unexpected() {
>     throw activemq::exceptions::ActiveMQException();
> }
> 
> int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
>     //std::set_unexpected(my_unexpected);
>     AppTest test(argc, argv);
>     return test.run();
> }
> 
> // END SNIPPET: demo

-- 
Tim Bish
------------
FuseSource
Email: tim.bish@fusesource.com
Web: http://fusesource.com
Twitter: tabish121
Blog: http://timbish.blogspot.com/



Mime
View raw message