activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dpatel <dhpa...@factset.com>
Subject ActiveMQCPP--Cannot publish to a deleted destinatination: temp-queue:
Date Sun, 16 Sep 2018 07:06:37 GMT
	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. 

P.S-- Attaching my activemq.xml.



--
Sent from: http://activemq.2283324.n4.nabble.com/ActiveMQ-User-f2341805.html

Mime
View raw message