Chinaunix首页 | 论坛 | 博客

大道之行，天下为公litanhua.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

litanhua_

  • 博客访问： 87904
  • 博文数量： 19
  • 博客积分： 487
  • 博客等级： 下士
  • 技术积分： 205
  • 用 户 组： 普通用户
  • 注册时间： 2011-03-18 15:57
文章分类

全部博文（19）

文章存档

2011年（19）

我的朋友
最近访客
推荐博文
相关博文
Producer & Consumer for Message Service

分类： C/C++

2011-04-20 11:07:33

Producer.hpp
  1. #ifndef _PRODUCER_H
  2. #define _PRODUCER_H

  4. #include <string>
  5. #include <memory>

  7. #include <cms/Session.h>
  8. #include <cms/MessageProducer.h>
  9. #include <cms/MessageConsumer.h>
  10. #include <cms/MessageListener.h>
  11. #include <cms/ConnectionFactory.h>

  13. #include <cms/Connection.h>
  14. #include <decaf/lang/Runnable.h>
  15. #include <decaf/util/concurrent/CountDownLatch.h>

  17. #include <activemq/commands/ActiveMQMessage.h>
  18. #include <activemq/commands/ProducerInfo.h>
  19. #include <activemq/library/ActiveMQCPP.h>

  21. #include "Message.hpp"
  22. using namespace PWRD;
  23. using namespace std;
  24. using namespace activemq;
  25. using namespace cms;
  26. using namespace decaf;
  27. using namespace decaf::lang;
  28. using namespace decaf::lang::exceptions;

  30. namespace PWRD{
  31.     namespace ActiveMQ{

  33.         /**
  34.          * A sample Prodcer that will only send Messages on its Topic when it has
  35.          * received an advisory indicating that there is an active MessageConsumer
  36.          * on the Topic. Once another message comes in indicating that there is no
  37.          * longer a consumer then this Producer stops producing again.
  38.          *
  39.          * @since 3.0
  40.          */
  41. #define DEFAULTBROKERURI "failover:(tcp://127.0.0.1:61616)"
  42.         class Producer : public decaf::lang::Runnable, public cms::MessageListener {
  43.         
  44.             private:

  46.                 volatile bool consumerOnline;
  47.                 volatile bool shutdown;
  48.                 decaf::util::concurrent::CountDownLatch shutdownLatch;

  50.                 cms::Session* session;
  51.                 std::auto_ptr<cms::MessageConsumer> consumer;
  52.                 std::auto_ptr<cms::MessageProducer> producer;
  53.                 std::string brokerURI;
  54.                 auto_ptr<cms::Connection> connection;

  56.             public:

  58.                 Producer(string _brokerURI = DEFAULTBROKERURI );
  59.                 virtual ~Producer();

  61.                 /**
  62.                  * Shut down the processing that occurs in the Run method.
  63.                  */
  64.                 void stop();

  66.                 /**
  67.                  * Run the producer code.
  68.                  */
  69.                 virtual void run();

  71.                 void transfer(Packet *pack);

  73.                 /**
  74.                  * Async Message callback.
  75.                  */
  76.                 virtual void onMessage( const cms::Message* message );

  78.         };

  80.     }
  81. }

  83. #endif /* _ACTIVEMQCPP_EXAMPLES_ADVISORIES_ADVISORYPRODUCER_H_ */
