Thanks for reply! issue opened: https://issues.apache.org/jira/browse/AMQCPP-354
Regards Henrique 2011/3/1 Timothy Bish <tabish...@gmail.com>: > 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.b...@fusesource.com > Web: http://fusesource.com > Twitter: tabish121 > Blog: http://timbish.blogspot.com/ > > >
