| /* |
| * 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. |
| */ |
| |
| #include "OpenWireMessageListenerRedeliveryTest.h" |
| |
| #include <cms/ConnectionFactory.h> |
| #include <cms/Connection.h> |
| #include <cms/Session.h> |
| |
| #include <activemq/core/policies/DefaultRedeliveryPolicy.h> |
| #include <activemq/core/ActiveMQConnectionFactory.h> |
| #include <activemq/core/ActiveMQConnection.h> |
| |
| #include <decaf/lang/Pointer.h> |
| #include <decaf/lang/Thread.h> |
| #include <decaf/lang/Runnable.h> |
| #include <decaf/util/ArrayList.h> |
| #include <decaf/util/concurrent/TimeUnit.h> |
| #include <decaf/util/concurrent/CountDownLatch.h> |
| #include <decaf/util/concurrent/atomic/AtomicInteger.h> |
| |
| #include <memory> |
| |
| using namespace cms; |
| using namespace std; |
| using namespace decaf; |
| using namespace decaf::lang; |
| using namespace decaf::lang::exceptions; |
| using namespace decaf::util; |
| using namespace decaf::util::concurrent; |
| using namespace decaf::util::concurrent::atomic; |
| using namespace activemq; |
| using namespace activemq::core; |
| using namespace activemq::test; |
| using namespace activemq::test::openwire; |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| namespace { |
| |
| static std::string DLQ_DELIVERY_FAILURE_CAUSE_PROPERTY = "dlqDeliveryFailureCause"; |
| |
| cms::Connection* createConnection(const std::string& brokerUri) { |
| |
| ActiveMQConnectionFactory factory(brokerUri); |
| |
| factory.getRedeliveryPolicy()->setInitialRedeliveryDelay(0); |
| factory.getRedeliveryPolicy()->setRedeliveryDelay(1000); |
| factory.getRedeliveryPolicy()->setMaximumRedeliveries(3); |
| factory.getRedeliveryPolicy()->setBackOffMultiplier((short) 2); |
| factory.getRedeliveryPolicy()->setUseExponentialBackOff(true); |
| |
| return factory.createConnection(); |
| } |
| |
| class TestMessageListener : public cms::MessageListener { |
| private: |
| |
| int counter; |
| cms::Session* session; |
| |
| public: |
| |
| TestMessageListener(cms::Session* session) : counter(0), session(session) {} |
| |
| int getCounter() { |
| return counter; |
| } |
| |
| virtual void onMessage(const cms::Message* message) { |
| |
| try { |
| counter++; |
| |
| if (counter <= 4) { |
| session->rollback(); |
| } else { |
| message->acknowledge(); |
| session->commit(); |
| } |
| } catch (CMSException& e) { |
| } |
| } |
| }; |
| |
| class ExceptionMessageListener : public cms::MessageListener { |
| private: |
| |
| CountDownLatch doneLatch; |
| ArrayList<std::string> received; |
| std::string testName; |
| int maxDeliveries; |
| int count; |
| |
| public: |
| |
| ExceptionMessageListener(const std::string& testName, int messageCount, int maxDeliveries) : |
| doneLatch(messageCount), |
| received(), |
| testName(testName), |
| maxDeliveries(maxDeliveries), |
| count(0) { |
| } |
| |
| int getCount() { |
| return count; |
| } |
| |
| ArrayList<std::string>& getReceived() { |
| return received; |
| } |
| |
| bool await(long long timeout, const TimeUnit& unit) { |
| return doneLatch.await(timeout, unit); |
| } |
| |
| virtual void onMessage(const cms::Message* message) { |
| try { |
| const TextMessage* textMessage = dynamic_cast<const cms::TextMessage*>(message); |
| received.add(textMessage->getText()); |
| } catch (cms::CMSException& e) { |
| e.printStackTrace(); |
| } |
| |
| if (++count < maxDeliveries) { |
| throw decaf::lang::exceptions::RuntimeException( |
| __FILE__, __LINE__, testName.append(" forced a redelivery").c_str()); |
| } |
| |
| // new blood |
| count = 0; |
| doneLatch.countDown(); |
| } |
| }; |
| |
| class TrackingMessageListener : public cms::MessageListener { |
| private: |
| |
| CountDownLatch doneLatch; |
| ArrayList< Pointer<cms::Message> > received; |
| std::string testName; |
| int count; |
| |
| public: |
| |
| TrackingMessageListener(const std::string& testName) : |
| doneLatch(1), |
| received(), |
| testName(testName), |
| count(0) { |
| } |
| |
| TrackingMessageListener(const std::string& testName, int expected) : |
| doneLatch(expected), |
| received(), |
| testName(testName), |
| count(0) { |
| } |
| |
| int getCount() { |
| return received.size(); |
| } |
| |
| ArrayList< Pointer<cms::Message> >& getReceived() { |
| return received; |
| } |
| |
| bool await(long long timeout, const TimeUnit& unit) { |
| return doneLatch.await(timeout, unit); |
| } |
| |
| virtual void onMessage(const cms::Message* message) { |
| try { |
| Pointer<cms::Message> copy(message->clone()); |
| received.add(copy); |
| doneLatch.countDown(); |
| } catch (cms::CMSException& e) { |
| e.printStackTrace(); |
| } |
| } |
| }; |
| |
| class FailingMessageListener : public cms::MessageListener { |
| private: |
| |
| CountDownLatch doneLatch; |
| cms::Session* session; |
| std::string testName; |
| |
| public: |
| |
| FailingMessageListener(cms::Session* session, const std::string& testName, int expected) : |
| doneLatch(expected), |
| session(session), |
| testName(testName) { |
| } |
| |
| bool await(long long timeout, const TimeUnit& unit) { |
| return doneLatch.await(timeout, unit); |
| } |
| |
| virtual void onMessage(const cms::Message* message) { |
| try { |
| doneLatch.countDown(); |
| if (session->isTransacted()) { |
| session->rollback(); |
| } |
| } catch (cms::CMSException& e) { |
| e.printStackTrace(); |
| } |
| |
| throw decaf::lang::exceptions::RuntimeException( |
| __FILE__, __LINE__, (testName + " forced a redelivery").c_str()); |
| } |
| }; |
| |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| OpenWireMessageListenerRedeliveryTest::OpenWireMessageListenerRedeliveryTest() { |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| OpenWireMessageListenerRedeliveryTest::~OpenWireMessageListenerRedeliveryTest() { |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void OpenWireMessageListenerRedeliveryTest::testQueueRollbackConsumerListener() { |
| |
| std::auto_ptr<cms::Connection> connection(createConnection(getBrokerURL())); |
| connection->start(); |
| |
| std::auto_ptr<Session> session(connection->createSession(cms::Session::SESSION_TRANSACTED)); |
| std::auto_ptr<Queue> queue(session->createQueue("testQueueRollbackConsumerListener")); |
| std::auto_ptr<MessageProducer> producer(session->createProducer(queue.get())); |
| std::auto_ptr<Message> message(session->createTextMessage("Hello")); |
| |
| ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>(connection.get()); |
| amqConnection->destroyDestination(queue.get()); |
| |
| producer->setDeliveryMode(cms::DeliveryMode::PERSISTENT); |
| producer->send(message.get()); |
| session->commit(); |
| |
| std::auto_ptr<MessageConsumer> consumer(session->createConsumer(queue.get())); |
| |
| TestMessageListener listener(session.get()); |
| consumer->setMessageListener(&listener); |
| |
| TimeUnit::MILLISECONDS.sleep(500); |
| |
| // first try.. should get 2 since there is no delay on the first redelivery. |
| CPPUNIT_ASSERT_EQUAL(2, listener.getCounter()); |
| |
| TimeUnit::MILLISECONDS.sleep(1000); |
| |
| // 2nd redeliver (redelivery after 1 sec) |
| CPPUNIT_ASSERT_EQUAL(3, listener.getCounter()); |
| |
| TimeUnit::MILLISECONDS.sleep(2000); |
| |
| // 3rd redeliver (redelivery after 2 seconds) - it should give up after that |
| CPPUNIT_ASSERT_EQUAL(4, listener.getCounter()); |
| |
| // create new message |
| std::auto_ptr<Message> secondMessage(session->createTextMessage("Hello 2")); |
| producer->send(secondMessage.get()); |
| session->commit(); |
| |
| TimeUnit::MILLISECONDS.sleep(500); |
| |
| // it should be committed, so no redelivery |
| CPPUNIT_ASSERT_EQUAL(5, listener.getCounter()); |
| |
| TimeUnit::MILLISECONDS.sleep(1500); |
| |
| // no redelivery, counter should still be 4 |
| CPPUNIT_ASSERT_EQUAL(5, listener.getCounter()); |
| |
| connection->close(); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void OpenWireMessageListenerRedeliveryTest::testQueueSessionListenerExceptionRetry() { |
| |
| std::auto_ptr<cms::Connection> connection(createConnection(getBrokerURL())); |
| connection->start(); |
| |
| std::auto_ptr<Session> session(connection->createSession(cms::Session::AUTO_ACKNOWLEDGE)); |
| std::auto_ptr<Queue> queue(session->createQueue("testQueueSessionListenerExceptionRetry")); |
| std::auto_ptr<MessageProducer> producer(session->createProducer(queue.get())); |
| std::auto_ptr<Message> message1(session->createTextMessage("1")); |
| std::auto_ptr<Message> message2(session->createTextMessage("2")); |
| |
| ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>(connection.get()); |
| amqConnection->destroyDestination(queue.get()); |
| |
| producer->setDeliveryMode(cms::DeliveryMode::PERSISTENT); |
| producer->send(message1.get()); |
| producer->send(message2.get()); |
| |
| std::auto_ptr<MessageConsumer> consumer(session->createConsumer(queue.get())); |
| |
| int maxDeliveries = amqConnection->getRedeliveryPolicy()->getMaximumRedeliveries(); |
| |
| ExceptionMessageListener listener("testQueueSessionListenerExceptionRetry", 2, maxDeliveries); |
| consumer->setMessageListener(&listener); |
| |
| CPPUNIT_ASSERT_MESSAGE("got message before retry expiry", listener.await(30, TimeUnit::SECONDS)); |
| |
| for (int i = 0; i < maxDeliveries; i++) { |
| CPPUNIT_ASSERT_EQUAL_MESSAGE("got first redelivered: " + Integer::toString(i), |
| std::string("1"), listener.getReceived().get(i)); |
| } |
| |
| for (int i = maxDeliveries; i < maxDeliveries * 2; i++) { |
| CPPUNIT_ASSERT_EQUAL_MESSAGE("got first redelivered: " + Integer::toString(i), |
| std::string("2"), listener.getReceived().get(i)); |
| } |
| |
| connection->close(); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void OpenWireMessageListenerRedeliveryTest::testQueueSessionListenerExceptionDlq() { |
| |
| const std::string TEST_NAME = "testQueueSessionListenerExceptionDlq"; |
| |
| std::auto_ptr<cms::Connection> connection(createConnection(getBrokerURL())); |
| connection->start(); |
| |
| std::auto_ptr<Session> session(connection->createSession(cms::Session::AUTO_ACKNOWLEDGE)); |
| std::auto_ptr<Queue> queue(session->createQueue(TEST_NAME)); |
| std::auto_ptr<Queue> dlq(session->createQueue("ActiveMQ.DLQ")); |
| |
| ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>(connection.get()); |
| amqConnection->destroyDestination(queue.get()); |
| amqConnection->destroyDestination(dlq.get()); |
| |
| std::auto_ptr<MessageProducer> producer(session->createProducer(queue.get())); |
| std::auto_ptr<Message> message(session->createTextMessage("1")); |
| producer->setDeliveryMode(cms::DeliveryMode::PERSISTENT); |
| producer->send(message.get()); |
| |
| // Track messages going to DLQ |
| TrackingMessageListener dlqListener(TEST_NAME); |
| std::auto_ptr<MessageConsumer> dlqConsumer(session->createConsumer(dlq.get())); |
| dlqConsumer->setMessageListener(&dlqListener); |
| |
| // Receive and throw |
| int maxDeliveries = amqConnection->getRedeliveryPolicy()->getMaximumRedeliveries(); |
| FailingMessageListener listener(session.get(), TEST_NAME, maxDeliveries); |
| std::auto_ptr<MessageConsumer> consumer(session->createConsumer(queue.get())); |
| consumer->setMessageListener(&listener); |
| |
| CPPUNIT_ASSERT_MESSAGE("got message before retry expiry", listener.await(20, TimeUnit::SECONDS)); |
| |
| // check DLQ |
| CPPUNIT_ASSERT_MESSAGE("got dlq message", dlqListener.await(30, TimeUnit::SECONDS)); |
| |
| // check DLQ message cause is captured |
| Pointer<cms::Message> dlqMessage = dlqListener.getReceived().get(0); |
| CPPUNIT_ASSERT_MESSAGE("dlq message captured", dlqMessage != NULL); |
| String cause = dlqMessage->getStringProperty(DLQ_DELIVERY_FAILURE_CAUSE_PROPERTY); |
| |
| CPPUNIT_ASSERT_MESSAGE("cause 'cause' exception is remembered", cause.contains("Exception")); |
| CPPUNIT_ASSERT_MESSAGE("is correct exception", cause.contains(TEST_NAME)); |
| CPPUNIT_ASSERT_MESSAGE("cause exception is remembered", cause.contains("JMSException")); |
| CPPUNIT_ASSERT_MESSAGE("cause policy is remembered", cause.contains("RedeliveryPolicy")); |
| |
| connection->close(); |
| } |
| |
| //////////////////////////////////////////////////////////////////////////////// |
| void OpenWireMessageListenerRedeliveryTest::testTransactedQueueSessionListenerExceptionDlq() { |
| |
| const std::string TEST_NAME = "testTransactedQueueSessionListenerExceptionDlq"; |
| |
| std::auto_ptr<cms::Connection> connection(createConnection(getBrokerURL())); |
| connection->start(); |
| |
| std::auto_ptr<Session> session(connection->createSession(cms::Session::SESSION_TRANSACTED)); |
| std::auto_ptr<Queue> queue(session->createQueue(TEST_NAME)); |
| std::auto_ptr<Queue> dlq(session->createQueue("ActiveMQ.DLQ")); |
| |
| ActiveMQConnection* amqConnection = dynamic_cast<ActiveMQConnection*>(connection.get()); |
| amqConnection->destroyDestination(queue.get()); |
| amqConnection->destroyDestination(dlq.get()); |
| |
| std::auto_ptr<MessageProducer> producer(session->createProducer(queue.get())); |
| std::auto_ptr<Message> message(session->createTextMessage("1")); |
| producer->setDeliveryMode(cms::DeliveryMode::PERSISTENT); |
| producer->send(message.get()); |
| session->commit(); |
| |
| // Track messages going to DLQ |
| TrackingMessageListener dlqListener(TEST_NAME); |
| std::auto_ptr<MessageConsumer> dlqConsumer(session->createConsumer(dlq.get())); |
| dlqConsumer->setMessageListener(&dlqListener); |
| |
| // Receive and throw |
| int maxDeliveries = amqConnection->getRedeliveryPolicy()->getMaximumRedeliveries(); |
| FailingMessageListener listener(session.get(), TEST_NAME, maxDeliveries); |
| std::auto_ptr<MessageConsumer> consumer(session->createConsumer(queue.get())); |
| consumer->setMessageListener(&listener); |
| |
| CPPUNIT_ASSERT_MESSAGE("got message before retry expiry", listener.await(20, TimeUnit::SECONDS)); |
| |
| // check DLQ |
| CPPUNIT_ASSERT_MESSAGE("got dlq message", dlqListener.await(30, TimeUnit::SECONDS)); |
| |
| // check DLQ message cause is captured |
| Pointer<cms::Message> dlqMessage = dlqListener.getReceived().get(0); |
| CPPUNIT_ASSERT_MESSAGE("dlq message captured", dlqMessage != NULL); |
| String cause = dlqMessage->getStringProperty(DLQ_DELIVERY_FAILURE_CAUSE_PROPERTY); |
| |
| CPPUNIT_ASSERT_MESSAGE("cause 'cause' exception is remembered", cause.contains("Exception")); |
| CPPUNIT_ASSERT_MESSAGE("is correct exception", cause.contains(TEST_NAME)); |
| CPPUNIT_ASSERT_MESSAGE("cause exception is remembered", cause.contains("JMSException")); |
| CPPUNIT_ASSERT_MESSAGE("cause policy is remembered", cause.contains("RedeliveryPolicy")); |
| |
| connection->close(); |
| } |