Producer.cpp
  1. #include "Producer.hpp"

  3. #include <cms/Topic.h>
  4. #include <cms/Message.h>
  5. #include <cms/TextMessage.h>
  6. #include <decaf/lang/exceptions/NullPointerException.h>
  7. #include <decaf/lang/Integer.h>


  10. namespace PWRD{
  11.     namespace ActiveMQ{
  12.         ////////////////////////////////////////////////////////////////////////////////
  13.         Producer::Producer(string _brokerURI) : shutdownLatch(1) {

  15.             brokerURI = _brokerURI;
  16.             if(!brokerURI.empty()){
  17.                 auto_ptr<cms::ConnectionFactory> connectionFactory(
  18.                         cms::ConnectionFactory::createCMSConnectionFactory( brokerURI ) );

  20.                 // Create a Connection
  21.                 try{
  22.                     connection.reset( connectionFactory->createConnection() );
  23.                 } catch( CM***ception& e ) {
  24.                     e.printStackTrace();
  25.                 }

  27.                 // Create the Session
  28.                 session = connection->createSession();
  29.                 if( session == NULL ) {
  30.                     throw NullPointerException(
  31.                             __FILE__, __LINE__, "Session Object passed was Null." );
  32.                 }

  34.                 std::auto_ptr<cms::Topic> destination( session->createTopic(
  35.                             "HEART-BEAT-CHANNEL" ) );
  36.                 std::auto_ptr<cms::Topic> advisories( session->createTopic(
  37.                             "ActiveMQ.Advisory.Consumer.Topic.HEART-BEAT-CHANNEL" ) );
  38.                 this->producer.reset( session->createProducer( destination.get() ) );
  39.                 this->consumer.reset( session->createConsumer( advisories.get() ) );
  40.                 this->consumer->setMessageListener( this );

  42.                 connection->start();
  43.             }

  45.             this->shutdown = false;
  46.             this->consumerOnline = false;
  47.         }

  49.         ////////////////////////////////////////////////////////////////////////////////
  50.         Producer::~Producer() {
  51.             connection->stop();
  52.         }

  54.         ////////////////////////////////////////////////////////////////////////////////
  55.         void Producer::stop() {
  56.             this->shutdown = true;
  57.             this->shutdownLatch.await( 3000 );
  58.         }

  60.         ////////////////////////////////////////////////////////////////////////////////
  61.         void Producer::run() {

  63.             /*
  64.             while( !this->shutdown ) {

  66.                 if( this->consumerOnline ) {

  68.                     std::auto_ptr<cms::TextMessage> message(
  69.                             this->session->createTextMessage( "Alive" ) );

  71.                     //this->producer->send( message.get() );
  72.                     Packet *packet = new Packet();
  73.                     packet->set_ip("10.14.2.40");
  74.                     packet->set_file("test");
  75.                     packet->set_offset(0);
  76.                     packet->set_data("Hello world");
  77.                     packet->set_type(ENORMAL);
  78.                     this->transfer(packet);

  80.                     Thread::sleep( 1000 );
  81.                     delete packet;
  82.                 }
  83.             }

  85.             this->shutdownLatch.countDown();
  86.             */
  87.         }

  89.         void Producer::transfer(Packet *pack){
  90.             if( !this->shutdown){
  91.                 if(this->consumerOnline){

  93.                     std::auto_ptr<cms::TextMessage> message(
  94.                             this->session->createTextMessage(pack->data()));

  96.                     this->producer->send( message.get() );
  97.                     logs.write_log(NORMAL, "Send message to default destination");

  99.                 }    
  100.                 else{
  101.                     logs.write_log(NORMAL, "There are no consumer");    
  102.                 }
  103.             }

  105.             this->shutdownLatch.countDown();
  106.         }

  108.         ////////////////////////////////////////////////////////////////////////////////
  109.         void Producer::onMessage( const cms::Message* message ) {

  111.             if( message->getCMSType() == "Advisory" ) {

  113.                 std::cout << "Received an Advisory Message!" << std::endl;

  115.                 if( message->propertyExists( "consumerCount" ) ) {

  117.                     std::string consumerCount = message->getStringProperty( "consumerCount" );
  118.                     std::cout << "Number of Consumers = " << consumerCount << std::endl;
  119.                     this->consumerOnline = Integer::parseInt( consumerCount ) > 0 ? true : false;
  120.                 }

  122.             } else {
  123.                 std::cout << "Received a Non-Advisory Message!" << std::endl;
  124.             }

  126.         }
  127.     }
  128. }
