ActiveMQCPP version: 3.7.1
AcitveMQBroker version: 5.10.0
Here is a simple example. The code includes both consumer and producer
// START SNIPPET: demo
#include <activemq/library/ActiveMQCPP.h>
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.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 <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <memory>
#include <decaf/util/Random.h>
using namespace activemq::core;
using namespace decaf::util::concurrent;
using namespace decaf::util;
using namespace decaf::lang;
using namespace cms;
using namespace std;
#define QUEUE_NAME "eventQueue"
#define NAME_BYTE_LEN 16
class HelloWorldProducer : public ExceptionListener,
public MessageListener,
public Runnable {
private:
CountDownLatch latch;
CountDownLatch doneLatch;
Connection* connection;
Session* session;
Destination* destination;
MessageProducer* producer;
int numMessages;
bool useTopic;
bool sessionTransacted;
std::string brokerURI;
bool bReciveMessage;
long waitMillis;
private:
HelloWorldProducer(const HelloWorldProducer&);
HelloWorldProducer& operator=(const HelloWorldProducer&);
public:
HelloWorldProducer(const std::string& brokerURI, int numMessages, bool
useTopic = false, bool sessionTransacted = false,
long waitMillis = 3000) :
latch(1),
doneLatch(numMessages),
connection(NULL),
session(NULL),
destination(NULL),
producer(NULL),
numMessages(numMessages),
useTopic(useTopic),
sessionTransacted(sessionTransacted),
brokerURI(brokerURI),
bReciveMessage(false),
waitMillis(waitMillis)
{ }
virtual ~HelloWorldProducer() {
cleanup();
}
void close() {
this->cleanup();
}
void waitUntilReady() {
latch.await();
}
virtual void run() {
try {
// Create a ConnectionFactory
auto_ptr<ConnectionFactory> connectionFactory(
ConnectionFactory::createCMSConnectionFactory(brokerURI));
// Create a Connection
connection = connectionFactory->createConnection();
connection->start();
// Create a Session
if (this->sessionTransacted) {
session =
connection->createSession(Session::SESSION_TRANSACTED);
}
else {
session =
connection->createSession(Session::AUTO_ACKNOWLEDGE);
}
session = connection->createSession();
// Create the destination (Topic or Queue)
if (useTopic) {
destination = session->createTopic(QUEUE_NAME);
}
else {
destination = session->createQueue(QUEUE_NAME);
}
// 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 =
Long::toString(Thread::currentThread()->getId());
// Create a messages
string text = (string) "Hello world! from thread " +
threadIdStr;
for (int ix = 0; ix < numMessages; ++ix) {
std::auto_ptr<TextMessage>
message(session->createTextMessage(text));
//????...
std::auto_ptr<Destination>
tempDest(session->createTemporaryQueue());
//cms::Destination
tempDest=session->createTemporaryTopic() ;
MessageConsumer * responseConsumer =
session->createConsumer(tempDest.get());
responseConsumer->setMessageListener(this);//??...
message->setCMSReplyTo(tempDest.get());
Random random;
char buffer[NAME_BYTE_LEN] = { 0 };
random.nextBytes((unsigned char *)buffer,
NAME_BYTE_LEN);
string correlationId = "";
for (int i = 0; i < NAME_BYTE_LEN; ++i)
{
char ch[NAME_BYTE_LEN * 2] = { 0 };
sprintf(ch, "%02X", (unsigned
char)buffer[i]);
string str(ch);
correlationId += str;
}
message->setCMSCorrelationID(correlationId);
message->setIntProperty("Integer", ix);
printf("Producer Sent message #%d from thread
%s\n", ix + 1,
threadIdStr.c_str());
producer->send(message.get());
// Indicate we are ready for messages.
latch.countDown();
// Wait while asynchronous messages come in.
doneLatch.await(waitMillis);
}
}
catch (CMSException& e) {
printf("Producer run() CMSException \n");
// Indicate we are ready for messages.
latch.countDown();
e.printStackTrace();
}
}
// Called from the Producer 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);
//ActiveMQMessageTransformation
//std::auto_ptr<TextMessage>
responsemessage(session->createTextMessage());
//responsemessage->setCMSCorrelationID(textMessage->getCMSCorrelationID());
//responsemessage->getCMSReplyTo()
string text = "";
if (textMessage != NULL) {
text = textMessage->getText();
}
else {
text = "NOT A TEXTMESSAGE!";
}
printf("Producer Message #%d Received: %s\n", count,
text.c_str());
//producer.send
}
catch (CMSException& e) {
printf("Producer onMessage() CMSException \n");
e.printStackTrace();
}
// Commit all messages.
if (this->sessionTransacted) {
session->commit();
}
// No matter what, tag the count down latch until done.
doneLatch.countDown();
}
// 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("Producer onException() CMS Exception occurred.
Shutting down
client. \n");
ex.printStackTrace();
exit(1);
}
private:
void cleanup() {
if (connection != NULL) {
try {
connection->close();
}
catch (cms::CMSException& ex) {
ex.printStackTrace();
}
}
// Destroy resources.
try {
delete destination;
destination = NULL;
delete producer;
producer = NULL;
delete session;
session = NULL;
delete connection;
connection = NULL;
}
catch (CMSException& e) {
e.printStackTrace();
}
}
};
class HelloWorldConsumer : public ExceptionListener,
public MessageListener,
public Runnable {
private:
CountDownLatch latch;
CountDownLatch doneLatch;
Connection* connection;
Session* session;
Destination* destination;
MessageConsumer* consumer;
MessageProducer *producer;
long waitMillis;
bool useTopic;
bool sessionTransacted;
std::string brokerURI;
private:
HelloWorldConsumer(const HelloWorldConsumer&);
HelloWorldConsumer& operator=(const HelloWorldConsumer&);
public:
HelloWorldConsumer(const std::string& brokerURI, int numMessages, bool
useTopic = false, bool sessionTransacted = false, int waitMillis = 30000) :
latch(1),
doneLatch(numMessages),
connection(NULL),
session(NULL),
destination(NULL),
consumer(NULL),
producer(NULL),
waitMillis(waitMillis),
useTopic(useTopic),
sessionTransacted(sessionTransacted),
brokerURI(brokerURI) {
}
virtual ~HelloWorldConsumer() {
cleanup();
}
void close() {
this->cleanup();
}
void waitUntilReady() {
latch.await();
}
virtual void run() {
try {
// Create a ConnectionFactory
auto_ptr<ConnectionFactory> connectionFactory(
ConnectionFactory::createCMSConnectionFactory(brokerURI));
// Create a Connection
connection = connectionFactory->createConnection();
connection->start();
connection->setExceptionListener(this);
// Create a Session
if (this->sessionTransacted == true) {
session =
connection->createSession(Session::SESSION_TRANSACTED);
}
else {
session =
connection->createSession(Session::AUTO_ACKNOWLEDGE);
}
// Create the destination (Topic or Queue)
if (useTopic) {
destination = session->createTopic(QUEUE_NAME);
}
else {
destination = session->createQueue(QUEUE_NAME);
}
producer = session->createProducer();
producer->setDeliveryMode(DeliveryMode::NON_PERSISTENT);
// Create a MessageConsumer from the Session to the
Topic or Queue
consumer = session->createConsumer(destination);
consumer->setMessageListener(this);
std::cout.flush();
std::cerr.flush();
// Indicate we are ready for messages.
latch.countDown();
// Wait while asynchronous messages come in.
doneLatch.await();
}
catch (CMSException& e) {
printf("Consumer onException() CMS Exception occurred.
Shutting down
client. \n");
// Indicate we are ready for messages.
latch.countDown();
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++;
// Create the Thread Id String
string threadIdStr =
Long::toString(Thread::currentThread()->getId());
static bool bPrintf = true;
if (bPrintf)
{
bPrintf = false;
printf("consumer Message threadid: %s\n",
threadIdStr.c_str());
}
string strReply = "consumer return xxx,ThreadID=" +
threadIdStr;
const TextMessage* textMessage = dynamic_cast<const
TextMessage*>
(message);
if (NULL == textMessage)
{
printf("NULL==textMessage %s",
message->getCMSType().c_str());
//const cms::MapMessage* mapMsg =
dynamic_cast<const
cms::MapMessage*>(message);
//if(mapMsg)
//{
//
// std::vector<std::string> elements =
mapMsg->getMapNames();
// std::vector<std::string>::iterator iter =
elements.begin();
// for(; iter != elements.end() ; ++iter)
// {
// std::string key = *iter;
// cms::Message::ValueType elementType =
mapMsg->getValueType(key);
// string strxxx;
// int cc=0;
// switch(elementType) {
// case cms::Message::BOOLEAN_TYPE:
// //msg->setBoolean(key,
mapMsg->getBoolean(key));
// break;
// case cms::Message::BYTE_TYPE:
// //msg->setByte(key,
mapMsg->getByte(key));
// break;
// case cms::Message::BYTE_ARRAY_TYPE:
// //msg->setBytes(key,
mapMsg->getBytes(key));
// break;
// case cms::Message::CHAR_TYPE:
// //msg->setChar(key,
mapMsg->getChar(key));
// break;
// case cms::Message::SHORT_TYPE:
// //msg->setShort(key,
mapMsg->getShort(key));
// break;
// case cms::Message::INTEGER_TYPE:
// //msg->setInt(key,
mapMsg->getInt(key));
// break;
// case cms::Message::LONG_TYPE:
// //msg->setLong(key,
mapMsg->getLong(key));
// break;
// case cms::Message::FLOAT_TYPE:
// //msg->setFloat(key,
mapMsg->getFloat(key));
// break;
// case cms::Message::DOUBLE_TYPE:
// //msg->setDouble(key,
mapMsg->getDouble(key));
// break;
// case cms::Message::STRING_TYPE:
// //msg->setString(key,
mapMsg->getString(key));
// strxxx=mapMsg->getString(key);
// cc=1;
// break;
// default:
// break;
// }
// }
//}
return;
}
std::auto_ptr<TextMessage>
responsemessage(session->createTextMessage(strReply));
responsemessage->setCMSCorrelationID(textMessage->getCMSCorrelationID());
string text = "";
if (textMessage != NULL) {
text = textMessage->getText();
}
else {
text = "NOT A TEXTMESSAGE!";
}
int nProPerty = textMessage->getIntProperty("Integer");
printf("consumer Message #%d Received:
%s,nProPerty[%d]\n", count,
text.c_str(), nProPerty);
const cms::Destination* destSend =
textMessage->getCMSReplyTo();
if (destSend)
{
this->producer->send(destSend,
responsemessage.get());
printf("consumer Message #%d send: %s\n",
count, strReply.c_str());
}
}
catch (CMSException& e) {
printf("Consumer onMessage() CMS Exception occurred.
Shutting down
client. \n");
e.printStackTrace();
}
// Commit all messages.
if (this->sessionTransacted) {
session->commit();
}
// No matter what, tag the count down latch until done.
//doneLatch.countDown();
}
// 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("Consumer onException() CMS Exception occurred.
Shutting down
client. \n");
//printf("CMS Exception occurred. Shutting down client.\n");
ex.printStackTrace();
exit(1);
}
private:
void cleanup() {
if (connection != NULL) {
try {
connection->close();
}
catch (cms::CMSException& ex) {
ex.printStackTrace();
}
}
// Destroy resources.
try {
delete destination;
destination = NULL;
delete consumer;
consumer = NULL;
delete session;
session = NULL;
delete connection;
connection = NULL;
}
catch (CMSException& e) {
e.printStackTrace();
}
}
};
int main(int argc AMQCPP_UNUSED, char* argv[] AMQCPP_UNUSED) {
//if(argc<2)
//{
// printf("argc<2\r\n");
// return 0;
//}
activemq::library::ActiveMQCPP::initializeLibrary();
{
std::cout <<
"=====================================================\n";
std::cout << "Starting the example:" << std::endl;
std::cout <<
"-----------------------------------------------------\n";
// Set the URI to point to the IP Address of your broker.
// add any optional params to the url to enable things like
// tightMarshalling or tcp logging etc. See the CMS web site
for
// a full list of configuration options.
//
// http://activemq.apache.org/cms/
//
// Wire Format Options:
// =========================
// Use either stomp or openwire, the default ports are
different for each
//
// Examples:
// tcp://127.0.0.1:61616 default to
openwire
// tcp://127.0.0.1:61616?wireFormat=openwire same as above
// tcp://127.0.0.1:61613?wireFormat=stomp use stomp
instead
//
// SSL:
// =========================
// To use SSL you need to specify the location of the trusted
Root CA or
the
// certificate for the broker you want to connect to. Using
the Root CA
allows
// you to use failover with multiple servers all using
certificates signed
by
// the trusted root. If using client authentication you also
need to
specify
// the location of the client Certificate.
//
// System::setProperty( "decaf.net.ssl.keyStore",
"<path>/client.pem"
);
// System::setProperty( "decaf.net.ssl.keyStorePassword",
"password"
);
// System::setProperty( "decaf.net.ssl.trustStore",
"<path>/rootCA.pem" );
//
// The you just specify the ssl transport in the URI, for
example:
//
// ssl://localhost:61617
//
std::string brokerURI =
"tcp://127.0.0.1:61616?jms.watchTopicAdvisories=false";
//============================================================
// 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 = false;
bool sessionTransacted = true;
int numMessages = 1;
bool useConsumer = true;
bool useProducer = true;
//int nSet=atoi(argv[1]);
//if(1==nSet)
//{
//#define USE_COMSUMER
//}
//else
//{
//#define USE_PRODUCER
//
//}
long long startTime = System::currentTimeMillis();
#ifdef USE_PRODUCER
printf("?? USE_PRODUCER \r\n");
int numProducerMessages = 30;
int nThreadNumber = 10;
vector<HelloWorldProducer *> vHelloWorldProducer;
for (int i = 0; i < nThreadNumber; ++i)
{
HelloWorldProducer * producerTemp = new
HelloWorldProducer(brokerURI,
numProducerMessages, useTopics);
vHelloWorldProducer.push_back(producerTemp);
}
#endif
#ifdef USE_COMSUMER
printf("?? USE_COMSUMER \r\n");
HelloWorldConsumer consumer(brokerURI, numMessages, useTopics,
sessionTransacted);
// Start the consumer thread.
Thread consumerThread(&consumer);
consumerThread.start();
// Wait for the consumer to indicate that its ready to go.
consumer.waitUntilReady();
#endif
#ifdef USE_PRODUCER
// Start the producer thread.
vector<Thread *> vThread;
for (int i = 0; i < nThreadNumber; ++i)
{
HelloWorldProducer & ProducerTemp =
*vHelloWorldProducer[i];
Thread * threadTemp = new Thread(&ProducerTemp);
vThread.push_back(threadTemp);
threadTemp->start();
ProducerTemp.waitUntilReady();
}
for (size_t i = 0; i < vThread.size(); ++i)
{
Thread * threadTemp = vThread[i];
//threadTemp->join();
}
while (1)
{
Thread::sleep(10);
}
//Thread producerThread1(&producer1);
//producerThread1.start();
//producer1.waitUntilReady();
//Thread producerThread2(&producer2);
//producerThread2.start();
//producer2.waitUntilReady();
//Thread producerThread3(&producer3);
//producerThread3.start();
//producer3.waitUntilReady();
#endif
#ifdef USE_PRODUCER
// Wait for the threads to complete.
//producerThread1.join();
//producerThread2.join();
//producerThread3.join();
#endif
#ifdef USE_COMSUMER
consumerThread.join();
#endif
long long endTime = System::currentTimeMillis();
double totalTime = (double)(endTime - startTime) / 1000.0;
#ifdef USE_PRODUCER
//producer1.close();
//producer2.close();
//producer3.close();
for (size_t i = 0; i < vHelloWorldProducer.size(); ++i)
{
HelloWorldProducer * ProducerTemp =
vHelloWorldProducer[i];
ProducerTemp->close();
if (ProducerTemp)
{
delete ProducerTemp;
ProducerTemp = NULL;
}
}
#endif
#ifdef USE_COMSUMER
consumer.close();
#endif
std::cout << "Time to completion = " << totalTime << "
seconds." <<
std::endl;
std::cout <<
"-----------------------------------------------------\n";
std::cout << "Finished with the example." << std::endl;
std::cout <<
"=====================================================\n";
}
activemq::library::ActiveMQCPP::shutdownLibrary();
return 0;
}
When I run the above example producer and consumer following happens:
1. Producer is able to put the message on the queue.
2. Consumer is able to retrieve the message from the queues.
3. When consumer tries to send response back using replyTo desitnation, the
send fails with error message listed above.
On the broker and consumer/producer I have advisorySupport turned OFF. When
I turn them on this work fine.
What I would like to know:
1. How can I make the error go away but still have advisorySupport off.
Thanks a lot for your help.
--
Sent from: http://activemq.2283324.n4.nabble.com/ActiveMQ-Dev-f2368404.html
