| /* |
| * 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/concurrent/Thread.h> |
| #include <activemq/concurrent/Runnable.h> |
| #include <activemq/core/ActiveMQConnectionFactory.h> |
| #include <activemq/util/Integer.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 <stdlib.h> |
| #include <iostream> |
| |
| using namespace activemq::core; |
| using namespace activemq::util; |
| using namespace activemq::concurrent; |
| using namespace cms; |
| using namespace std; |
| |
| class HelloWorldProducer : public Runnable { |
| private: |
| |
| Connection* connection; |
| Session* session; |
| Destination* destination; |
| MessageProducer* producer; |
| int numMessages; |
| bool useTopic; |
| std::string brokerURI; |
| |
| public: |
| |
| HelloWorldProducer( const std::string& brokerURI, |
| int numMessages, bool useTopic = false ){ |
| connection = NULL; |
| session = NULL; |
| destination = NULL; |
| producer = NULL; |
| this->numMessages = numMessages; |
| this->useTopic = useTopic; |
| this->brokerURI = brokerURI; |
| } |
| |
| virtual ~HelloWorldProducer(){ |
| cleanup(); |
| } |
| |
| virtual void run() { |
| try { |
| // Create a ConnectionFactory |
| ActiveMQConnectionFactory* connectionFactory = |
| new ActiveMQConnectionFactory( brokerURI ); |
| |
| // Create a Connection |
| connection = connectionFactory->createConnection(); |
| connection->start(); |
| |
| // free the factory, we are done with it. |
| delete connectionFactory; |
| |
| // Create a Session |
| session = connection->createSession( Session::AUTO_ACKNOWLEDGE ); |
| |
| // Create the destination (Topic or Queue) |
| if( useTopic ) { |
| destination = session->createTopic( "TEST.FOO" ); |
| } else { |
| destination = session->createQueue( "TEST.FOO" ); |
| } |
| |
| // Create a MessageProducer from the Session to the Topic or Queue |
| producer = session->createProducer( destination ); |
| producer->setDeliveryMode( DeliveryMode::NON_PERSISTENT ); |
| |
| // 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 ); |
| |
| message->setIntProperty( "Integer", ix ); |
| |
| // Tell the producer to send the message |
| printf( "Sent message #%d from thread %s\n", ix, threadIdStr.c_str() ); |
| producer->send( message ); |
| |
| delete message; |
| } |
| |
| }catch ( CMSException& e ) { |
| e.printStackTrace(); |
| } |
| } |
| |
| private: |
| |
| void cleanup(){ |
| |
| // Destroy resources. |
| try{ |
| if( destination != NULL ) delete destination; |
| }catch ( CMSException& e ) { e.printStackTrace(); } |
| destination = NULL; |
| |
| try{ |
| if( producer != NULL ) delete producer; |
| }catch ( CMSException& e ) { e.printStackTrace(); } |
| producer = NULL; |
| |
| // Close open resources. |
| try{ |
| if( session != NULL ) session->close(); |
| if( connection != NULL ) connection->close(); |
| }catch ( CMSException& e ) { e.printStackTrace(); } |
| |
| try{ |
| if( session != NULL ) delete session; |
| }catch ( CMSException& e ) { e.printStackTrace(); } |
| session = NULL; |
| |
| try{ |
| if( connection != NULL ) delete connection; |
| }catch ( CMSException& e ) { e.printStackTrace(); } |
| connection = NULL; |
| } |
| }; |
| |
| class HelloWorldConsumer : public ExceptionListener, |
| public MessageListener, |
| public Runnable { |
| |
| private: |
| |
| Connection* connection; |
| Session* session; |
| Destination* destination; |
| MessageConsumer* consumer; |
| long waitMillis; |
| bool useTopic; |
| std::string brokerURI; |
| |
| public: |
| |
| HelloWorldConsumer( const std::string& brokerURI, |
| long waitMillis, bool useTopic = false ){ |
| connection = NULL; |
| session = NULL; |
| destination = NULL; |
| consumer = NULL; |
| this->waitMillis = waitMillis; |
| this->useTopic = useTopic; |
| this->brokerURI = brokerURI; |
| } |
| virtual ~HelloWorldConsumer(){ |
| cleanup(); |
| } |
| |
| virtual void run() { |
| |
| try { |
| |
| // Create a ConnectionFactory |
| ActiveMQConnectionFactory* connectionFactory = |
| new ActiveMQConnectionFactory( brokerURI ); |
| |
| // Create a Connection |
| connection = connectionFactory->createConnection(); |
| delete connectionFactory; |
| connection->start(); |
| |
| connection->setExceptionListener(this); |
| |
| // Create a Session |
| session = connection->createSession( Session::AUTO_ACKNOWLEDGE ); |
| |
| // Create the destination (Topic or Queue) |
| if( useTopic ) { |
| destination = session->createTopic( "TEST.FOO" ); |
| } else { |
| destination = session->createQueue( "TEST.FOO" ); |
| } |
| |
| // Create a MessageConsumer from the Session to the Topic or Queue |
| consumer = session->createConsumer( destination ); |
| |
| consumer->setMessageListener( this ); |
| |
| std::cout.flush(); |
| std::cerr.flush(); |
| |
| // Sleep while asynchronous messages come in. |
| Thread::sleep( waitMillis ); |
| |
| } catch (CMSException& e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| // 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 ); |
| string text = textMessage->getText(); |
| |
| printf( "Message #%d Received: %s\n", count, text.c_str() ); |
| } catch (CMSException& e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| // 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) { |
| printf("JMS Exception occured. Shutting down client.\n"); |
| } |
| |
| private: |
| |
| void cleanup(){ |
| |
| //************************************************* |
| // Always close destination, consumers and producers before |
| // you destroy their sessions and connection. |
| //************************************************* |
| |
| // Destroy resources. |
| try{ |
| if( destination != NULL ) delete destination; |
| }catch (CMSException& e) { e.printStackTrace(); } |
| destination = NULL; |
| |
| try{ |
| if( consumer != NULL ) delete consumer; |
| }catch (CMSException& e) { e.printStackTrace(); } |
| consumer = NULL; |
| |
| // Close open resources. |
| try{ |
| if( session != NULL ) session->close(); |
| if( connection != NULL ) connection->close(); |
| }catch (CMSException& e) { e.printStackTrace(); } |
| |
| // Now Destroy them |
| try{ |
| if( session != NULL ) delete session; |
| }catch (CMSException& e) { e.printStackTrace(); } |
| session = NULL; |
| |
| try{ |
| if( connection != NULL ) delete connection; |
| }catch (CMSException& e) { e.printStackTrace(); } |
| connection = NULL; |
| } |
| }; |
| |
| int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) { |
| |
| std::cout << "=====================================================\n"; |
| std::cout << "Starting the example:" << std::endl; |
| std::cout << "-----------------------------------------------------\n"; |
| |
| // Set the URI to point to the IPAddress of your broker. |
| // add any optional params to the url to enable things like |
| // tightMarshalling or tcp logging etc. |
| std::string brokerURI = |
| "tcp://127.0.0.1:61616" |
| "?wireFormat=openwire"; |
| // "&commandTracingEnabled=true" |
| // "&tcpTracingEnabled=true" |
| // "&wireFormat.tightEncodingEnabled=true"; |
| |
| //============================================================ |
| // set to true to use topics instead of queues |
| // Note in the code above that this causes createTopic or |
| // createQueue to be used in both consumer an producer. |
| //============================================================ |
| bool useTopics = true; |
| |
| HelloWorldProducer producer( brokerURI, 2000, useTopics ); |
| HelloWorldConsumer consumer( brokerURI, 12000, useTopics ); |
| |
| // Start the consumer thread. |
| Thread consumerThread( &consumer ); |
| consumerThread.start(); |
| |
| // Give the consumer a but to start up. |
| Thread::sleep( 75 ); |
| |
| // Start the producer thread. |
| Thread producerThread( &producer ); |
| producerThread.start(); |
| |
| // Wait for the threads to complete. |
| producerThread.join(); |
| consumerThread.join(); |
| |
| std::cout << "-----------------------------------------------------\n"; |
| std::cout << "Finished with the example, ignore errors from this" |
| << std::endl |
| << "point on as the sockets breaks when we shutdown." |
| << std::endl; |
| std::cout << "=====================================================\n"; |
| } |
| |
| // END SNIPPET: demo |