ProducerMain.cpp
  1. #include "Producer.hpp"

  3. #include <decaf/lang/Thread.h>
  4. #include <decaf/lang/Runnable.h>
  5. #include <activemq/library/ActiveMQCPP.h>
  6. #include <cms/ConnectionFactory.h>
  7. #include <cms/Connection.h>
  8. #include <cms/Session.h>
  9. #include <stdlib.h>
  10. #include <iostream>
  11. #include <memory>

  13. using namespace cms;
  14. using namespace std;
  15. using namespace PWRD;
  16. using namespace PWRD::ActiveMQ;

  18. using namespace decaf;
  19. using namespace decaf::lang;
  20. using namespace decaf::lang::exceptions;
  21. ////////////////////////////////////////////////////////////////////////////////
  22. int main( int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED ) {

  24.     // We must always init the library first before using any methods in it.
  25.     activemq::library::ActiveMQCPP::initializeLibrary();

  27.     std::cout << "=====================================================\n";
  28.     std::cout << "Starting the example:" << std::endl;
  29.     std::cout << "-----------------------------------------------------\n";

  31.     // Set the URI to point to the IPAddress of your broker.
  32.     // add any optional params to the url to enable things like
  33.     // tightMarshalling or tcp logging etc. See the CMS web site for
  34.     // a full list of configuration options.
  35.     //
  36.     // http://activemq.apache.org/cms/
  37.     //
  38.     std::string brokerURI = "failover:(tcp://10.14.2.40:61616)";

  40.     {
  41.         // Create the Connection
  42.         Producer producer( brokerURI );
  43.         Thread runner( &producer );
  44.         runner.start();

  46.         // Start the Connection now.

  48.         // Wait until we are told to quit.
  49.         while(1){
  50.             Packet *packet = new Packet();
  51.             packet->set_ip("10.14.2.40");
  52.             packet->set_file("test");
  53.             packet->set_offset(0);
  54.             packet->set_data("Hello world");
  55.             packet->set_type(ENORMAL);
  56.             producer.transfer(packet);

  58.             Thread::sleep( 1000 );
  59.             delete packet;

  61.         }
  62.         std::cout << "Press 'q' to quit" << std::endl;
  63.         while( std::cin.get() != 'q') {}

  65.         // Shutdown now
  66.         producer.stop();
  67.     }

  69.     std::cout << "-----------------------------------------------------\n";
  70.     std::cout << "Finished with the example." << std::endl;
  71.     std::cout << "=====================================================\n";

  73.     // We must also always remember to shut down the Library when done.
  74.     activemq::library::ActiveMQCPP::shutdownLibrary();

  76.     return 0;
  77. }
 
/*******************************************************************************************/
 
Consumer.hpp
  1. #ifndef _CONSUMER_H_
  2. #define _CONSUMER_H_

  4. #include <string>
  5. #include <memory>

  7. #include <cms/Closeable.h>
  8. #include <cms/Session.h>
  9. #include <cms/MessageConsumer.h>
  10. #include <cms/MessageListener.h>
  11. #include <cms/MessageListener.h>
  12. #include <cms/ConnectionFactory.h>
  13. #include "Message.hpp"
  14. #include <activemq/commands/ActiveMQMessage.h>
  15. #include <activemq/commands/ProducerInfo.h>

  17. using namespace std;
  18. using namespace PWRD;
  19. using namespace activemq;
  20. using namespace cms;
  21. using namespace activemq::commands;

  23. namespace PWRD{
  24.     namespace ActiveMQ{

  26.         /**
  27.          * A simple Consumer that compliements the AdvisoryProducer example. This
  28.          * consumer listens on the Topic that the Producer is waiting to publish on
  29.          * and will display the count of Producers that are active on the Topic
  30.          * any time that it sees an advisory message indicating a consumer has
  31.          * stopped or started.
  32.          *
  33.          * @since 3.0
  34.          */
  35. #define FILEPATH "LocalFile"
  36. #define DEFAULTBROKERURL "failover:(tcp://127.0.0.1:61616)"
  37.         class Consumer : public cms::Closeable, public cms::MessageListener {

  39.             private:

  41.                 cms::Session* session;
  42.                 std::auto_ptr<cms::MessageConsumer> consumer;
  43.                 std::auto_ptr<cms::MessageConsumer> advisoryConsumer;
  44.                 auto_ptr<cms::Connection> connection;
  45.                 FILE *fp;
  46.                 const char *path_;
  47.                 std::string brokerURI;
  48.             protected:
  49.                 int fini();
  50.                 int write();

  52.             public:

  54.                 Consumer(string _brokerURI, const char *path = FILEPATH);
  55.                 int init(const char *path = FILEPATH);
  56.                 virtual ~Consumer();

  58.                 /**
  59.                  * Close down Consumer resources.
  60.                  */
  61.                 virtual void close() throw( cms::CM***ception );

  63.                 /**
  64.                  * Async Message callback.
  65.                  */
  66.                 virtual void onMessage( const cms::Message* message );

  68.         };
  69.     }
  70. }

  72. #endif /* _ACTIVEMQCPP_EXAMPLES_ADVISORIES_ADVISORYCONSUMER_H_ */
