| /* |
| * 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. |
| */ |
| package org.apache.qpid.jms.integration; |
| |
| import static org.apache.qpid.jms.provider.amqp.AmqpSupport.ANONYMOUS_RELAY; |
| import static org.hamcrest.Matchers.arrayContaining; |
| import static org.hamcrest.Matchers.equalTo; |
| import static org.hamcrest.Matchers.nullValue; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| |
| import jakarta.jms.CompletionListener; |
| import jakarta.jms.Connection; |
| import jakarta.jms.Destination; |
| import jakarta.jms.Message; |
| import jakarta.jms.MessageProducer; |
| import jakarta.jms.Queue; |
| import jakarta.jms.Session; |
| import jakarta.jms.TemporaryQueue; |
| import jakarta.jms.TemporaryTopic; |
| import jakarta.jms.TextMessage; |
| import jakarta.jms.Topic; |
| |
| import org.apache.qpid.jms.test.QpidJmsTestCase; |
| import org.apache.qpid.jms.test.testpeer.ListDescribedType; |
| import org.apache.qpid.jms.test.testpeer.TestAmqpPeer; |
| import org.apache.qpid.jms.test.testpeer.describedtypes.Accepted; |
| import org.apache.qpid.jms.test.testpeer.describedtypes.TransactionalState; |
| import org.apache.qpid.jms.test.testpeer.matchers.CoordinatorMatcher; |
| import org.apache.qpid.jms.test.testpeer.matchers.TransactionalStateMatcher; |
| import org.apache.qpid.jms.test.testpeer.matchers.sections.MessageAnnotationsSectionMatcher; |
| import org.apache.qpid.jms.test.testpeer.matchers.sections.MessageHeaderSectionMatcher; |
| import org.apache.qpid.jms.test.testpeer.matchers.sections.TransferPayloadCompositeMatcher; |
| import org.apache.qpid.proton.amqp.Binary; |
| import org.apache.qpid.proton.amqp.Symbol; |
| import org.apache.qpid.proton.amqp.transaction.TxnCapability; |
| import org.hamcrest.Matcher; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Test MessageProducers created using various configuration of the presettle options |
| */ |
| public class PresettledProducerIntegrationTest extends QpidJmsTestCase { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(PresettledProducerIntegrationTest.class); |
| |
| private final IntegrationTestFixture testFixture = new IntegrationTestFixture(); |
| |
| private final Symbol[] serverCapabilities = new Symbol[] { ANONYMOUS_RELAY }; |
| |
| //----- Test the jms.presettleAll option ---------------------------------// |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllSendToTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllSendToQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllSendToTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllSendToTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, TemporaryQueue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllAnonymousSendToTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllAnonymousSendToQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllAnonymousSendToTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleAllAnonymousSendToTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, TemporaryQueue.class); |
| } |
| |
| //----- Test the jms.presettleProducers option ---------------------------------// |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, TemporaryQueue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersAnonymousTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersAnonymousQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersAnonymousTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleProducersAnonymousTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, true, true, TemporaryQueue.class); |
| } |
| |
| //----- Test the jms.presettleTopicProducers option ---------------------------------// |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, TemporaryQueue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersAnonymousTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersAnonymousQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, false, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersAnonymousTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTopicProducersAnonymousTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTopicProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, false, TemporaryQueue.class); |
| } |
| |
| //----- Test the jms.presettleQueueProducers option ---------------------------------// |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, true, true, TemporaryQueue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersAnonymousTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, false, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersAnonymousQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersAnonymousTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, false, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleQueueProducersAnonymousTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleQueueProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, true, false, true, TemporaryQueue.class); |
| } |
| |
| //----- Test the jms.presettleTransactedProducers option ---------------------------------// |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, true, false, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, true, false, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersTempTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, true, false, true, true, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersTempQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, true, false, true, true, TemporaryQueue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersTopicNoTX() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersQueueNoTX() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersTempTopicNoTX() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, TemporaryTopic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testJmsPresettlePolicyPresettleTransactedProducersTempQueueNoTX() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleTransactedProducers=true"; |
| doTestProducerWithPresettleOptions(presettleConfig, false, false, false, false, TemporaryQueue.class); |
| } |
| |
| //----- Test Method implementation ---------------------------------------// |
| |
| private void doTestProducerWithPresettleOptions(String uriOptions, boolean transacted, boolean anonymous, boolean senderSettled, boolean transferSettled, Class<? extends Destination> destType) throws Exception { |
| doTestProducerWithPresettleOptions(uriOptions, transacted, anonymous, true, senderSettled, transferSettled, destType); |
| } |
| |
| private void doTestProducerWithPresettleOptions(String uriOptions, boolean transacted, boolean anonymous, boolean relaySupported, boolean senderSettled, boolean transferSettled, Class<? extends Destination> destType) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| Connection connection = testFixture.establishConnecton(testPeer, uriOptions, relaySupported ? serverCapabilities : null, null); |
| testPeer.expectBegin(); |
| |
| Session session = null; |
| Binary txnId = null; |
| |
| if (transacted) { |
| // Expect the session, with an immediate link to the transaction coordinator |
| // using a target with the expected capabilities only. |
| CoordinatorMatcher txCoordinatorMatcher = new CoordinatorMatcher(); |
| txCoordinatorMatcher.withCapabilities(arrayContaining(TxnCapability.LOCAL_TXN)); |
| testPeer.expectSenderAttach(txCoordinatorMatcher, false, false); |
| |
| // First expect an unsettled 'declare' transfer to the txn coordinator, and |
| // reply with a declared disposition state containing the txnId. |
| txnId = new Binary(new byte[]{ (byte) 1, (byte) 2, (byte) 3, (byte) 4}); |
| testPeer.expectDeclare(txnId); |
| |
| session = connection.createSession(true, Session.SESSION_TRANSACTED); |
| } else { |
| session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| } |
| |
| Destination destination = null; |
| if (destType == Queue.class) { |
| destination = session.createQueue("MyQueue"); |
| } else if (destType == Topic.class) { |
| destination = session.createTopic("MyTopis"); |
| } else if (destType == TemporaryQueue.class) { |
| String dynamicAddress = "myTempQueueAddress"; |
| testPeer.expectTempQueueCreationAttach(dynamicAddress); |
| destination = session.createTemporaryQueue(); |
| } else if (destType == TemporaryTopic.class) { |
| String dynamicAddress = "myTempTopicAddress"; |
| testPeer.expectTempTopicCreationAttach(dynamicAddress); |
| destination = session.createTemporaryTopic(); |
| } else { |
| fail("unexpected type"); |
| } |
| |
| if (senderSettled) { |
| testPeer.expectSettledSenderAttach(); |
| } else { |
| testPeer.expectSenderAttach(); |
| } |
| |
| MessageProducer producer = null; |
| if (anonymous) { |
| producer = session.createProducer(null); |
| } else { |
| producer = session.createProducer(destination); |
| } |
| |
| // Create and transfer a new message |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| headersMatcher.withDurable(equalTo(true)); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| Matcher<?> stateMatcher = nullValue(); |
| if (transacted) { |
| stateMatcher = new TransactionalStateMatcher(); |
| ((TransactionalStateMatcher) stateMatcher).withTxnId(equalTo(txnId)); |
| ((TransactionalStateMatcher) stateMatcher).withOutcome(nullValue()); |
| } |
| |
| ListDescribedType responseState = new Accepted(); |
| if (transacted) { |
| TransactionalState txState = new TransactionalState(); |
| txState.setTxnId(txnId); |
| txState.setOutcome(new Accepted()); |
| } |
| |
| if (transferSettled) { |
| testPeer.expectTransfer(messageMatcher, stateMatcher, true, false, responseState, false); |
| } else { |
| testPeer.expectTransfer(messageMatcher, stateMatcher, false, true, responseState, true); |
| } |
| |
| if (anonymous && !relaySupported) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| Message message = session.createTextMessage(); |
| |
| if (anonymous) { |
| producer.send(destination, message); |
| } else { |
| producer.send(message); |
| } |
| |
| if (transacted) { |
| testPeer.expectDischarge(txnId, true); |
| } |
| |
| testPeer.expectClose(); |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| //----- Test the jms.presettleAll with asynchronous completion -----------// |
| |
| @Test(timeout = 20000) |
| public void testAsyncCompletionPresettleAllSendToTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, false, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testAsyncCompletionPresettleAllSendToQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, false, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testsyncCompletionPresettleAllAnonymousSendToTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, true, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testsyncCompletionPresettleAllAnonymousSendToQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleAll=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, true, true, true, Queue.class); |
| } |
| |
| //----- Test the jms.presettleProducers with asynchronous completion -----// |
| |
| @Test(timeout = 20000) |
| public void testAsyncCompletionPresettleProducersTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, false, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testAsyncCompletionPresettleProducersQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, false, true, true, Queue.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testAsyncCompletionPresettleProducersAnonymousTopic() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, true, true, true, Topic.class); |
| } |
| |
| @Test(timeout = 20000) |
| public void testAsyncCompletionPresettleProducersAnonymousQueue() throws Exception { |
| String presettleConfig = "?jms.presettlePolicy.presettleProducers=true"; |
| doTestAsyncCompletionProducerWithPresettleOptions(presettleConfig, false, true, true, true, Queue.class); |
| } |
| |
| //----- Asynchronous Completion test method implementation ---------------// |
| |
| private void doTestAsyncCompletionProducerWithPresettleOptions(String uriOptions, boolean transacted, boolean anonymous, boolean senderSettled, boolean transferSettled, Class<? extends Destination> destType) throws Exception { |
| doTestAsyncCompletionProducerWithPresettleOptions(uriOptions, transacted, anonymous, true, senderSettled, transferSettled, destType); |
| } |
| |
| private void doTestAsyncCompletionProducerWithPresettleOptions(String uriOptions, boolean transacted, boolean anonymous, boolean relaySupported, boolean senderSettled, boolean transferSettled, Class<? extends Destination> destType) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| Connection connection = testFixture.establishConnecton(testPeer, uriOptions, relaySupported ? serverCapabilities : null, null); |
| testPeer.expectBegin(); |
| |
| Session session = null; |
| Binary txnId = null; |
| |
| if (transacted) { |
| // Expect the session, with an immediate link to the transaction coordinator |
| // using a target with the expected capabilities only. |
| CoordinatorMatcher txCoordinatorMatcher = new CoordinatorMatcher(); |
| txCoordinatorMatcher.withCapabilities(arrayContaining(TxnCapability.LOCAL_TXN)); |
| testPeer.expectSenderAttach(txCoordinatorMatcher, false, false); |
| |
| // First expect an unsettled 'declare' transfer to the txn coordinator, and |
| // reply with a declared disposition state containing the txnId. |
| txnId = new Binary(new byte[]{ (byte) 1, (byte) 2, (byte) 3, (byte) 4}); |
| testPeer.expectDeclare(txnId); |
| |
| session = connection.createSession(true, Session.SESSION_TRANSACTED); |
| } else { |
| session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| } |
| |
| Destination destination = null; |
| if (destType == Queue.class) { |
| destination = session.createQueue("MyQueue"); |
| } else if (destType == Topic.class) { |
| destination = session.createTopic("MyTopis"); |
| } else if (destType == TemporaryQueue.class) { |
| String dynamicAddress = "myTempQueueAddress"; |
| testPeer.expectTempQueueCreationAttach(dynamicAddress); |
| destination = session.createTemporaryQueue(); |
| } else if (destType == TemporaryTopic.class) { |
| String dynamicAddress = "myTempTopicAddress"; |
| testPeer.expectTempTopicCreationAttach(dynamicAddress); |
| destination = session.createTemporaryTopic(); |
| } else { |
| fail("unexpected type"); |
| } |
| |
| if (senderSettled) { |
| testPeer.expectSettledSenderAttach(); |
| } else { |
| testPeer.expectSenderAttach(); |
| } |
| |
| TestJmsCompletionListener listener = new TestJmsCompletionListener(); |
| MessageProducer producer = null; |
| if (anonymous) { |
| producer = session.createProducer(null); |
| } else { |
| producer = session.createProducer(destination); |
| } |
| |
| // Create and transfer a new message |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| headersMatcher.withDurable(equalTo(true)); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| Matcher<?> stateMatcher = nullValue(); |
| if (transacted) { |
| stateMatcher = new TransactionalStateMatcher(); |
| ((TransactionalStateMatcher) stateMatcher).withTxnId(equalTo(txnId)); |
| ((TransactionalStateMatcher) stateMatcher).withOutcome(nullValue()); |
| } |
| |
| ListDescribedType responseState = new Accepted(); |
| if (transacted) { |
| TransactionalState txState = new TransactionalState(); |
| txState.setTxnId(txnId); |
| txState.setOutcome(new Accepted()); |
| } |
| |
| if (transferSettled) { |
| testPeer.expectTransfer(messageMatcher, stateMatcher, true, false, responseState, false); |
| } else { |
| testPeer.expectTransfer(messageMatcher, stateMatcher, false, true, responseState, true); |
| } |
| |
| if (anonymous && !relaySupported) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| Message message = session.createTextMessage(); |
| |
| if (anonymous) { |
| producer.send(destination, message, listener); |
| } else { |
| producer.send(message, listener); |
| } |
| |
| if (transacted) { |
| testPeer.expectDischarge(txnId, true); |
| } |
| |
| testPeer.expectClose(); |
| |
| assertTrue("Did not get async callback", listener.awaitCompletion(2000, TimeUnit.SECONDS)); |
| assertNull(listener.exception); |
| assertNotNull(listener.message); |
| assertTrue(listener.message instanceof TextMessage); |
| assertEquals(1, listener.successCount); |
| assertEquals(0, listener.errorCount); |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| private class TestJmsCompletionListener implements CompletionListener { |
| |
| private final CountDownLatch completed; |
| |
| public volatile int successCount; |
| public volatile int errorCount; |
| |
| public volatile Message message; |
| public volatile Exception exception; |
| |
| public TestJmsCompletionListener() { |
| this(1); |
| } |
| |
| public TestJmsCompletionListener(int expected) { |
| this.completed = new CountDownLatch(expected); |
| } |
| |
| public boolean awaitCompletion(long timeout, TimeUnit units) throws InterruptedException { |
| return completed.await(timeout, units); |
| } |
| |
| @Override |
| public void onCompletion(Message message) { |
| LOG.info("JmsCompletionListener onCompletion called with message: {}", message); |
| this.message = message; |
| this.successCount++; |
| |
| completed.countDown(); |
| } |
| |
| @Override |
| public void onException(Message message, Exception exception) { |
| LOG.info("JmsCompletionListener onException called with message: {} error {}", message, exception); |
| |
| this.message = message; |
| this.exception = exception; |
| this.errorCount++; |
| |
| completed.countDown(); |
| } |
| } |
| } |
