| /* |
| * 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.hamcrest.Matchers.equalTo; |
| import static org.hamcrest.Matchers.nullValue; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| 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.UUID; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import javax.jms.CompletionListener; |
| import javax.jms.Connection; |
| import javax.jms.IllegalStateException; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.MessageProducer; |
| import javax.jms.Queue; |
| import javax.jms.ResourceAllocationException; |
| import javax.jms.Session; |
| import javax.jms.TextMessage; |
| import javax.jms.Topic; |
| |
| import org.apache.qpid.jms.JmsConnection; |
| import org.apache.qpid.jms.JmsDefaultConnectionListener; |
| import org.apache.qpid.jms.JmsSendTimedOutException; |
| import org.apache.qpid.jms.test.QpidJmsTestCase; |
| import org.apache.qpid.jms.test.testpeer.TestAmqpPeer; |
| import org.apache.qpid.jms.test.testpeer.basictypes.AmqpError; |
| import org.apache.qpid.jms.test.testpeer.basictypes.TerminusDurability; |
| import org.apache.qpid.jms.test.testpeer.describedtypes.Rejected; |
| import org.apache.qpid.jms.test.testpeer.matchers.TargetMatcher; |
| 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.MessagePropertiesSectionMatcher; |
| import org.apache.qpid.jms.test.testpeer.matchers.sections.TransferPayloadCompositeMatcher; |
| import org.apache.qpid.jms.test.testpeer.matchers.types.EncodedAmqpValueMatcher; |
| import org.apache.qpid.jms.util.QpidJMSTestRunner; |
| import org.apache.qpid.jms.util.Repeat; |
| import org.junit.Test; |
| import org.junit.runner.RunWith; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Tests for the Anonymous Fallback producer implementation. |
| * |
| * DO NOT add capability to indicate server support for ANONYMOUS-RELAY for any of these tests |
| */ |
| @RunWith(QpidJMSTestRunner.class) |
| public class AnonymousFallbackProducerIntegrationTest extends QpidJmsTestCase { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(AnonymousFallbackProducerIntegrationTest.class); |
| |
| private final IntegrationTestFixture testFixture = new IntegrationTestFixture(); |
| |
| @Test(timeout = 20000) |
| public void testCloseSenderWithNoActiveFallbackProducers() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| Connection connection = testFixture.establishConnecton(testPeer); |
| testPeer.expectBegin(); |
| testPeer.expectClose(); |
| |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| MessageProducer producer = session.createProducer(null); |
| |
| producer.close(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseProducerDuringSyncSendNoCache() throws Exception { |
| doTestRemotelyCloseProducerDuringSyncSend(0); |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseProducerDuringSyncSendOneCached() throws Exception { |
| doTestRemotelyCloseProducerDuringSyncSend(1); |
| } |
| |
| private void doTestRemotelyCloseProducerDuringSyncSend(int cacheSize) throws Exception { |
| final String BREAD_CRUMB = "ErrorMessageBreadCrumb"; |
| |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| // Use a long timeout to ensure no early evictions in this test. |
| Connection connection = testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + cacheSize + "&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Expect producer creation, give it credit. |
| testPeer.expectSenderAttach(); |
| |
| String text = "myMessage"; |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true)); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| messageMatcher.setPropertiesMatcher(propsMatcher); |
| messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(text)); |
| |
| // Expect a message to be sent, but don't send a disposition in |
| // response, simply remotely close the producer instead. |
| testPeer.expectTransfer(messageMatcher, nullValue(), false, false, null, false); |
| testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, true, AmqpError.RESOURCE_LIMIT_EXCEEDED, BREAD_CRUMB); |
| testPeer.expectClose(); |
| |
| Queue queue = session.createQueue("myQueue"); |
| final MessageProducer producer = session.createProducer(null); |
| |
| Message message = session.createTextMessage(text); |
| |
| try { |
| producer.send(queue, message); |
| fail("Expected exception to be thrown"); |
| } catch (JMSException jmse) { |
| LOG.trace("JMSException thrown from send: ", jmse); |
| // Expected but requires some context to be correct. |
| assertTrue(jmse instanceof ResourceAllocationException); |
| assertNotNull("Expected exception to have a message", jmse.getMessage()); |
| assertTrue("Expected breadcrumb to be present in message", jmse.getMessage().contains(BREAD_CRUMB)); |
| } |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(3000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseProducerWithSendWaitingForCreditNoCache() throws Exception { |
| doTestRemotelyCloseProducerWithSendWaitingForCredit(0); |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseProducerWithSendWaitingForCreditOneCached() throws Exception { |
| doTestRemotelyCloseProducerWithSendWaitingForCredit(1); |
| } |
| |
| private void doTestRemotelyCloseProducerWithSendWaitingForCredit(int cacheSize) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Use a long timeout to ensure no early evictions in this test. |
| Connection connection = testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + cacheSize + "&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Expect producer creation, don't give it credit. |
| testPeer.expectSenderAttachWithoutGrantingCredit(); |
| |
| // Producer has no credit so the send should block waiting for it, then fail when the remote close occurs |
| testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, true, AmqpError.RESOURCE_LIMIT_EXCEEDED, "Producer closed", 50); |
| testPeer.expectClose(); |
| |
| Queue queue = session.createQueue("myQueue"); |
| final MessageProducer producer = session.createProducer(null); |
| |
| Message message = session.createTextMessage("myMessage"); |
| |
| try { |
| producer.send(queue, message); |
| fail("Expected exception to be thrown due to close of producer"); |
| } catch (ResourceAllocationException rae) { |
| // Expected if remote close beat the send to the provider |
| } catch (IllegalStateException ise) { |
| // Can happen if send fires before remote close if processed. |
| } |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(3000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseConnectionDuringSyncSendNoCache() throws Exception { |
| doTestRemotelyCloseConnectionDuringSyncSend(0); |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseConnectionDuringSyncSendOneCached() throws Exception { |
| doTestRemotelyCloseConnectionDuringSyncSend(1); |
| } |
| |
| private void doTestRemotelyCloseConnectionDuringSyncSend(int cacheSize) throws Exception { |
| final String BREAD_CRUMB = "ErrorMessageBreadCrumb"; |
| |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Use a long timeout to ensure no early evictions in this test. |
| Connection connection = testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + cacheSize + "&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Expect producer creation, give it credit. |
| testPeer.expectSenderAttach(); |
| |
| String text = "myMessage"; |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true)); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| messageMatcher.setPropertiesMatcher(propsMatcher); |
| messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(text)); |
| |
| // Expect a message to be sent, but don't send a disposition in |
| // response, simply remotely close the connection instead. |
| testPeer.expectTransfer(messageMatcher, nullValue(), false, false, null, false); |
| testPeer.remotelyCloseConnection(true, AmqpError.RESOURCE_LIMIT_EXCEEDED, BREAD_CRUMB); |
| |
| Queue queue = session.createQueue("myQueue"); |
| final MessageProducer producer = session.createProducer(null); |
| |
| Message message = session.createTextMessage(text); |
| |
| try { |
| producer.send(queue, message); |
| fail("Expected exception to be thrown"); |
| } catch (JMSException jmse) { |
| // Expected exception with specific context |
| assertTrue(jmse instanceof ResourceAllocationException); |
| assertNotNull("Expected exception to have a message", jmse.getMessage()); |
| assertTrue("Expected breadcrumb to be present in message", jmse.getMessage().contains(BREAD_CRUMB)); |
| } |
| |
| testPeer.waitForAllHandlersToComplete(3000); |
| |
| connection.close(); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testSendWhenLinkCreditIsZeroAndTimeoutNoCache() throws Exception { |
| doTestSendWhenLinkCreditIsZeroAndTimeout(0); |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testSendWhenLinkCreditIsZeroAndTimeoutCacheOne() throws Exception { |
| doTestSendWhenLinkCreditIsZeroAndTimeout(1); |
| } |
| |
| private void doTestSendWhenLinkCreditIsZeroAndTimeout(int cacheSize) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Use a long timeout to ensure no early evictions in this test. |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + cacheSize + "&amqp.anonymousFallbackCacheTimeout=60000"); |
| connection.setSendTimeout(500); |
| |
| testPeer.expectBegin(); |
| |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| String queueName = "myQueue"; |
| Queue queue = session.createQueue(queueName); |
| |
| Message message = session.createTextMessage("text"); |
| |
| // Expect the producer to attach. Don't send any credit so that the client will |
| // block on a send and we can test our timeouts. |
| testPeer.expectSenderAttachWithoutGrantingCredit(); |
| if (cacheSize == 0) { |
| testPeer.expectDetach(true, true, true); |
| } |
| testPeer.expectClose(); |
| |
| MessageProducer producer = session.createProducer(null); |
| |
| try { |
| producer.send(queue, message); |
| fail("Send should time out."); |
| } catch (JmsSendTimedOutException jmsEx) { |
| LOG.info("Caught expected error: {}", jmsEx.getMessage()); |
| } catch (Throwable error) { |
| fail("Send should time out, but got: " + error.getMessage()); |
| } |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testSyncSendFailureHandled() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=0&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination, then closing the link after the message is sent. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(topicName)); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher, nullValue(), new Rejected(), true); |
| testPeer.expectDetach(true, true, true); |
| |
| Message message = session.createMessage(); |
| try { |
| producer.send(dest, message); |
| fail("Send should fail"); |
| } catch (JMSException jmsEx) { |
| LOG.debug("Caught expected error from failed send."); |
| } |
| |
| // Repeat the send and observe another attach->transfer->detach. |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| producer.send(dest, message); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testAsyncSendFailureHandled() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final CountDownLatch sendFailureReportedToListener = new CountDownLatch(1); |
| final AtomicReference<Throwable> sendFailureError = new AtomicReference<>(); |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?jms.forceAsyncSend=true&amqp.anonymousFallbackCacheSize=0&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| connection.setExceptionListener((error) -> { |
| sendFailureError.compareAndSet(null, error); |
| sendFailureReportedToListener.countDown(); |
| }); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| //Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination, then closing the link after the message is sent. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(topicName)); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| final String BREAD_CRUMB = "SEND FAILURE EXPECTED"; |
| |
| org.apache.qpid.jms.test.testpeer.describedtypes.Error rejectError = new org.apache.qpid.jms.test.testpeer.describedtypes.Error(); |
| rejectError.setCondition(AmqpError.RESOURCE_LIMIT_EXCEEDED); |
| rejectError.setDescription(BREAD_CRUMB); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher, nullValue(), new Rejected().setError(rejectError), true); |
| testPeer.expectDetach(true, true, true); |
| |
| // Producer should act as synchronous regardless of asynchronous send setting. |
| Message message = session.createMessage(); |
| try { |
| producer.send(dest, message); |
| } catch (JMSException jmsEx) { |
| LOG.debug("Caught expected error from failed send."); |
| fail("Send should not fail as it should have fired asynchronously"); |
| } |
| |
| // Repeat the send and observe another attach->transfer->detach. |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| assertTrue("Send failure not reported to exception handler", sendFailureReportedToListener.await(5, TimeUnit.SECONDS)); |
| assertNotNull(sendFailureError.get()); |
| assertTrue(sendFailureError.get() instanceof ResourceAllocationException); |
| assertTrue(sendFailureError.get().getMessage().contains(BREAD_CRUMB)); |
| |
| producer.send(dest, message); |
| |
| // Send here is asynchronous so we need to wait for disposition to arrive and detach to happen |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testAsyncCompletionListenerSendFailureHandled() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=0&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| //Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination, then closing the link after the message is sent. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(topicName)); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| String content = "testContent"; |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(new MessageHeaderSectionMatcher(true)); |
| messageMatcher.setMessageAnnotationsMatcher(new MessageAnnotationsSectionMatcher(true)); |
| messageMatcher.setPropertiesMatcher(new MessagePropertiesSectionMatcher(true)); |
| messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(content)); |
| |
| TestJmsCompletionListener completionListener = new TestJmsCompletionListener(); |
| Message message = session.createTextMessage(content); |
| |
| final String BREAD_CRUMB = "SEND FAILURE EXPECTED"; |
| |
| org.apache.qpid.jms.test.testpeer.describedtypes.Error rejectError = new org.apache.qpid.jms.test.testpeer.describedtypes.Error(); |
| rejectError.setCondition(AmqpError.RESOURCE_LIMIT_EXCEEDED); |
| rejectError.setDescription(BREAD_CRUMB); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher, nullValue(), new Rejected().setError(rejectError), true); |
| testPeer.expectDetach(true, true, true); |
| |
| // The fallback producer acts as synchronous regardless of the completion listener, |
| // so exceptions are thrown from send. Only onComplete uses the listener. |
| try { |
| producer.send(dest, message, completionListener); |
| } catch (JMSException jmsEx) { |
| LOG.debug("Caught unexpected error from failed send."); |
| fail("Send should not fail for asychrnous completion sends"); |
| } |
| |
| // Repeat the send (but accept this time) and observe another attach->transfer->detach. |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| assertTrue("Send failure not reported to exception handler", completionListener.awaitCompletion(5, TimeUnit.SECONDS)); |
| assertNotNull(completionListener.exception); |
| assertTrue(completionListener.exception instanceof ResourceAllocationException); |
| assertTrue(completionListener.exception.getMessage().contains(BREAD_CRUMB)); |
| |
| TestJmsCompletionListener completionListener2 = new TestJmsCompletionListener(); |
| |
| producer.send(dest, message, completionListener2); |
| |
| assertTrue("Did not get completion callback", completionListener2.awaitCompletion(5, TimeUnit.SECONDS)); |
| assertNull(completionListener2.exception); |
| Message receivedMessage2 = completionListener2.message; |
| assertNotNull(receivedMessage2); |
| assertTrue(receivedMessage2 instanceof TextMessage); |
| assertEquals(content, ((TextMessage) receivedMessage2).getText()); |
| |
| // Asynchronous send requires a wait otherwise we can close before the detach which we are testing for. |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testAsyncCompletionListenerSendWhenNoCacheConfigured() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=0&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| //Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination, then closing the link after the message is sent. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(topicName)); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| String content = "testContent"; |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(new MessageHeaderSectionMatcher(true)); |
| messageMatcher.setMessageAnnotationsMatcher(new MessageAnnotationsSectionMatcher(true)); |
| messageMatcher.setPropertiesMatcher(new MessagePropertiesSectionMatcher(true)); |
| messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(content)); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| TestJmsCompletionListener completionListener = new TestJmsCompletionListener(); |
| Message message = session.createTextMessage(content); |
| |
| producer.send(dest, message, completionListener); |
| |
| assertTrue("Did not get completion callback", completionListener.awaitCompletion(5, TimeUnit.SECONDS)); |
| assertNull(completionListener.exception); |
| Message receivedMessage = completionListener.message; |
| assertNotNull(receivedMessage); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals(content, ((TextMessage) receivedMessage).getText()); |
| |
| // Repeat the send and observe another attach->transfer->detach. |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| TestJmsCompletionListener completionListener2 = new TestJmsCompletionListener(); |
| |
| producer.send(dest, message, completionListener2); |
| |
| assertTrue("Did not get completion callback", completionListener2.awaitCompletion(5, TimeUnit.SECONDS)); |
| assertNull(completionListener2.exception); |
| Message receivedMessage2 = completionListener2.message; |
| assertNotNull(receivedMessage2); |
| assertTrue(receivedMessage2 instanceof TextMessage); |
| assertEquals(content, ((TextMessage) receivedMessage2).getText()); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyEndFallbackProducerCompletesAsyncSends() throws Exception { |
| final String BREAD_CRUMB = "ErrorMessage"; |
| |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| final CountDownLatch producerClosed = new CountDownLatch(1); |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=0&amqp.anonymousFallbackCacheTimeout=60000"); |
| connection.addConnectionListener(new JmsDefaultConnectionListener() { |
| @Override |
| public void onProducerClosed(MessageProducer producer, Throwable exception) { |
| producerClosed.countDown(); |
| } |
| }); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Create a producer, then remotely end the session afterwards. |
| testPeer.expectSenderAttach(); |
| |
| Queue queue = session.createQueue("myQueue"); |
| final MessageProducer producer = session.createProducer(queue); |
| |
| final int MSG_COUNT = 3; |
| |
| for (int i = 0; i < MSG_COUNT; ++i) { |
| testPeer.expectTransferButDoNotRespond(new TransferPayloadCompositeMatcher()); |
| } |
| |
| testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, true, AmqpError.RESOURCE_LIMIT_EXCEEDED, BREAD_CRUMB, 50); |
| |
| TestJmsCompletionListener listener = new TestJmsCompletionListener(MSG_COUNT); |
| try { |
| for (int i = 0; i < MSG_COUNT; ++i) { |
| Message message = session.createTextMessage("content"); |
| producer.send(message, listener); |
| } |
| } catch (JMSException e) { |
| LOG.warn("Caught unexpected error: {}", e.getMessage()); |
| fail("No expected exception for this send."); |
| } |
| |
| testPeer.waitForAllHandlersToComplete(2000); |
| |
| // Verify the sends gets marked as having failed |
| assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS)); |
| assertEquals(MSG_COUNT, listener.errorCount); |
| |
| // Verify the producer gets marked closed |
| assertTrue("Producer closed callback didn't trigger", producerClosed.await(5, TimeUnit.SECONDS)); |
| try { |
| producer.getDeliveryMode(); |
| fail("Expected ISE to be thrown due to being closed"); |
| } catch (IllegalStateException jmsise) { |
| String errorMessage = jmsise.getCause().getMessage(); |
| assertTrue(errorMessage.contains(AmqpError.RESOURCE_LIMIT_EXCEEDED.toString())); |
| assertTrue(errorMessage.contains(BREAD_CRUMB)); |
| } |
| |
| // Try closing it explicitly, should effectively no-op in client. |
| // The test peer will throw during close if it sends anything. |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseSessionAndAttemptAsyncCompletionSendThrowsAndLeavesMessageReadable() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=0&amqp.anonymousFallbackCacheTimeout=60000"); |
| |
| final CountDownLatch sessionClosed = new CountDownLatch(1); |
| |
| connection.addConnectionListener(new JmsDefaultConnectionListener() { |
| |
| @Override |
| public void onSessionClosed(Session session, Throwable cause) { |
| sessionClosed.countDown(); |
| } |
| }); |
| |
| testPeer.expectBegin(); |
| |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Queue queue = session.createQueue("myQueue"); |
| MessageProducer producer = session.createProducer(null); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination, then closing the link after the message is sent. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo("myQueue")); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| String content = "testContent"; |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(new MessageHeaderSectionMatcher(true)); |
| messageMatcher.setMessageAnnotationsMatcher(new MessageAnnotationsSectionMatcher(true)); |
| messageMatcher.setPropertiesMatcher(new MessagePropertiesSectionMatcher(true)); |
| messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(content)); |
| |
| // Perform a send and observe an attach->transfer->detach. |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| testPeer.remotelyEndLastOpenedSession(true); |
| |
| Message message1 = session.createTextMessage(content); |
| Message message2 = session.createTextMessage(content); |
| assertNull("Should not yet have a JMSDestination", message1.getJMSDestination()); |
| assertNull("Should not yet have a JMSDestination", message2.getJMSDestination()); |
| producer.send(queue, message1); |
| |
| testPeer.waitForAllHandlersToComplete(2000); |
| |
| assertTrue("Session should have been closed", sessionClosed.await(2, TimeUnit.SECONDS)); |
| |
| TestJmsCompletionListener listener = new TestJmsCompletionListener(); |
| try { |
| producer.send(queue, message2, listener); |
| fail("Expected exception to be thrown for this send."); |
| } catch (JMSException e) { |
| LOG.trace("Caught expected exception: {}", e.getMessage()); |
| } |
| |
| assertFalse("Should not get async callback", listener.awaitCompletion(5, TimeUnit.MILLISECONDS)); |
| |
| // Message should be readable but not carry a destination as it wasn't actually sent anywhere |
| assertNull("Should not have a readable JMSDestination", message2.getJMSDestination()); |
| assertEquals("Message body not as expected", content, ((TextMessage) message2).getText()); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(2000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testFallbackProducerRecoversFromRefusalOfSenderOpenOnNextSend() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=1&amqp.anonymousFallbackCacheTimeout=0"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(topicName)); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Topic dest = session.createTopic(topicName); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| Message message = session.createMessage(); |
| |
| // Refuse the attach which should result in fallback producer detaching the refused |
| // link attach and the send should then fail. |
| testPeer.expectSenderAttach(targetMatcher, true, false); |
| testPeer.expectDetach(true, false, false); |
| |
| try { |
| producer.send(dest, message); |
| fail("Send should have failed because sender link was refused."); |
| } catch (JMSException ex) { |
| LOG.trace("Caught expected exception: ", ex); |
| } |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| producer.send(dest, message); |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 60000) |
| public void testRepeatedSendToSameAddressWhenCacheSizeOfOneKeepsFallbackProducerInCache() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final int MESSAGE_COUNT = 25; |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=1&amqp.anonymousFallbackCacheTimeout=200"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(topicName)); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, true); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| Topic dest = session.createTopic(topicName); |
| Message message = session.createMessage(); |
| |
| // Setup our expectations |
| for (int i = 1; i <= MESSAGE_COUNT; ++i) { |
| testPeer.expectTransfer(messageMatcher); |
| } |
| |
| testPeer.expectDetach(true, true, true); |
| |
| // First round of sends should open and cache sender links |
| for (int i = 1; i <= MESSAGE_COUNT; ++i) { |
| producer.send(dest, message); |
| } |
| |
| LOG.debug("Finished with send cycle, producer should now timeout"); |
| |
| // The eviction timer should reduce the cache to zero after we go idle |
| testPeer.waitForAllHandlersToComplete(3000); |
| |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testSendToMultipleDestinationsOpensNewSendersWhenCaching() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final int CACHE_SIZE = 5; |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + CACHE_SIZE + "&amqp.anonymousFallbackCacheTimeout=0"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| // First round of sends should open and cache sender links |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + i); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| // This round of sends should reuse the cached links |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + i); |
| |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| // Cached senders should all close when the producer is closed. |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 30000) |
| public void testCachedFallbackProducersAreTimedOut() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final int CACHE_SIZE = 5; |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + CACHE_SIZE + "&amqp.anonymousFallbackCacheTimeout=300"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| // First round of sends should open and cache sender links |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + i); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| // Cached senders should all close when the cache timeout is reached and they are expired |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| // On a slow CI machine we could fail here due to the timeouts not having run. |
| testPeer.waitForAllHandlersToComplete(6000); |
| |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testCachedFallbackProducerEvictedBySendToUncachedAddress() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final int CACHE_SIZE = 2; |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + CACHE_SIZE + "&amqp.anonymousFallbackCacheTimeout=0"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| // First round of sends should open and cache sender links |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + i); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| // Second round using different addresses for the sends should evict old links and open new ones |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + UUID.randomUUID().toString()); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectDetach(true, true, true); |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| // The current cached senders should all close when the producer is closed. |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testCachedFallbackProducerEvictedBySendToUncachedAddressHandlesDelayedResponse() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=1&amqp.anonymousFallbackCacheTimeout=0"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| Topic dest1 = session.createTopic(topicName + 1); |
| Topic dest2 = session.createTopic(topicName + 2); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest1.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest1, message); |
| |
| // Expect new send to a different destination to detach the previous cached link |
| // and once the response arrives the send should complete normally. |
| targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest2.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| message = session.createMessage(); |
| |
| testPeer.expectDetach(true, false, false); |
| // Workaround to allow a deferred detach at a later time. |
| testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(false, true, AmqpError.RESOURCE_DELETED, "error", 20); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.expectDetach(true, true, true); |
| |
| producer.send(dest2, message); |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testRemotelyCloseCachedFallbackProducerFreesSlotInCache() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final int CACHE_SIZE = 3; |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + CACHE_SIZE + "&amqp.anonymousFallbackCacheTimeout=0"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| // First round of sends should open and cache sender links |
| for (int i = 1; i < CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + i); |
| |
| // Expect a new message sent to the above created destination to result in a new |
| // sender link attached to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| Topic dest = session.createTopic(topicName + CACHE_SIZE); |
| |
| // Expect a new message sent to the above created destination to result in a new |
| // sender link attached to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, true); |
| |
| producer.send(dest, message); |
| |
| // Must ensure that the next send only fires after the remote detach has occurred otherwise |
| // it will definitely evict another producer from the cache. |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| dest = session.createTopic(topicName + UUID.randomUUID().toString()); |
| |
| // Expect a new message sent to the above created destination to result in a new |
| // sender link attached to the given destination. Existing cached producers should |
| // remain in the cache as a slot should now be open. |
| targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| |
| // The current cached senders should all close when the producer is closed. |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| producer.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| @Repeat(repetitions = 1) |
| @Test(timeout = 20000) |
| public void testFailureOfOneCacheProducerCloseOnPropagatedToMainProducerClose() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| |
| final int CACHE_SIZE = 3; |
| |
| JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer, |
| "?amqp.anonymousFallbackCacheSize=" + CACHE_SIZE + "&amqp.anonymousFallbackCacheTimeout=0"); |
| |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| |
| // Expect no AMQP traffic when we create the anonymous producer, as it will wait |
| // for an actual send to occur on the producer before anything occurs on the wire |
| |
| // Create an anonymous producer |
| MessageProducer producer = session.createProducer(null); |
| assertNotNull("Producer object was null", producer); |
| |
| MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true); |
| MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true); |
| TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher(); |
| messageMatcher.setHeadersMatcher(headersMatcher); |
| messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher); |
| |
| // First round of sends should open and cache sender links |
| for (int i = 1; i <= CACHE_SIZE; ++i) { |
| Topic dest = session.createTopic(topicName + i); |
| |
| // Expect a new message sent by the above producer to cause creation of a new |
| // sender link to the given destination. |
| TargetMatcher targetMatcher = new TargetMatcher(); |
| targetMatcher.withAddress(equalTo(dest.getTopicName())); |
| targetMatcher.withDynamic(equalTo(false)); |
| targetMatcher.withDurable(equalTo(TerminusDurability.NONE)); |
| |
| Message message = session.createMessage(); |
| |
| testPeer.expectSenderAttach(targetMatcher, false, false); |
| testPeer.expectTransfer(messageMatcher); |
| |
| producer.send(dest, message); |
| } |
| |
| // The current cached senders should all close when the producer is closed. |
| for (int i = 1; i < CACHE_SIZE; ++i) { |
| testPeer.expectDetach(true, true, true); |
| } |
| |
| // Last one carries error but since we asked for close it should be ignored as we got what we wanted. |
| testPeer.expectDetach(true, true, true, AmqpError.RESOURCE_LOCKED, "Some error on detach"); |
| |
| try { |
| producer.close(); |
| } catch (JMSException ex) { |
| LOG.trace("Caught unexpected error: ", ex); |
| fail("Should not have thrown an error as close was requested so errors are ignored."); |
| } |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| private class TestJmsCompletionListener implements CompletionListener { |
| |
| private final CountDownLatch completed; |
| |
| @SuppressWarnings("unused") |
| 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(); |
| } |
| } |
| } |
