| /* |
| * 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.SHARED_SUBS; |
| import static org.apache.qpid.jms.provider.amqp.AmqpSupport.SUB_NAME_DELIMITER; |
| import static org.hamcrest.Matchers.equalTo; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.io.IOException; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.jms.Connection; |
| import javax.jms.ConnectionFactory; |
| import javax.jms.InvalidDestinationException; |
| import javax.jms.JMSException; |
| import javax.jms.MessageConsumer; |
| import javax.jms.Session; |
| import javax.jms.Topic; |
| import javax.jms.TopicSubscriber; |
| |
| import org.apache.qpid.jms.JmsConnection; |
| import org.apache.qpid.jms.JmsConnectionFactory; |
| import org.apache.qpid.jms.JmsDefaultConnectionListener; |
| import org.apache.qpid.jms.provider.amqp.AmqpSupport; |
| 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.proton.amqp.Symbol; |
| import org.hamcrest.Matcher; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public class SubscriptionsIntegrationTest extends QpidJmsTestCase { |
| |
| private final IntegrationTestFixture testFixture = new IntegrationTestFixture(); |
| |
| private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsIntegrationTest.class); |
| |
| // -------------------------------------- // |
| |
| @Test |
| public void testConstants() throws Exception { |
| assertEquals(Symbol.valueOf("SHARED-SUBS"), AmqpSupport.SHARED_SUBS); |
| assertEquals("|", AmqpSupport.SUB_NAME_DELIMITER); |
| assertEquals(Symbol.valueOf("shared"), AmqpSupport.SHARED); |
| assertEquals(Symbol.valueOf("global"), AmqpSupport.GLOBAL); |
| } |
| |
| /** |
| * Verifies that a shared durable subscriber detaches links with closed = false. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCloseSharedDurableTopicSubscriberDetachesWithCloseFalse() throws Exception { |
| doSharedTopicSubscriberDetachTestImpl(true); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscriber detaches links with closed = true. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCloseSharedVolatileTopicSubscriberDetachesWithCloseTrue() throws Exception { |
| doSharedTopicSubscriberDetachTestImpl(false); |
| } |
| |
| private void doSharedTopicSubscriberDetachTestImpl(boolean durable) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| if (durable) { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| } else { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber; |
| if (durable) { |
| subscriber = session.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| subscriber = session.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| testPeer.expectDetach(!durable, true, !durable); |
| subscriber.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a shared durable subscribers name their links such that the first link is the |
| * bare subscription name, and subsquent links use a counter suffix to ensure they are unique. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubscriberLinkNames() throws Exception { |
| doSharedSubsriberLinkNamesHaveUniqueCounterSuffixTestImpl(true, true); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscribers names their links such that the subscription name |
| * is suffixed with a counter to ensure they are unique. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubscriberLinkNamesHaveUniqueCounterSuffix() throws Exception { |
| doSharedSubsriberLinkNamesHaveUniqueCounterSuffixTestImpl(false, true); |
| } |
| |
| /** |
| * Verifies that on a connection without a ClientID, shared durable subscribers name their links |
| * such that the first link is the subscription name with 'global' suffix, and subsequent links |
| * additionally append a counter suffix to ensure they are unique. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubscriberLinkNamesNoClientID() throws Exception { |
| doSharedSubsriberLinkNamesHaveUniqueCounterSuffixTestImpl(true, false); |
| } |
| |
| /** |
| * Verifies that on a connection without a ClientID, shared volatile subscribers names their links |
| * such that the subscription name is suffixed with a 'global' qualifier an counter to ensure |
| * they are unique. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubscriberLinkNamesHaveUniqueCounterSuffixNoClientID() throws Exception { |
| doSharedSubsriberLinkNamesHaveUniqueCounterSuffixTestImpl(false, false); |
| } |
| |
| private void doSharedSubsriberLinkNamesHaveUniqueCounterSuffixTestImpl(boolean durable, boolean useClientID) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| Connection connection; |
| if(useClientID) { |
| connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| } else { |
| connection = testFixture.establishConnectonWithoutClientID(testPeer, serverCapabilities); |
| } |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first shared receiver |
| if (durable) { |
| String linkName = subscriptionName; |
| if(!useClientID) { |
| linkName += SUB_NAME_DELIMITER + "global"; |
| } |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, equalTo(linkName), useClientID); |
| } else { |
| String linkName; |
| if(useClientID) { |
| linkName = subscriptionName + SUB_NAME_DELIMITER + "volatile1"; |
| } else { |
| linkName = subscriptionName + SUB_NAME_DELIMITER + "global-volatile1"; |
| } |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, equalTo(linkName), useClientID); |
| } |
| testPeer.expectLinkFlow(); |
| |
| if (durable) { |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| session.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| // Attach the second shared receiver, expect the link name to differ |
| if (durable) { |
| String linkName = subscriptionName; |
| if(useClientID) { |
| linkName += SUB_NAME_DELIMITER + "2"; |
| } else { |
| linkName += SUB_NAME_DELIMITER + "global2"; |
| } |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, equalTo(linkName), useClientID); |
| } else { |
| String linkName; |
| if(useClientID) { |
| linkName = subscriptionName + SUB_NAME_DELIMITER + "volatile2"; |
| } else { |
| linkName = subscriptionName + SUB_NAME_DELIMITER + "global-volatile2"; |
| } |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, equalTo(linkName), useClientID); |
| } |
| testPeer.expectLinkFlow(); |
| |
| if (durable) { |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| session.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a shared durable subscribers names their links on a per-connection basis, such that |
| * suffix counter on one connection is not impacted by subscriptions on another connection. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubsriberLinkNamesMultipleConnectionSubs() throws Exception { |
| doMultipleConnectionSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(true); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscribers names their links on a per-connection basis, such that |
| * suffix counter on one connection is not impacted by subscriptions on another connection. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubsriberLinkNamesHaveUniqueCounterSuffixMultipleConnectionSubs() throws Exception { |
| doMultipleConnectionSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(false); |
| } |
| |
| private void doMultipleConnectionSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(boolean durable) throws Exception { |
| try (TestAmqpPeer peer1 = new TestAmqpPeer(); |
| TestAmqpPeer peer2 = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish first connection |
| Connection connection1 = testFixture.establishConnecton(peer1, serverCapabilities); |
| connection1.start(); |
| |
| peer1.expectBegin(); |
| Session sessionConn1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Establish second connection |
| Connection connection2 = testFixture.establishConnecton(peer2, serverCapabilities); |
| connection2.start(); |
| |
| peer2.expectBegin(); |
| Session sessionConn2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = sessionConn1.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| Matcher<?> durSubLinkNameMatcher = equalTo(subscriptionName); |
| Matcher<?> volatileSubLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| |
| // Attach the first connections shared receiver |
| if (durable) { |
| peer1.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durSubLinkNameMatcher, true); |
| } else { |
| peer1.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, volatileSubLinkNameMatcher, true); |
| } |
| peer1.expectLinkFlow(); |
| |
| if (durable) { |
| sessionConn1.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| sessionConn1.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| // Attach the second connections shared receiver, expect the link name to be the same since its per-connection |
| if (durable) { |
| peer2.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durSubLinkNameMatcher, true); |
| } else { |
| peer2.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, volatileSubLinkNameMatcher, true); |
| } |
| peer2.expectLinkFlow(); |
| |
| if (durable) { |
| sessionConn2.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| sessionConn2.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| peer1.expectClose(); |
| connection1.close(); |
| |
| peer2.expectClose(); |
| connection2.close(); |
| |
| peer1.waitForAllHandlersToComplete(1000); |
| peer2.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a shared durable subscribers names their links on a per-connection basis, such that |
| * suffix counter for one session is impacted by subscriptions on another session on the same connection. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubsriberLinkNamesMultipleSessionSubs() throws Exception { |
| doMultipleSessionSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(true); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscribers names their links on a per-connection basis, such that |
| * suffix counter for one session is impacted by subscriptions on another session on the same connection. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubsriberLinkNamesHaveUniqueCounterSuffixMultipleSessionSubs() throws Exception { |
| doMultipleSessionSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(false); |
| } |
| |
| private void doMultipleSessionSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(boolean durable) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session1 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| testPeer.expectBegin(); |
| Session session2 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session1.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first sessions shared receiver |
| if (durable) { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| } else { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| if (durable) { |
| session1.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| session1.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| // Attach the second sessions shared receiver, expect the link name to be different since its per-connection |
| if (durable) { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| } else { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile2"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| if (durable) { |
| session2.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| session2.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that shared durable subscribers name their links on a per-subscription name basis, such that |
| * suffix counter for one subscription nameis not impacted by those for another subscription on the same connection. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubsriberLinkNamesMultipleNamedSubs() throws Exception { |
| doMultipleNamedSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(true); |
| } |
| |
| /** |
| * Verifies that shared volatile subscribers name their links on a per-subscription name basis, such that |
| * suffix counter for one subscription name is not impacted by those for another subscription on the same connection. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubsriberLinkNamesHaveUniqueCounterSuffixMultipleNamedSubs() throws Exception { |
| doMultipleNamedSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(false); |
| } |
| |
| private void doMultipleNamedSharedSubscriberLinkNamesHaveUniqueCounterSuffixTestImpl(boolean durable) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName1 = "mySubscription1"; |
| String subscriptionName2 = "mySubscription2"; |
| |
| // Attach the first subscriptions shared receivers |
| if (durable) { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName1); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName1, linkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| linkNameMatcher = equalTo(subscriptionName1 + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName1, linkNameMatcher, true); |
| } else { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName1 + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName1, linkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| linkNameMatcher = equalTo(subscriptionName1 + SUB_NAME_DELIMITER + "volatile2"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName1, linkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| if (durable) { |
| session.createSharedDurableConsumer(dest, subscriptionName1); |
| session.createSharedDurableConsumer(dest, subscriptionName1); |
| } else { |
| session.createSharedConsumer(dest, subscriptionName1); |
| session.createSharedConsumer(dest, subscriptionName1); |
| } |
| |
| // Attach the first subscriptions shared receivers |
| if (durable) { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName2); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName2, linkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| linkNameMatcher = equalTo(subscriptionName2 + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName2, linkNameMatcher, true); |
| } else { |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName2 + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName2, linkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| linkNameMatcher = equalTo(subscriptionName2 + SUB_NAME_DELIMITER + "volatile2"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName2, linkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| if (durable) { |
| session.createSharedDurableConsumer(dest, subscriptionName2); |
| session.createSharedDurableConsumer(dest, subscriptionName2); |
| } else { |
| session.createSharedConsumer(dest, subscriptionName2); |
| session.createSharedConsumer(dest, subscriptionName2); |
| } |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a shared volatile subscriber and shared durable subscriber with the same subscription name |
| * can be active on the same connection at the same time and names their links appropriately to distinguish |
| * themselves from each other. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableAndVolatileSubsCoexistUsingDistinctLinkNames() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the durable shared receiver |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| |
| // Attach the volatile shared receiver |
| Matcher<?> volatileLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, volatileLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| session.createSharedConsumer(dest, subscriptionName); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that on a connection which doesn't identify as supporting shared subscriptions, the |
| * attempt to create a shared durable subscriber fails if the link also doesn't identify as |
| * supporting shared subscriptions. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCreateSharedDurableTopicSubscriberFailsIfNotSupported() throws Exception { |
| doSharedSubscriptionNotSupportedTestImpl(true, false); |
| } |
| |
| /** |
| * Verifies that on a connection which doesn't identify as supporting shared subscriptions, the |
| * attempt to create a shared volatile subscriber fails if the link also doesn't identify as |
| * supporting shared subscriptions. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCreateSharedVolatileTopicSubscriberFailsIfNotSupported() throws Exception { |
| doSharedSubscriptionNotSupportedTestImpl(false, false); |
| } |
| |
| /** |
| * Verifies that when a durable shared subscriber creation fails because the connection and link |
| * both lack the capability indicating support, the subscriber is appropriately removed from |
| * record, releasing use of the link name such that attempting another subscriber creation |
| * uses the same initial link name. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCreateSharedDurableTopicSubscriberFailsIfNotSupportedReleasesLinkName() throws Exception { |
| doSharedSubscriptionNotSupportedTestImpl(true, true); |
| } |
| |
| /** |
| * Verifies that when a volatile shared subscriber creation fails because the connection and link |
| * both lack the capability indicating support, the subscriber is appropriately removed from |
| * record, releasing use of the link name such that attempting another subscriber creation |
| * uses the same initial link name. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCreateSharedVolatileTopicSubscriberFailsIfNotSupportedReleasesLinkName() throws Exception { |
| doSharedSubscriptionNotSupportedTestImpl(false, true); |
| } |
| |
| private void doSharedSubscriptionNotSupportedTestImpl(boolean durable, boolean repeat) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // DONT include server connection capability to indicate support for shared-subs |
| // This will cause the link capability to be desired, and we verify failure if not offered. |
| Symbol[] serverCapabilities = new Symbol[]{}; |
| |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| int iterations = repeat ? 2 : 1; |
| |
| // Expect a shared receiver to attach, then detach due to the link also not |
| // reporting it offers the shared subs capability, i.e sharing not supported. |
| for (int i = 0; i < iterations; i++) { |
| try { |
| if (durable) { |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true, false, true, true, false); |
| testPeer.expectDetach(false, true, false); |
| |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| Matcher<?> volatileLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedSubscriberAttach(topicName, subscriptionName, volatileLinkNameMatcher, false, false, true, true, false); |
| testPeer.expectDetach(true, true, true); |
| |
| session.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| fail("Expected an exception to be thrown"); |
| } catch (JMSException jmse) { |
| // expected |
| } |
| } |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| /** |
| * Verifies that on a connection which doesn't identify as supporting shared subscriptions, the |
| * attempt to create a shared durable subscriber succeeds if the server offers link capability for |
| * shared subscriptions. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCreateSharedDurableTopicSubscriberSucceedsWithOnlyLinkCapability() throws Exception { |
| doSharedSubscriptionLinkCapabilitySupportedTestImpl(true); |
| } |
| |
| /** |
| * Verifies that on a connection which doesn't identify as supporting shared subscriptions, the |
| * attempt to create a shared volatile subscriber succeeds if the server offers link capability for |
| * shared subscriptions. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testCreateSharedVolatileTopicSubscriberSucceedsWithOnlyLinkCapability() throws Exception { |
| doSharedSubscriptionLinkCapabilitySupportedTestImpl(false); |
| } |
| |
| private void doSharedSubscriptionLinkCapabilitySupportedTestImpl(boolean durable) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // DONT include server connection capability to indicate support for shared-subs. |
| // This will cause the link capability to be desired, and we verify success if offered. |
| Symbol[] serverCapabilities = new Symbol[]{}; |
| |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Expect a shared receiver to attach, and succeed due to the server offering |
| // the shared subs capability, i.e sharing is supported. |
| if (durable) { |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true, false, true, true, true); |
| testPeer.expectLinkFlow(); |
| |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| Matcher<?> volatileLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedSubscriberAttach(topicName, subscriptionName, volatileLinkNameMatcher, false, false, true, true, true); |
| testPeer.expectLinkFlow(); |
| |
| session.createSharedConsumer(dest, subscriptionName); |
| } |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a shared durable subscriber and exclusive durable subscriber with the same subscription name |
| * can't be active on the same connection at the same time (shared first, then exclusive). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedAndExclusiveDurableSubsCantCoexistSharedFirst() throws Exception { |
| doSharedAndExclusiveDurableSubsCantCoexistTestImpl(true); |
| } |
| |
| /** |
| * Verifies that a shared durable subscriber and exclusive durable subscriber with the same subscription name |
| * can't be active on the same connection at the same time (exclusive first, then shared). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedAndExclusiveDurableSubsCantCoexistExclusiveFirst() throws Exception { |
| doSharedAndExclusiveDurableSubsCantCoexistTestImpl(false); |
| } |
| |
| private void doSharedAndExclusiveDurableSubsCantCoexistTestImpl(boolean sharedFirst) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| if(sharedFirst) { |
| // Attach the durable shared receiver |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| } else { |
| // Attach the durable exclusive receiver |
| testPeer.expectDurableSubscriberAttach(topicName, subscriptionName); |
| testPeer.expectLinkFlow(); |
| |
| session.createDurableConsumer(dest, subscriptionName); |
| } |
| |
| testPeer.expectClose(); |
| |
| try { |
| if (sharedFirst) { |
| // Now try to attach a durable exclusive receiver |
| session.createDurableConsumer(dest, subscriptionName); |
| } else { |
| // Now try to attach a durable shared receiver |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| } |
| fail("Expected to fail due to concurrent shared + non-shared durable sub attempt"); |
| } catch (JMSException jmse) { |
| // Expected |
| } |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| @Test(timeout = 20000) |
| public void testExclusiveDurableSubCanOnlyBeActiveOnceAtATime() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the durable exclusive receiver |
| testPeer.expectDurableSubscriberAttach(topicName, subscriptionName); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber1 = session.createDurableConsumer(dest, subscriptionName); |
| |
| try { |
| // Now try to attach a second active durable exclusive receiver |
| session.createDurableConsumer(dest, subscriptionName); |
| fail("Expected to fail due to concurrent active subscriber attempt"); |
| } catch (JMSException jmse) { |
| // Expected |
| } |
| |
| testPeer.expectDetach(false, true, false); |
| subscriber1.close(); |
| |
| // Now try to attach a new active durable exclusive receiver |
| testPeer.expectDurableSubscriberAttach(topicName, subscriptionName); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber2 = session.createDurableConsumer(dest, subscriptionName); |
| assertNotNull(subscriber2); |
| |
| testPeer.expectClose(); |
| |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a upon failure to locate a subscription link during an |
| * unsubscribe attempt an [InvalidDestination] exception is thrown. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testUnsubscribeNonExistingSubscription() throws Exception { |
| doUnsubscribeNonExistingSubscriptionTestImpl(true); |
| } |
| |
| /** |
| * Verifies that a upon failure to locate a subscription link during an |
| * unsubscribe attempt an [InvalidDestination] exception is thrown, but |
| * this time using a connection without a ClientID set (and thus |
| * expecting a different link name to be used). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testUnsubscribeNonExistingSubscriptionWithoutClientID() throws Exception { |
| doUnsubscribeNonExistingSubscriptionTestImpl(false); |
| } |
| |
| private void doUnsubscribeNonExistingSubscriptionTestImpl(boolean hasClientID) throws JMSException, InterruptedException, Exception, IOException { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| Connection connection; |
| if(hasClientID) { |
| connection = testFixture.establishConnecton(testPeer); |
| } else { |
| connection = testFixture.establishConnectonWithoutClientID(testPeer, null); |
| } |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String subscriptionName = "myInvalidSubscriptionName"; |
| // Try to unsubscribe non-existing subscription |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(true, false, subscriptionName, null, hasClientID); |
| testPeer.expectDetach(true, true, true); |
| |
| try { |
| session.unsubscribe(subscriptionName); |
| fail("Should have thrown a InvalidDestinationException"); |
| } catch (InvalidDestinationException ide) { |
| // Expected |
| } |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| /** |
| * Verifies that a upon attempt to unsubscribe a subscription that is |
| * currently in active use by a subscriber, an exception is thrown, |
| * and that once the subscriber is closed a further unsubscribe attempt |
| * is successfully undertaken. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testUnsubscribeExclusiveDurableSubWhileActiveThenInactive() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| Connection connection = testFixture.establishConnecton(testPeer); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the durable exclusive receiver |
| testPeer.expectDurableSubscriberAttach(topicName, subscriptionName); |
| testPeer.expectLinkFlow(); |
| |
| TopicSubscriber subscriber = session.createDurableSubscriber(dest, subscriptionName); |
| assertNotNull("TopicSubscriber object was null", subscriber); |
| |
| // Now try to unsubscribe, should fail |
| try { |
| session.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| } |
| |
| // Now close the subscriber |
| testPeer.expectDetach(false, true, false); |
| |
| subscriber.close(); |
| |
| // Try to unsubscribe again, should work now |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true); |
| |
| session.unsubscribe(subscriptionName); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| /** |
| * Verifies that a upon attempt to unsubscribe a shared durable subscription that is |
| * currently in active use by multiple subscribers, an exception is thrown, and that |
| * only once the last subscriber is closed is an unsubscribe attempt successful. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testUnsubscribeSharedDurableSubWhileActiveThenInactive() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first durable shared receiver |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber1 = session.createSharedDurableConsumer(dest, subscriptionName); |
| |
| // Attach the second durable shared receiver |
| linkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber2 = session.createSharedDurableConsumer(dest, subscriptionName); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Now try to unsubscribe, should fail |
| try { |
| session.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| // Now close the first subscriber |
| testPeer.expectDetach(false, true, false); |
| subscriber1.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Try to unsubscribe again, should still fail |
| try { |
| session.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Now close the second subscriber |
| testPeer.expectDetach(false, true, false); |
| |
| subscriber2.close(); |
| |
| // Try to unsubscribe again, should now work |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true); |
| |
| session.unsubscribe(subscriptionName); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a upon a failed attempt to create a shared subscriber that it |
| * is possible to unsubscribe (not that it should be required), i.e that the |
| * failed creation does not leave an incorrect recording of an active subscriber. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testUnsubscribeAfterFailedCreation() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Fail to attach a shared durable receiver |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, linkNameMatcher, true, true); |
| testPeer.expectDetach(true, false, true); |
| |
| try { |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| // Try to unsubscribe, should be able to (strictly speaking an unsub attempt |
| // would probably fail normally, due to no subscription, but this test |
| // doesn't care about that, just that the attempt proceeds, so overlook that. |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true); |
| |
| session.unsubscribe(subscriptionName); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| /** |
| * Verifies that an unsubscribe attempt detects when the remote indicates that |
| * the request failed by attaching an error condition to the detach response. |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testUnsubscribeFailsWhenRemoteDetachResponseIndicatesFailure() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| Connection connection = testFixture.establishConnecton(testPeer); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| String subscriptionName = "mySubscription"; |
| |
| // Try to unsubscribe, should be able to (strictly speaking an unsub attempt |
| // would probably fail normally, due to no subscription, but this test |
| // doesn't care about that, just that the attempt proceeds, so overlook that. |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true, AmqpError.RESOURCE_LOCKED, |
| "Cannot unsubscibe when there are active consumers"); |
| |
| try { |
| session.unsubscribe(subscriptionName); |
| fail("Should throw an error if the remote indicate detach failed"); |
| } catch (JMSException jmsEx) { |
| LOG.info("Caught expected exception on unsubscribe: {}", jmsEx.getMessage()); |
| } |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| /** |
| * Verifies that subscriber cleanup occurs when the subscriber is remotely closed (after creation). |
| * |
| * @throws Exception if an unexpected error is encountered |
| */ |
| @Test(timeout = 20000) |
| public void testRemotelyDetachLinkWithDurableSharedConsumer() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| |
| final CountDownLatch subscriberClosed = new CountDownLatch(1); |
| ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { |
| @Override |
| public void onConsumerClosed(MessageConsumer consumer, Throwable exception) { |
| subscriberClosed.countDown(); |
| } |
| }); |
| |
| // Create first session |
| testPeer.expectBegin(); |
| Session session1 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session1.createTopic(topicName); |
| |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first shared receiver on the first session |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber1 = session1.createSharedDurableConsumer(dest, subscriptionName); |
| |
| // Create second session |
| testPeer.expectBegin(); |
| Session session2 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Attach the second shared receiver on the second session |
| durableLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| |
| testPeer.expectLinkFlow(); |
| |
| // Then remotely detach the link after the flow is received. |
| testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, false, AmqpError.RESOURCE_LIMIT_EXCEEDED, "TestingRemoteDetach"); |
| |
| MessageConsumer subscriber2 = session2.createSharedDurableConsumer(dest, subscriptionName); |
| assertNotNull(subscriber2); |
| |
| assertTrue("Consumer closed callback didn't trigger", subscriberClosed.await(5, TimeUnit.SECONDS)); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Now try to unsubscribe (using first session). It should fail due to sub still |
| // being in use on the first session. No frames should be sent. |
| try { |
| session1.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| // Now close the first subscriber |
| testPeer.expectDetach(false, true, false); |
| subscriber1.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Try to unsubscribe again (using first session), should now work. |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true); |
| |
| session1.unsubscribe(subscriptionName); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that a shared durable subscriber creation using the same subscription |
| * name but different topic than an existing subscriber can't be active on the |
| * same connection at the same time fails (can't change details in use). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubChangeOfTopic() throws Exception { |
| doSharedSubChangeOfDetailsTestImpl(true, true, false); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscriber creation using the same subscription |
| * name but different topic than an existing subscriber can't be active on the |
| * same connection at the same time fails (can't change details in use). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubChangeOfTopic() throws Exception { |
| doSharedSubChangeOfDetailsTestImpl(false, true, false); |
| } |
| |
| /** |
| * Verifies that a shared durable subscriber creation using the same subscription |
| * name but different selector than an existing subscriber can't be active on the |
| * same connection at the same time fails (can't change details in use). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubChangeOfSelector() throws Exception { |
| doSharedSubChangeOfDetailsTestImpl(true, false, true); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscriber creation using the same subscription |
| * name but different selector than an existing subscriber can't be active on the |
| * same connection at the same time fails (can't change details in use). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubChangeOfSelector() throws Exception { |
| doSharedSubChangeOfDetailsTestImpl(false, false, true); |
| } |
| |
| /** |
| * Verifies that a shared durable subscriber creation using the same subscription |
| * name but different topic and selector than an existing subscriber can't be active |
| * on the same connection at the same time fails (can't change details in use). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedDurableSubChangeOfTopicAndSelector() throws Exception { |
| doSharedSubChangeOfDetailsTestImpl(true, true, true); |
| } |
| |
| /** |
| * Verifies that a shared volatile subscriber creation using the same subscription |
| * name but different topic and selector than an existing subscriber can't be active |
| * on the same connection at the same time fails (can't change details in use). |
| * |
| * @throws Exception if an unexpected exception occurs |
| */ |
| @Test(timeout = 20000) |
| public void testSharedVolatileSubChangeOfTopicAndSelector() throws Exception { |
| doSharedSubChangeOfDetailsTestImpl(false, true, true); |
| } |
| |
| private void doSharedSubChangeOfDetailsTestImpl(boolean durable, boolean changeTopic, boolean changeSelector) throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| connection.start(); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName1 = "myTopic"; |
| String topicName2 = topicName1; |
| if (changeTopic) { |
| topicName2 += "2"; |
| } |
| |
| String noSelectorNull = null; |
| String selector2 = null; |
| if (changeSelector) { |
| selector2 = "someProperty = 2"; |
| } |
| |
| Topic dest1 = session.createTopic(topicName1); |
| Topic dest2 = session.createTopic(topicName2); |
| |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first shared receiver |
| if (durable) { |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName1, subscriptionName, durableLinkNameMatcher, true); |
| } else { |
| Matcher<?> volatileLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName1, subscriptionName, volatileLinkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber1; |
| if (durable) { |
| subscriber1 = session.createSharedDurableConsumer(dest1, subscriptionName, noSelectorNull); |
| } else { |
| subscriber1 = session.createSharedConsumer(dest1, subscriptionName, noSelectorNull); |
| } |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Now try to create another subscriber, same name but different topic. Should fail. No frames should be sent. |
| try { |
| if (durable) { |
| session.createSharedDurableConsumer(dest2, subscriptionName, selector2); |
| } else { |
| session.createSharedConsumer(dest2, subscriptionName, selector2); |
| } |
| |
| fail("Expected to fail due to attempting change of subscription details while subscriber is active"); |
| } catch (JMSException jmse) { |
| // Expected |
| } |
| |
| // Now close the first subscriber |
| if (durable) { |
| testPeer.expectDetach(false, true, false); |
| } else { |
| testPeer.expectDetach(true, true, true); |
| } |
| subscriber1.close(); |
| |
| // Now try a new subscriber again, with changed topic, it should succeed (note also the verified reuse of link names). |
| if (durable) { |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName2, subscriptionName, durableLinkNameMatcher, true); |
| } else { |
| Matcher<?> volatileLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName2, subscriptionName, volatileLinkNameMatcher, true); |
| } |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber2; |
| if (durable) { |
| subscriber2 = session.createSharedDurableConsumer(dest2, subscriptionName, selector2); |
| } else { |
| subscriber2 = session.createSharedConsumer(dest2, subscriptionName, selector2); |
| } |
| assertNotNull(subscriber2); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that subscriber cleanup occurs when the session it is on is remotely closed. |
| * |
| * @throws Exception if an unexpected error is encountered |
| */ |
| @Test(timeout = 20000) |
| public void testRemotelyEndSessionWithDurableSharedConsumer() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| |
| final CountDownLatch sessionClosed = new CountDownLatch(1); |
| ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { |
| @Override |
| public void onSessionClosed(Session session, Throwable exception) { |
| sessionClosed.countDown(); |
| } |
| }); |
| |
| // Create first session |
| testPeer.expectBegin(); |
| Session session1 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session1.createTopic(topicName); |
| |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first shared receiver on the first session |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber1 = session1.createSharedDurableConsumer(dest, subscriptionName); |
| |
| // Create second session |
| testPeer.expectBegin(); |
| Session session2 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Attach the second shared receiver on the second session |
| durableLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| |
| testPeer.expectLinkFlow(); |
| |
| // Then remotely end the session (and thus the subscriber along with it) after the flow is received. |
| testPeer.remotelyEndLastOpenedSession(true, 0, AmqpError.RESOURCE_LIMIT_EXCEEDED, "TestingRemoteClosure"); |
| |
| MessageConsumer subscriber2 = session2.createSharedDurableConsumer(dest, subscriptionName); |
| assertNotNull(subscriber2); |
| |
| assertTrue("Session closed callback didn't trigger", sessionClosed.await(5, TimeUnit.SECONDS)); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Now try to unsubscribe (using first session, still open). It should fail due to sub still |
| // being in use on the first session. No frames should be sent. |
| try { |
| session1.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| // Now close the first subscriber |
| testPeer.expectDetach(false, true, false); |
| subscriber1.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Try to unsubscribe again (using first session, still open), should now work. |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true); |
| |
| session1.unsubscribe(subscriptionName); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| } |
| } |
| |
| /** |
| * Verifies that subscriber cleanup occurs when the session it is on is locally closed. |
| * |
| * @throws Exception if an unexpected error is encountered |
| */ |
| @Test(timeout = 20000) |
| public void testLocallyEndSessionWithSharedConsumer() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| |
| final CountDownLatch sessionClosed = new CountDownLatch(1); |
| ((JmsConnection) connection).addConnectionListener(new JmsDefaultConnectionListener() { |
| @Override |
| public void onSessionClosed(Session session, Throwable exception) { |
| sessionClosed.countDown(); |
| } |
| }); |
| |
| // Create first session |
| testPeer.expectBegin(); |
| Session session1 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session1.createTopic(topicName); |
| |
| String subscriptionName = "mySubscription"; |
| |
| // Attach the first shared receiver on the first session |
| Matcher<?> durableLinkNameMatcher = equalTo(subscriptionName); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber1 = session1.createSharedDurableConsumer(dest, subscriptionName); |
| |
| // Create second session |
| testPeer.expectBegin(); |
| Session session2 = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Attach the second shared receiver on the second session |
| durableLinkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "2"); |
| testPeer.expectSharedDurableSubscriberAttach(topicName, subscriptionName, durableLinkNameMatcher, true); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber2 = session2.createSharedDurableConsumer(dest, subscriptionName); |
| assertNotNull(subscriber2); |
| |
| // Now close the second session (and thus the subscriber along with it). |
| testPeer.expectEnd(); |
| session2.close(); |
| |
| // Now try to unsubscribe (using first session, still open). It should fail due to sub still |
| // being in use on the first session. No frames should be sent. |
| try { |
| session1.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| // Now close the first subscriber |
| testPeer.expectDetach(false, true, false); |
| subscriber1.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Try to unsubscribe again (using first session, still open), should now work. |
| testPeer.expectDurableSubUnsubscribeNullSourceLookup(false, false, subscriptionName, topicName, true); |
| testPeer.expectDetach(true, true, true); |
| |
| session1.unsubscribe(subscriptionName); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| /** |
| * Verifies that subscription name passed is not allowed to have the subscription name |
| * delimiter used in the receiver link naming to separate the subscription name from |
| * a suffix used to ensure unique link names are used on a connection. |
| * |
| * @throws Exception if an unexpected error is encountered |
| */ |
| @Test(timeout = 20000) |
| public void testSubscriptionNameNotAllowedToHaveNameSeparator() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| // Establish connection |
| Connection connection = testFixture.establishConnecton(testPeer, serverCapabilities); |
| |
| // Create session |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Try the various methods that take subscription name with a name that |
| // contains the delimiter, they should fail. No frames should be sent. |
| String subscriptionName = "thisName|hasTheDelimiterAlready"; |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| |
| try { |
| session.createDurableSubscriber(dest, subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| try { |
| session.createDurableConsumer(dest, subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| try { |
| session.createSharedDurableConsumer(dest, subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| try { |
| session.createSharedConsumer(dest, subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| try { |
| session.unsubscribe(subscriptionName); |
| fail("Should have thrown a JMSException"); |
| } catch (JMSException ex) { |
| // Expected |
| } |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| |
| // Now close connection, should work. |
| testPeer.expectClose(); |
| connection.close(); |
| } |
| } |
| |
| // -------------------------------------- // |
| |
| @Test(timeout = 20000) |
| public void testSharedTopicSubscriberBehavesLikeNoClientIDWasSetWhenAwaitClientIdOptionIsFalse() throws Exception { |
| try (TestAmqpPeer testPeer = new TestAmqpPeer();) { |
| int serverPort = testPeer.getServerPort(); |
| |
| // Add server connection capability to indicate support for shared-subs |
| Symbol[] serverCapabilities = new Symbol[]{SHARED_SUBS}; |
| |
| testPeer.expectSaslAnonymous(); |
| testPeer.expectOpen(null, serverCapabilities); |
| testPeer.expectBegin(); |
| |
| ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:" + serverPort + "?jms.awaitClientID=false"); |
| Connection connection = factory.createConnection(); |
| |
| // Verify that all handlers complete, i.e. the awaitClientID=false option |
| // setting was effective in provoking the AMQP Open immediately even |
| // though it has no ClientID and we haven't used the Connection. |
| testPeer.waitForAllHandlersToComplete(3000); |
| |
| testPeer.expectBegin(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| String topicName = "myTopic"; |
| Topic dest = session.createTopic(topicName); |
| String subscriptionName = "mySubscription"; |
| |
| // Expect a shared subscription with 'global' capability and link name qualifier, |
| // since this connection does not have a ClientID set. |
| Matcher<?> linkNameMatcher = equalTo(subscriptionName + SUB_NAME_DELIMITER + "global-volatile1"); |
| testPeer.expectSharedVolatileSubscriberAttach(topicName, subscriptionName, linkNameMatcher, false); |
| testPeer.expectLinkFlow(); |
| |
| MessageConsumer subscriber = session.createSharedConsumer(dest, subscriptionName); |
| |
| testPeer.expectDetach(true, true, true); |
| subscriber.close(); |
| |
| testPeer.expectClose(); |
| connection.close(); |
| |
| testPeer.waitForAllHandlersToComplete(1000); |
| } |
| } |
| } |
