activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r1498748 - in /activemq/activemq-cpp/branches/3.7.x: ./ activemq-cpp/src/main/activemq/core/kernels/ activemq-cpp/src/test-integration/activemq/test/ activemq-cpp/src/test-integration/activemq/test/openwire/ activemq-cpp/src/test/activemq/c...
Date Tue, 02 Jul 2013 00:38:03 GMT
Author: tabish
Date: Tue Jul  2 00:38:02 2013
New Revision: 1498748

URL: http://svn.apache.org/r1498748
Log:
fix for: https://issues.apache.org/jira/browse/AMQCPP-494

Modified:
    activemq/activemq-cpp/branches/3.7.x/   (props changed)
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.cpp
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.h
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.cpp
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.h
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/openwire/OpenwireTransactionTest.h
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.cpp
    activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.h

Propchange: activemq/activemq-cpp/branches/3.7.x/
------------------------------------------------------------------------------
  Merged /activemq/activemq-cpp/trunk:r1494715-1498566,1498568-1498611,1498613-1498746

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.cpp?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.cpp
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQConsumerKernel.cpp
Tue Jul  2 00:38:02 2013
@@ -97,6 +97,7 @@ namespace kernels {
         decaf::util::concurrent::Mutex listenerMutex;
         AtomicBoolean deliveringAcks;
         AtomicBoolean started;
+        AtomicBoolean closeSyncRegistered;
         Pointer<MessageDispatchChannel> unconsumedMessages;
         decaf::util::LinkedList< decaf::lang::Pointer<commands::MessageDispatch>
> dispatchedMessages;
         long long lastDeliveredSequenceId;
@@ -133,6 +134,7 @@ namespace kernels {
                                          listenerMutex(),
                                          deliveringAcks(),
                                          started(),
+                                         closeSyncRegistered(),
                                          unconsumedMessages(),
                                          dispatchedMessages(),
                                          lastDeliveredSequenceId(0),
@@ -340,7 +342,7 @@ namespace {
     class TransactionSynhcronization : public Synchronization {
     private:
 
-        ActiveMQConsumerKernel* consumer;
+        Pointer<ActiveMQConsumerKernel> consumer;
         ActiveMQConsumerKernelConfig* impl;
 
     private:
@@ -350,7 +352,7 @@ namespace {
 
     public:
 
-        TransactionSynhcronization(ActiveMQConsumerKernel* consumer, ActiveMQConsumerKernelConfig*
impl) :
+        TransactionSynhcronization(Pointer<ActiveMQConsumerKernel> consumer, ActiveMQConsumerKernelConfig*
impl) :
             Synchronization(), consumer(consumer), impl(impl) {
 
             if (consumer == NULL) {
@@ -377,11 +379,13 @@ namespace {
         virtual void afterCommit() {
             consumer->commit();
             consumer->setSynchronizationRegistered(false);
+            consumer.reset(NULL);
         }
 
         virtual void afterRollback() {
             consumer->rollback();
             consumer->setSynchronizationRegistered(false);
+            consumer.reset(NULL);
         }
     };
 
@@ -393,7 +397,7 @@ namespace {
     class CloseSynhcronization : public Synchronization {
     private:
 
-        ActiveMQConsumerKernel* consumer;
+        Pointer<ActiveMQConsumerKernel> consumer;
 
     private:
 
@@ -402,7 +406,7 @@ namespace {
 
     public:
 
-        CloseSynhcronization(ActiveMQConsumerKernel* consumer) : consumer(consumer) {
+        CloseSynhcronization(Pointer<ActiveMQConsumerKernel> consumer) : consumer(consumer)
{
             if (consumer == NULL) {
                 throw NullPointerException(__FILE__, __LINE__, "Synchronization Created with
NULL Consumer.");
             }
@@ -415,10 +419,12 @@ namespace {
 
         virtual void afterCommit() {
             consumer->doClose();
+            consumer.reset(NULL);
         }
 
         virtual void afterRollback() {
             consumer->doClose();
+            consumer.reset(NULL);
         }
     };
 
@@ -787,9 +793,12 @@ void ActiveMQConsumerKernel::close() {
         if (!this->isClosed()) {
 
             if (this->session->getTransactionContext() != NULL &&
-                this->session->getTransactionContext()->isInTransaction()) {
+                this->session->getTransactionContext()->isInTransaction() &&
+                this->internal->closeSyncRegistered.compareAndSet(false, true)) {
 
-                Pointer<Synchronization> sync(new CloseSynhcronization(this));
+                Pointer<ActiveMQConsumerKernel> self =
+                    this->session->lookupConsumerKernel(this->consumerInfo->getConsumerId());
+                Pointer<Synchronization> sync(new CloseSynhcronization(self));
                 this->session->getTransactionContext()->addSynchronization(sync);
             } else {
                 doClose();
@@ -1134,7 +1143,9 @@ void ActiveMQConsumerKernel::registerSyn
     this->session->doStartTransaction();
     if (!this->internal->synchronizationRegistered) {
         this->internal->synchronizationRegistered = true;
-        Pointer<Synchronization> sync(new TransactionSynhcronization(this, this->internal));
+        Pointer<ActiveMQConsumerKernel> self =
+            this->session->lookupConsumerKernel(this->consumerInfo->getConsumerId());
+        Pointer<Synchronization> sync(new TransactionSynhcronization(self, this->internal));
         this->session->getTransactionContext()->addSynchronization(sync);
     }
 }

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.h?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.h
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/main/activemq/core/kernels/ActiveMQSessionKernel.h
Tue Jul  2 00:38:02 2013
@@ -529,7 +529,7 @@ namespace kernels {
         Pointer<ActiveMQProducerKernel> lookupProducerKernel(Pointer<commands::ProducerId>
id);
 
         /**
-         * @returns a Pointer to an ActiveMQProducerKernel using its ProducerId, or NULL.
+         * @returns a Pointer to an ActiveMQConsumerKernel using its ConsumerId, or NULL.
          */
         Pointer<ActiveMQConsumerKernel> lookupConsumerKernel(Pointer<commands::ConsumerId>
id);
 

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.cpp?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.cpp
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.cpp
Tue Jul  2 00:38:02 2013
@@ -17,6 +17,7 @@
 
 #include "TransactionTest.h"
 
+#include <activemq/core/ActiveMQConnectionFactory.h>
 #include <activemq/util/CMSListener.h>
 #include <activemq/exceptions/ActiveMQException.h>
 #include <stdexcept>
@@ -24,11 +25,20 @@
 using namespace std;
 using namespace cms;
 using namespace activemq;
+using namespace activemq::core;
 using namespace activemq::test;
 using namespace activemq::util;
 using namespace activemq::exceptions;
 
 ////////////////////////////////////////////////////////////////////////////////
+TransactionTest::TransactionTest() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
+TransactionTest::~TransactionTest() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
 void TransactionTest::testSendReceiveTransactedBatches() {
 
     try {
@@ -248,3 +258,33 @@ void TransactionTest::testWithTTLSet() {
     AMQ_CATCH_RETHROW( ActiveMQException )
     AMQ_CATCHALL_THROW( ActiveMQException )
 }
+
+////////////////////////////////////////////////////////////////////////////////
+void TransactionTest::testSessionCommitAfterConsumerClosed() {
+
+    ActiveMQConnectionFactory factory(getBrokerURL());
+    auto_ptr<cms::Connection> connection(factory.createConnection());
+
+    {
+        auto_ptr<cms::Session> session(connection->createSession(cms::Session::AUTO_ACKNOWLEDGE));
+        auto_ptr<cms::Queue> queue(session->createQueue("testSessionCommitAfterConsumerClosed"));
+        auto_ptr<cms::MessageProducer> producer(session->createProducer(queue.get()));
+
+        auto_ptr<cms::Message> message(session->createTextMessage("Hello"));
+        producer->send(message.get());
+        producer->close();
+        session->close();
+    }
+
+    auto_ptr<cms::Session> session(connection->createSession(cms::Session::SESSION_TRANSACTED));
+    auto_ptr<cms::Queue> queue(session->createQueue("testSessionCommitAfterConsumerClosed"));
+    auto_ptr<cms::MessageConsumer> consumer(session->createConsumer(queue.get()));
+
+    connection->start();
+
+    auto_ptr<cms::Message> message(consumer->receive(5000));
+    CPPUNIT_ASSERT(message.get() != NULL);
+
+    consumer->close();
+    session->commit();
+}

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.h?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.h
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/TransactionTest.h
Tue Jul  2 00:38:02 2013
@@ -32,8 +32,8 @@ namespace test{
 
     public:
 
-        TransactionTest() {}
-        virtual ~TransactionTest() {}
+        TransactionTest();
+        virtual ~TransactionTest();
 
         virtual void setUp() {
             cmsProvider.reset(
@@ -45,6 +45,7 @@ namespace test{
         void testSendSessionClose();
         void testWithTTLSet();
         void testSendRollbackCommitRollback();
+        void testSessionCommitAfterConsumerClosed();
 
     };
 

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/openwire/OpenwireTransactionTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/openwire/OpenwireTransactionTest.h?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/openwire/OpenwireTransactionTest.h
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test-integration/activemq/test/openwire/OpenwireTransactionTest.h
Tue Jul  2 00:38:02 2013
@@ -31,6 +31,7 @@ namespace openwire{
         CPPUNIT_TEST( testSendRollback );
         CPPUNIT_TEST( testWithTTLSet );
         CPPUNIT_TEST( testSendRollbackCommitRollback );
+        CPPUNIT_TEST( testSessionCommitAfterConsumerClosed );
 //        CPPUNIT_TEST( testSendSessionClose );
         CPPUNIT_TEST_SUITE_END();
 

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.cpp?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.cpp
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.cpp
Tue Jul  2 00:38:02 2013
@@ -42,8 +42,8 @@ using namespace decaf;
 using namespace decaf::lang;
 
 ////////////////////////////////////////////////////////////////////////////////
-namespace activemq{
-namespace core{
+namespace activemq {
+namespace core {
 
     class MyCMSMessageListener : public cms::MessageListener {
     public:
@@ -104,6 +104,14 @@ namespace core{
 }}
 
 ////////////////////////////////////////////////////////////////////////////////
+ActiveMQSessionTest::ActiveMQSessionTest() : connection(), dTransport(), exListener() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
+ActiveMQSessionTest::~ActiveMQSessionTest() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
 void ActiveMQSessionTest::testCreateManyConsumersAndSetListeners() {
 
     MyCMSMessageListener msgListener1;
@@ -541,7 +549,7 @@ void ActiveMQSessionTest::testTransactio
     CPPUNIT_ASSERT_EQUAL(MSG_COUNT, (int)msgListener1.messages.size());
 
     // This is what we are testing, since there was no commit, the session
-    // will rollback the transaction when this are closed.
+    // will rollback the transaction when this is closed.
     // session->commit();
 
     consumer1->close();
@@ -606,77 +614,118 @@ void ActiveMQSessionTest::testExpiration
 }
 
 ////////////////////////////////////////////////////////////////////////////////
-void ActiveMQSessionTest::setUp()
-{
-    try
-    {
+void ActiveMQSessionTest::testTransactionCommitAfterConsumerClosed() {
+
+    static const int MSG_COUNT = 50;
+    MyCMSMessageListener msgListener1;
+
+    CPPUNIT_ASSERT(connection.get() != NULL);
+
+    // Create an Transacted Session
+    std::auto_ptr<cms::Session> session(connection->createSession(cms::Session::SESSION_TRANSACTED));
+
+    // Create a Topic
+    std::auto_ptr<cms::Topic> topic1(session->createTopic("TestTopic1"));
+
+    CPPUNIT_ASSERT(topic1.get() != NULL);
+
+    // Create a consumer
+    std::auto_ptr<ActiveMQConsumer> consumer1(
+        dynamic_cast<ActiveMQConsumer*>( session->createConsumer( topic1.get() )
) );
+
+    CPPUNIT_ASSERT(consumer1.get() != NULL);
+    CPPUNIT_ASSERT(consumer1->getMessageSelector() == "");
+    CPPUNIT_ASSERT(consumer1->receiveNoWait() == NULL);
+    CPPUNIT_ASSERT(consumer1->receive(5) == NULL);
+
+    consumer1->setMessageListener(&msgListener1);
+
+    for (int i = 0; i < MSG_COUNT; ++i) {
+        injectTextMessage("This is a Test 1", *topic1, *(consumer1->getConsumerId()));
+    }
+
+    msgListener1.asyncWaitForMessages(MSG_COUNT);
+    CPPUNIT_ASSERT_EQUAL(MSG_COUNT, (int)msgListener1.messages.size());
+
+    consumer1->close();
+    consumer1.reset();
+    session->commit();
+
+    Pointer<cms::TextMessage> msg1 = msgListener1.messages[0].dynamicCast<cms::TextMessage>();
+    std::string text1 = msg1->getText();
+
+    CPPUNIT_ASSERT(text1 == "This is a Test 1");
+    msgListener1.clear();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void ActiveMQSessionTest::setUp() {
+
+    try {
         ActiveMQConnectionFactory factory("mock://127.0.0.1:12345?wireFormat=openwire");
 
-        connection.reset( dynamic_cast< ActiveMQConnection*>( factory.createConnection()
) );
+        connection.reset(dynamic_cast<ActiveMQConnection*>(factory.createConnection()));
 
         // Get a pointer to the Mock Transport for Message injection.
         dTransport = dynamic_cast<transport::mock::MockTransport*>(
-            connection->getTransport().narrow( typeid( transport::mock::MockTransport
) ) );
-        CPPUNIT_ASSERT( dTransport != NULL );
+            connection->getTransport().narrow(typeid(transport::mock::MockTransport)));
+        CPPUNIT_ASSERT(dTransport != NULL);
 
-        connection->setExceptionListener( &exListener );
+        connection->setExceptionListener(&exListener);
         connection->start();
-    }
-    catch(...)
-    {
+    } catch (...) {
         bool exceptionThrown = false;
-
-        CPPUNIT_ASSERT( exceptionThrown );
+        CPPUNIT_ASSERT(exceptionThrown);
     }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 void ActiveMQSessionTest::tearDown() {
-    connection.reset( NULL );
+    connection.reset(NULL);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
-void ActiveMQSessionTest::injectTextMessage( const std::string message,
-                                             const cms::Destination& destination,
-                                             const commands::ConsumerId& id,
-                                             const long long timeStamp,
-                                             const long long timeToLive )
-{
-    Pointer<ActiveMQTextMessage> msg( new ActiveMQTextMessage() );
-
-    Pointer<ProducerId> producerId( new ProducerId() );
-    producerId->setConnectionId( id.getConnectionId() );
-    producerId->setSessionId( id.getSessionId() );
-    producerId->setValue( 1 );
-
-    Pointer<MessageId> messageId( new MessageId() );
-    messageId->setProducerId( producerId );
-    messageId->setProducerSequenceId( 2 );
+void ActiveMQSessionTest::injectTextMessage(const std::string message,
+                                            const cms::Destination& destination,
+                                            const commands::ConsumerId& id,
+                                            const long long timeStamp,
+                                            const long long timeToLive) {
+
+    Pointer<ActiveMQTextMessage> msg(new ActiveMQTextMessage());
+
+    Pointer<ProducerId> producerId(new ProducerId());
+    producerId->setConnectionId(id.getConnectionId());
+    producerId->setSessionId(id.getSessionId());
+    producerId->setValue(1);
+
+    Pointer<MessageId> messageId(new MessageId());
+    messageId->setProducerId(producerId);
+    messageId->setProducerSequenceId(2);
 
     // Init Message
-    msg->setText( message.c_str() );
-    msg->setCMSDestination( &destination );
-    msg->setCMSMessageID( "Id: 123456" );
-    msg->setMessageId( messageId );
+    msg->setText(message.c_str());
+    msg->setCMSDestination(&destination);
+    msg->setCMSMessageID("Id: 123456");
+    msg->setMessageId(messageId);
 
     long long expiration = 0LL;
 
-    if( timeStamp != 0 ) {
-        msg->setCMSTimestamp( timeStamp );
+    if (timeStamp != 0) {
+        msg->setCMSTimestamp(timeStamp);
 
-        if( timeToLive > 0LL ) {
+        if (timeToLive > 0LL) {
             expiration = timeToLive + timeStamp;
         }
     }
 
-    msg->setCMSExpiration( expiration );
+    msg->setCMSExpiration(expiration);
 
     // Send the Message
-    CPPUNIT_ASSERT( dTransport != NULL );
+    CPPUNIT_ASSERT(dTransport != NULL);
 
-    Pointer<MessageDispatch> dispatch( new MessageDispatch() );
-    dispatch->setMessage( msg );
-    dispatch->setConsumerId( Pointer<ConsumerId>( id.cloneDataStructure() ) );
+    Pointer<MessageDispatch> dispatch(new MessageDispatch());
+    dispatch->setMessage(msg);
+    dispatch->setConsumerId(Pointer<ConsumerId>(id.cloneDataStructure()));
 
-    dTransport->fireCommand( dispatch );
+    dTransport->fireCommand(dispatch);
 }

Modified: activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.h?rev=1498748&r1=1498747&r2=1498748&view=diff
==============================================================================
--- activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.h
(original)
+++ activemq/activemq-cpp/branches/3.7.x/activemq-cpp/src/test/activemq/core/ActiveMQSessionTest.h
Tue Jul  2 00:38:02 2013
@@ -36,13 +36,14 @@
 namespace activemq{
 namespace core{
 
-    class ActiveMQSessionTest : public CppUnit::TestFixture
-    {
+    class ActiveMQSessionTest : public CppUnit::TestFixture {
+
         CPPUNIT_TEST_SUITE( ActiveMQSessionTest );
         CPPUNIT_TEST( testAutoAcking );
         CPPUNIT_TEST( testClientAck );
         CPPUNIT_TEST( testTransactionCommitOneConsumer );
         CPPUNIT_TEST( testTransactionCommitTwoConsumer );
+        CPPUNIT_TEST( testTransactionCommitAfterConsumerClosed );
         CPPUNIT_TEST( testTransactionRollbackOneConsumer );
         CPPUNIT_TEST( testTransactionRollbackTwoConsumer );
         CPPUNIT_TEST( testTransactionCloseWithoutCommit );
@@ -76,20 +77,21 @@ namespace core{
         ActiveMQSessionTest(const ActiveMQSessionTest&);
         ActiveMQSessionTest& operator= (const ActiveMQSessionTest&);
 
-    public:    // CPPUNIT Method Overrides.
+    public:
+
+        virtual void setUp();
+        virtual void tearDown();
 
-        void setUp();
-        void tearDown();
-        void injectTextMessage( const std::string message,
-                                const cms::Destination& destination,
-                                const commands::ConsumerId& id,
-                                const long long timeStamp = -1,
-                                const long long timeToLive = -1 );
+        void injectTextMessage(const std::string message,
+                               const cms::Destination& destination,
+                               const commands::ConsumerId& id,
+                               const long long timeStamp = -1,
+                               const long long timeToLive = -1);
 
     public:
 
-        ActiveMQSessionTest() : connection(), dTransport(), exListener() {}
-        virtual ~ActiveMQSessionTest() {}
+        ActiveMQSessionTest();
+        virtual ~ActiveMQSessionTest();
 
         void testAutoAcking();
         void testClientAck();
@@ -99,6 +101,7 @@ namespace core{
         void testTransactionRollbackOneConsumer();
         void testTransactionRollbackTwoConsumer();
         void testTransactionCloseWithoutCommit();
+        void testTransactionCommitAfterConsumerClosed();
         void testExpiration();
 
     };



Mime
View raw message