Consumer.cpp
  1. #include "Consumer.hpp"

  3. #include <cms/Topic.h>
  4. #include <cms/Message.h>
  5. #include <cms/TextMessage.h>
  6. #include <decaf/lang/exceptions/NullPointerException.h>

  8. #include <activemq/commands/ActiveMQMessage.h>
  9. #include <activemq/commands/ProducerInfo.h>

  11. using namespace std;
  12. using namespace decaf;
  13. using namespace decaf::lang;
  14. using namespace cms;
  15. using namespace decaf::lang::exceptions;

  17. namespace PWRD{
  18.     namespace ActiveMQ{
  19.         ////////////////////////////////////////////////////////////////////////////////
  20.         Consumer::Consumer(string _brokerURI, const char *path) {

  22.             brokerURI = _brokerURI;
  23.             if(!brokerURI.empty()){
  24.                 auto_ptr<cms::ConnectionFactory> connectionFactory(
  25.                         cms::ConnectionFactory::createCMSConnectionFactory( brokerURI ) );

  27.                 // Create a Connection
  28.                 try{
  29.                     connection.reset( connectionFactory->createConnection() );
  30.                 } catch( CM***ception& e ) {
  31.                     e.printStackTrace();
  32.                 }

  34.                 // Create the Session
  35.                 session = connection->createSession();
  36.                 if( session == NULL ) {
  37.                     throw NullPointerException(
  38.                             __FILE__, __LINE__, "Session Object passed was Null." );
  39.                 }

  41.                 std::auto_ptr<cms::Topic> destination( session->createTopic(
  42.                             "HEART-BEAT-CHANNEL" ) );
  43.                 std::auto_ptr<cms::Topic> advisories( session->createTopic(
  44.                             "ActiveMQ.Advisory.Consumer.Topic.HEART-BEAT-CHANNEL" ) );
  45.                 this->consumer.reset( session->createConsumer( advisories.get() ) );
  46.                 this->consumer->setMessageListener( this );
  47.                 connection->start();
  48.             }

  50.             init(path);
  51.         }

  53.         ////////////////////////////////////////////////////////////////////////////////
  54.         Consumer::~Consumer() {
  55.             this->fini();
  56.             connection->stop();
  57.         }

  59.         ////////////////////////////////////////////////////////////////////////////////
  60.         void Consumer::close() throw( cms::CM***ception ) {
  61.             this->consumer.reset( NULL );
  62.         }

  64.         int Consumer::init(const char *path){
  65.             path_ = path;
  66.             fp = fopen(path, "a+");
  67.             if(NULL == fp){
  68.                 logs.write_log(NORMAL, "Can't open this file: %s", path_);    
  69.                 return -1;
  70.             }
  71.             return 1;
  72.         }

  74.         int Consumer::fini(){
  75.             return fclose(fp);
  76.         }
  77.         ////////////////////////////////////////////////////////////////////////////////
  78.         void Consumer::onMessage( const cms::Message* message ) {

  80.             if( message->getCMSType() == "Advisory" ) {

  82.                 const ActiveMQMessage* amqMessage =
  83.                     dynamic_cast<const ActiveMQMessage*>( message );

  85.                 // If you want you can get the ProducerInfo for instance, you could get
  86.                 // the ConsumerInfo and ConnectionInfo.
  87.                 if( amqMessage != NULL && amqMessage->getDataStructure() != NULL ) {
  88.                     const ProducerInfo* info = dynamic_cast<const ProducerInfo*>(
  89.                             amqMessage->getDataStructure().get() );

  91.                     std::cout << "Got ProducerInfo for producer: " << info->getProducerId()->toString() << std::endl;
  92.                 }

  94.                 if( message->propertyExists( "producerCount" ) ) {
  95.                     std::string producerCount = message->getStringProperty( "producerCount" );
  96.                     std::cout << "Number of Producers = " << producerCount << std::endl;
  97.                 }

  99.             } else {

  101.                 const cms::TextMessage* txtMessage =
  102.                     dynamic_cast<const cms::TextMessage*>( message );

  104.                 if( txtMessage != NULL ) {
  105.                     string text = txtMessage->getText();
  106.                     cout << text << endl;
  107.                     int len = fwrite(text.c_str(), 1, text.length(), fp);
  108.                     fflush(fp);
  109.                     logs.write_log(NORMAL, "Consumer write %d bytes to file %s", len, path_);
  110.                 }
  111.             }
  112.         }
  113.     }
  114. }
ConsumerMain.cpp
  1. #include "Consumer.hpp"

  3. #include <activemq/library/ActiveMQCPP.h>
  4. #include <cms/ConnectionFactory.h>
  5. #include <cms/Connection.h>
  6. #include <cms/Session.h>
  7. #include <stdlib.h>
  8. #include <iostream>
  9. #include <memory>

  11. using namespace cms;
  12. using namespace std;
  13. using namespace PWRD;
  14. using namespace PWRD::ActiveMQ;
  15. using namespace decaf;
  16. using namespace decaf::lang;
  17. using namespace decaf::lang::exceptions;

  19. ////////////////////////////////////////////////////////////////////////////////
  20. int main( int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED ) {

  22.     // We must always init the library first before using any methods in it.
  23.     activemq::library::ActiveMQCPP::initializeLibrary();

  25.     std::cout << "=====================================================\n";
  26.     std::cout << "Starting the example:" << std::endl;
  27.     std::cout << "-----------------------------------------------------\n";

  29.     // Set the URI to point to the IPAddress of your broker.
  30.     // add any optional params to the url to enable things like
  31.     // tightMarshalling or tcp logging etc. See the CMS web site for
  32.     // a full list of configuration options.
  33.     //
  34.     // http://activemq.apache.org/cms/
  35.     //
  36.     std::string brokerURI = "failover:(tcp://10.14.2.45:61616)";

  38.     /*
  39.     // Create the Connection
  40.     auto_ptr<cms::ConnectionFactory> connectionFactory(
  41.             cms::ConnectionFactory::createCMSConnectionFactory( brokerURI ) );

  43.     auto_ptr<cms::Connection> connection;

  45.     // Create a Connection
  46.     try{
  47.         connection.reset( connectionFactory->createConnection() );
  48.     } catch( CM***ception& e ) {
  49.         e.printStackTrace();
  50.         return 1;
  51.     }

  53.     // Create the Session
  54.     std::auto_ptr<cms::Session> session( connection->createSession() );
  55.     */

  57.     // Create the producer and run it.
  58.     Consumer consumer( brokerURI, "Localfile" );

  60.     // Start the Connection now.

  62.     // Wait until we are told to quit.
  63.     std::cout << "Press 'q' to quit" << std::endl;
  64.     while( std::cin.get() != 'q') {}

  66.     // Shutdown now
  67.     consumer.close();

  69.     std::cout << "-----------------------------------------------------\n";
  70.     std::cout << "Finished with the example." << std::endl;
  71.     std::cout << "=====================================================\n";

  73.     // We must also always remember to shut down the Library when done.
  74.     activemq::library::ActiveMQCPP::shutdownLibrary();

  76.     return 0;
  77. }
阅读(1543) | 评论(0) | 转发(0) |
0

上一篇：所谓一见钟情，不过见色起意

下一篇：boost::asio async_read vs async_receive

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6