| /* 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.systests.jms_1_1.topic; |
| |
| import static org.hamcrest.CoreMatchers.equalTo; |
| import static org.hamcrest.CoreMatchers.is; |
| import static org.hamcrest.CoreMatchers.not; |
| import static org.junit.jupiter.api.Assertions.assertEquals; |
| import static org.junit.jupiter.api.Assertions.assertTrue; |
| import static org.junit.jupiter.api.Assertions.fail; |
| import static org.junit.jupiter.api.Assumptions.assumeTrue; |
| |
| import java.util.Arrays; |
| import java.util.List; |
| |
| import javax.jms.Connection; |
| import javax.jms.InvalidDestinationException; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.MessageConsumer; |
| import javax.jms.MessageProducer; |
| import javax.jms.Session; |
| import javax.jms.TextMessage; |
| import javax.jms.Topic; |
| import javax.jms.TopicConnection; |
| import javax.jms.TopicPublisher; |
| import javax.jms.TopicSession; |
| import javax.jms.TopicSubscriber; |
| |
| import org.junit.jupiter.api.Test; |
| |
| import org.apache.qpid.server.model.Protocol; |
| import org.apache.qpid.systests.JmsTestBase; |
| |
| public class DurableSubscriptionTest extends JmsTestBase |
| { |
| @Test |
| public void publishedMessagesAreSavedAfterSubscriberClose() throws Exception |
| { |
| Topic topic = createTopic(getTestName()); |
| String subscriptionName = getTestName() + "_sub"; |
| String clientId = "testClientId"; |
| |
| TopicConnection connection = (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| MessageProducer producer = producerSession.createProducer(topic); |
| |
| Session durableSubscriberSession = connection.createSession(true, Session.SESSION_TRANSACTED); |
| TopicSubscriber durableSubscriber = |
| durableSubscriberSession.createDurableSubscriber(topic, subscriptionName); |
| |
| connection.start(); |
| |
| producer.send(producerSession.createTextMessage("A")); |
| |
| Message message = durableSubscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("A", ((TextMessage) message).getText()); |
| |
| durableSubscriberSession.commit(); |
| |
| producer.send(producerSession.createTextMessage("B")); |
| |
| message = durableSubscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("B", ((TextMessage) message).getText()); |
| |
| durableSubscriberSession.rollback(); |
| |
| durableSubscriber.close(); |
| durableSubscriberSession.close(); |
| |
| producer.send(producerSession.createTextMessage("C")); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| |
| if (getBrokerAdmin().supportsRestart()) |
| { |
| getBrokerAdmin().restart(); |
| } |
| |
| TopicConnection connection2 = (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| connection2.start(); |
| final Session durableSubscriberSession = connection2.createSession(true, Session.SESSION_TRANSACTED); |
| final TopicSubscriber durableSubscriber = |
| durableSubscriberSession.createDurableSubscriber(topic, subscriptionName); |
| |
| final List<String> expectedMessages = Arrays.asList("B", "C"); |
| for (String expectedMessageText : expectedMessages) |
| { |
| final Message message = durableSubscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals(expectedMessageText, ((TextMessage) message).getText()); |
| |
| durableSubscriberSession.commit(); |
| } |
| |
| durableSubscriber.close(); |
| durableSubscriberSession.unsubscribe(subscriptionName); |
| } |
| finally |
| { |
| connection2.close(); |
| } |
| } |
| |
| @Test |
| public void testUnsubscribe() throws Exception |
| { |
| Topic topic = createTopic(getTestName()); |
| String subscriptionName = getTestName() + "_sub"; |
| String clientId = "clientId"; |
| int numberOfQueuesBeforeTest = getQueueCount(); |
| |
| Connection connection = getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| Session durableSubscriberSession = connection.createSession(true, Session.SESSION_TRANSACTED); |
| Session nonDurableSubscriberSession = connection.createSession(true, Session.SESSION_TRANSACTED); |
| Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| MessageConsumer subscriber = nonDurableSubscriberSession.createConsumer(topic); |
| MessageProducer producer = producerSession.createProducer(topic); |
| TopicSubscriber durableSubscriber = |
| durableSubscriberSession.createDurableSubscriber(topic, subscriptionName); |
| |
| connection.start(); |
| producer.send(nonDurableSubscriberSession.createTextMessage("A")); |
| |
| Message message = subscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("A", ((TextMessage) message).getText()); |
| |
| message = durableSubscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("A", ((TextMessage) message).getText()); |
| |
| nonDurableSubscriberSession.commit(); |
| durableSubscriberSession.commit(); |
| |
| durableSubscriber.close(); |
| durableSubscriberSession.unsubscribe(subscriptionName); |
| |
| producer.send(nonDurableSubscriberSession.createTextMessage("B")); |
| |
| Session durableSubscriberSession2 = connection.createSession(true, Session.SESSION_TRANSACTED); |
| TopicSubscriber durableSubscriber2 = |
| durableSubscriberSession2.createDurableSubscriber(topic, subscriptionName); |
| |
| producer.send(nonDurableSubscriberSession.createTextMessage("C")); |
| |
| message = subscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("B", ((TextMessage) message).getText()); |
| |
| message = subscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("C", ((TextMessage) message).getText()); |
| |
| message = durableSubscriber2.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage); |
| assertEquals("C", ((TextMessage) message).getText()); |
| |
| nonDurableSubscriberSession.commit(); |
| durableSubscriberSession2.commit(); |
| |
| assertEquals(0, getTotalDepthOfQueuesMessages(), "Message count should be 0"); |
| |
| durableSubscriber2.close(); |
| durableSubscriberSession2.unsubscribe(subscriptionName); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| |
| int numberOfQueuesAfterTest = getQueueCount(); |
| assertEquals(numberOfQueuesBeforeTest, numberOfQueuesAfterTest, "Unexpected number of queues"); |
| } |
| |
| @Test |
| public void unsubscribeTwice() throws Exception |
| { |
| Topic topic = createTopic(getTestName()); |
| Connection connection = getConnection(); |
| String subscriptionName = getTestName() + "_sub"; |
| try |
| { |
| |
| Session subscriberSession = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = subscriberSession.createDurableSubscriber(topic, subscriptionName); |
| MessageProducer publisher = subscriberSession.createProducer(topic); |
| |
| connection.start(); |
| |
| publisher.send(subscriberSession.createTextMessage("Test")); |
| subscriberSession.commit(); |
| |
| Message message = subscriber.receive(getReceiveTimeout()); |
| assertTrue(message instanceof TextMessage, "TextMessage should be received"); |
| assertEquals("Test", ((TextMessage)message).getText(), "Unexpected message"); |
| subscriberSession.commit(); |
| subscriber.close(); |
| subscriberSession.unsubscribe(subscriptionName); |
| |
| try |
| { |
| subscriberSession.unsubscribe(subscriptionName); |
| fail("expected InvalidDestinationException when unsubscribing from unknown subscription"); |
| } |
| catch (InvalidDestinationException e) |
| { |
| // PASS |
| } |
| catch (Exception e) |
| { |
| fail("expected InvalidDestinationException when unsubscribing from unknown subscription, got: " + e); |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| /** |
| * <ul> |
| * <li>create and register a durable subscriber with no message selector |
| * <li>try to create another durable with the same name, should fail |
| * </ul> |
| * <p> |
| * QPID-2418 |
| */ |
| @Test |
| public void multipleSubscribersWithTheSameName() throws Exception |
| { |
| String subscriptionName = getTestName() + "_sub"; |
| Topic topic = createTopic(subscriptionName); |
| Connection conn = getConnection(); |
| try |
| { |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // create and register a durable subscriber with no message selector |
| session.createDurableSubscriber(topic, subscriptionName, null, false); |
| |
| // try to recreate the durable subscriber |
| try |
| { |
| session.createDurableSubscriber(topic, subscriptionName, null, false); |
| fail("Subscription should not have been created"); |
| } |
| catch (JMSException e) |
| { |
| // pass |
| } |
| } |
| finally |
| { |
| conn.close(); |
| } |
| } |
| |
| @Test |
| public void testDurableSubscribeWithTemporaryTopic() throws Exception |
| { |
| assumeTrue(is(not(equalTo(Protocol.AMQP_1_0))).matches(getProtocol()), "Not investigated - fails on AMQP 1.0"); |
| Connection connection = getConnection(); |
| try |
| { |
| connection.start(); |
| Session ssn = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Topic topic = ssn.createTemporaryTopic(); |
| try |
| { |
| ssn.createDurableSubscriber(topic, "test"); |
| fail("expected InvalidDestinationException"); |
| } |
| catch (InvalidDestinationException ex) |
| { |
| // this is expected |
| } |
| try |
| { |
| ssn.createDurableSubscriber(topic, "test", null, false); |
| fail("expected InvalidDestinationException"); |
| } |
| catch (InvalidDestinationException ex) |
| { |
| // this is expected |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| @Test |
| public void noLocalMessagesNotDelivered() throws Exception |
| { |
| String noLocalSubscriptionName = getTestName() + "_no_local_sub"; |
| Topic topic = createTopic(getTestName()); |
| String clientId = "testClientId"; |
| |
| Connection publishingConnection = getConnectionBuilder().setClientId("publishingConnection").build(); |
| try |
| { |
| Session session = publishingConnection.createSession(true, Session.SESSION_TRANSACTED); |
| MessageProducer sessionProducer = session.createProducer(topic); |
| |
| Connection noLocalConnection = getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| Session noLocalSession = noLocalConnection.createSession(true, Session.SESSION_TRANSACTED); |
| MessageProducer noLocalSessionProducer = noLocalSession.createProducer(topic); |
| |
| TopicSubscriber noLocalSubscriber = |
| noLocalSession.createDurableSubscriber(topic, noLocalSubscriptionName, null, true); |
| noLocalConnection.start(); |
| publishingConnection.start(); |
| |
| noLocalSessionProducer.send(noLocalSession.createTextMessage("Message1")); |
| noLocalSession.commit(); |
| sessionProducer.send(session.createTextMessage("Message2")); |
| session.commit(); |
| |
| Message durableSubscriberMessage = noLocalSubscriber.receive(getReceiveTimeout()); |
| assertTrue(durableSubscriberMessage instanceof TextMessage); |
| assertEquals("Message2", ((TextMessage) durableSubscriberMessage).getText(), |
| "Unexpected local message received"); |
| noLocalSession.commit(); |
| } |
| finally |
| { |
| noLocalConnection.close(); |
| } |
| |
| Connection noLocalConnection2 = getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| Session noLocalSession = noLocalConnection2.createSession(true, Session.SESSION_TRANSACTED); |
| noLocalConnection2.start(); |
| TopicSubscriber noLocalSubscriber = |
| noLocalSession.createDurableSubscriber(topic, noLocalSubscriptionName, null, true); |
| try |
| { |
| sessionProducer.send(session.createTextMessage("Message3")); |
| session.commit(); |
| |
| final Message durableSubscriberMessage = noLocalSubscriber.receive(getReceiveTimeout()); |
| assertTrue(durableSubscriberMessage instanceof TextMessage); |
| assertEquals("Message3", ((TextMessage) durableSubscriberMessage).getText(), |
| "Unexpected local message received"); |
| noLocalSession.commit(); |
| } |
| finally |
| { |
| noLocalSubscriber.close(); |
| noLocalSession.unsubscribe(noLocalSubscriptionName); |
| } |
| } |
| finally |
| { |
| noLocalConnection2.close(); |
| } |
| } |
| finally |
| { |
| publishingConnection.close(); |
| } |
| } |
| |
| /** |
| * Tests that messages are delivered normally to a subscriber on a separate connection despite |
| * the use of durable subscriber with no-local on the first connection. |
| */ |
| @Test |
| public void testNoLocalSubscriberAndSubscriberOnSeparateConnection() throws Exception |
| { |
| String noLocalSubscriptionName = getTestName() + "_no_local_sub"; |
| String subscriobtionName = getTestName() + "_sub"; |
| Topic topic = createTopic(getTestName()); |
| final String clientId = "clientId"; |
| |
| Connection noLocalConnection = getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| Connection connection = getConnection(); |
| try |
| { |
| Session noLocalSession = noLocalConnection.createSession(true, Session.SESSION_TRANSACTED); |
| Session session = connection.createSession(true, Session.SESSION_TRANSACTED); |
| |
| MessageProducer noLocalSessionProducer = noLocalSession.createProducer(topic); |
| MessageProducer sessionProducer = session.createProducer(topic); |
| |
| try |
| { |
| TopicSubscriber noLocalSubscriber = |
| noLocalSession.createDurableSubscriber(topic, noLocalSubscriptionName, null, true); |
| TopicSubscriber subscriber = session.createDurableSubscriber(topic, subscriobtionName, null, false); |
| noLocalConnection.start(); |
| connection.start(); |
| |
| noLocalSessionProducer.send(noLocalSession.createTextMessage("Message1")); |
| noLocalSession.commit(); |
| sessionProducer.send(session.createTextMessage("Message2")); |
| sessionProducer.send(session.createTextMessage("Message3")); |
| session.commit(); |
| |
| Message durableSubscriberMessage = noLocalSubscriber.receive(getReceiveTimeout()); |
| assertTrue(durableSubscriberMessage instanceof TextMessage); |
| assertEquals("Message2", ((TextMessage) durableSubscriberMessage).getText(), |
| "Unexpected local message received"); |
| noLocalSession.commit(); |
| |
| Message nonDurableSubscriberMessage = subscriber.receive(getReceiveTimeout()); |
| assertTrue(nonDurableSubscriberMessage instanceof TextMessage); |
| assertEquals("Message1", ((TextMessage) nonDurableSubscriberMessage).getText(), |
| "Unexpected message received"); |
| |
| session.commit(); |
| noLocalSubscriber.close(); |
| subscriber.close(); |
| } |
| finally |
| { |
| noLocalSession.unsubscribe(noLocalSubscriptionName); |
| session.unsubscribe(subscriobtionName); |
| } |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| finally |
| { |
| noLocalConnection.close(); |
| } |
| } |
| |
| |
| @Test |
| public void testResubscribeWithChangedNoLocal() throws Exception |
| { |
| assumeTrue(is(equalTo(Protocol.AMQP_1_0)).matches(getProtocol()), "QPID-8068"); |
| String subscriptionName = getTestName() + "_sub"; |
| Topic topic = createTopic(getTestName()); |
| String clientId = "testClientId"; |
| Connection connection = getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| Session session = connection.createSession(true, Session.SESSION_TRANSACTED); |
| TopicSubscriber durableSubscriber = |
| session.createDurableSubscriber(topic, subscriptionName, null, false); |
| |
| MessageProducer producer = session.createProducer(topic); |
| producer.send(session.createTextMessage("A")); |
| producer.send(session.createTextMessage("B")); |
| session.commit(); |
| |
| connection.start(); |
| |
| Message receivedMessage = durableSubscriber.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage, "TextMessage should be received"); |
| assertEquals("A", ((TextMessage)receivedMessage).getText(), "Unexpected message received"); |
| |
| session.commit(); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| |
| connection = getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| connection.start(); |
| |
| Session session2 = connection.createSession(true, Session.SESSION_TRANSACTED); |
| TopicSubscriber noLocalSubscriber2 = session2.createDurableSubscriber(topic, subscriptionName, null, true); |
| |
| Connection secondConnection = getConnectionBuilder().setClientId("secondConnection").build(); |
| try |
| { |
| Session secondSession = secondConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| MessageProducer secondProducer = secondSession.createProducer(topic); |
| secondProducer.send(secondSession.createTextMessage("C")); |
| } |
| finally |
| { |
| secondConnection.close(); |
| } |
| |
| Message noLocalSubscriberMessage = noLocalSubscriber2.receive(getReceiveTimeout()); |
| assertTrue(noLocalSubscriberMessage instanceof TextMessage, "TextMessage should be received"); |
| assertEquals("C", ((TextMessage)noLocalSubscriberMessage).getText(), |
| "Unexpected message received"); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| } |
| |
| /** |
| * create and register a durable subscriber with a message selector and then close it |
| * crash the broker |
| * create a publisher and send 5 right messages and 5 wrong messages |
| * recreate the durable subscriber and check we receive the 5 expected messages |
| */ |
| @Test |
| public void testMessageSelectorRecoveredOnBrokerRestart() throws Exception |
| { |
| assumeTrue(getBrokerAdmin().supportsRestart()); |
| |
| final Topic topic = createTopic(getTestName()); |
| |
| String clientId = "testClientId"; |
| String subscriptionName = getTestName() + "_sub"; |
| TopicConnection subscriberConnection = |
| (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| TopicSession session = subscriberConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = |
| session.createDurableSubscriber(topic, subscriptionName, "testprop='true'", false); |
| subscriberConnection.start(); |
| subscriber.close(); |
| session.close(); |
| } |
| finally |
| { |
| subscriberConnection.close(); |
| } |
| |
| getBrokerAdmin().restart(); |
| |
| TopicConnection connection = (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicPublisher publisher = session.createPublisher(topic); |
| for (int i = 0; i < 10; i++) |
| { |
| Message message = session.createMessage(); |
| message.setStringProperty("testprop", String.valueOf(i % 2 == 0)); |
| publisher.publish(message); |
| } |
| publisher.close(); |
| session.close(); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| |
| TopicConnection subscriberConnection2 = |
| (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| TopicSession session = subscriberConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = |
| session.createDurableSubscriber(topic, subscriptionName, "testprop='true'", false); |
| subscriberConnection2.start(); |
| for (int i = 0; i < 5; i++) |
| { |
| Message message = subscriber.receive(1000); |
| if (message == null) |
| { |
| fail(String.format("Message '%d' was received", i)); |
| } |
| else |
| { |
| assertEquals("true", message.getStringProperty("testprop"), |
| String.format("Received message %d with not matching selector", i)); |
| } |
| } |
| subscriber.close(); |
| session.unsubscribe(subscriptionName); |
| } |
| finally |
| { |
| subscriberConnection2.close(); |
| } |
| } |
| |
| /** |
| * create and register a durable subscriber without a message selector and then unsubscribe it |
| * create and register a durable subscriber with a message selector and then close it |
| * restart the broker |
| * send matching and non matching messages |
| * recreate and register the durable subscriber with a message selector |
| * verify only the matching messages are received |
| */ |
| @Test |
| public void testChangeSubscriberToHaveSelector() throws Exception |
| { |
| assumeTrue(getBrokerAdmin().supportsRestart()); |
| |
| final String subscriptionName = getTestName() + "_sub"; |
| Topic topic = createTopic(getTestName()); |
| String testClientId = "testClientId"; |
| |
| TopicConnection subscriberConnection = |
| (TopicConnection) getConnectionBuilder().setClientId(testClientId).build(); |
| try |
| { |
| TopicSession session = subscriberConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = session.createDurableSubscriber(topic, subscriptionName); |
| |
| TopicPublisher publisher = session.createPublisher(topic); |
| publisher.send(session.createTextMessage("Message1")); |
| publisher.send(session.createTextMessage("Message2")); |
| |
| subscriberConnection.start(); |
| Message receivedMessage = subscriber.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals("Message1", ((TextMessage) receivedMessage).getText(), |
| "Unexpected message content"); |
| |
| subscriber.close(); |
| session.close(); |
| } |
| finally |
| { |
| subscriberConnection.close(); |
| } |
| |
| //create and register a durable subscriber with a message selector and then close it |
| TopicConnection subscriberConnection2 = |
| (TopicConnection) getConnectionBuilder().setClientId(testClientId).build(); |
| try |
| { |
| TopicSession session = subscriberConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = |
| session.createDurableSubscriber(topic, subscriptionName, "testprop='true'", false); |
| |
| TopicPublisher publisher = session.createPublisher(topic); |
| TextMessage message = session.createTextMessage("Message3"); |
| message.setStringProperty("testprop", "false"); |
| publisher.send(message); |
| message = session.createTextMessage("Message4"); |
| message.setStringProperty("testprop", "true"); |
| publisher.send(message); |
| |
| subscriberConnection2.start(); |
| |
| Message receivedMessage = subscriber.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals("Message4", ((TextMessage) receivedMessage).getText(), |
| "Unexpected message content"); |
| |
| subscriber.close(); |
| session.close(); |
| } |
| finally |
| { |
| subscriberConnection2.close(); |
| } |
| |
| getBrokerAdmin().restart(); |
| |
| TopicConnection publisherConnection = getTopicConnection(); |
| try |
| { |
| TopicSession session = publisherConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicPublisher publisher = session.createPublisher(topic); |
| for (int i = 0; i < 10; i++) |
| { |
| Message message = session.createMessage(); |
| message.setStringProperty("testprop", String.valueOf(i % 2 == 0)); |
| publisher.publish(message); |
| } |
| publisher.close(); |
| session.close(); |
| } |
| finally |
| { |
| publisherConnection.close(); |
| } |
| |
| TopicConnection subscriberConnection3 = |
| (TopicConnection) getConnectionBuilder().setClientId(testClientId).build(); |
| try |
| { |
| TopicSession session = (TopicSession) subscriberConnection3.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = |
| session.createDurableSubscriber(topic, subscriptionName, "testprop='true'", false); |
| subscriberConnection3.start(); |
| |
| for (int i = 0; i < 5; i++) |
| { |
| Message message = subscriber.receive(2000); |
| if (message == null) |
| { |
| fail(String.format("Message '%d' was not received", i)); |
| } |
| else |
| { |
| assertEquals("true", message.getStringProperty("testprop"), |
| String.format("Received message %d with not matching selector", i)); |
| } |
| } |
| |
| subscriber.close(); |
| session.unsubscribe(subscriptionName); |
| session.close(); |
| } |
| finally |
| { |
| subscriberConnection3.close(); |
| } |
| } |
| |
| /** |
| * create and register a durable subscriber with a message selector and then unsubscribe it |
| * create and register a durable subscriber without a message selector and then close it |
| * restart the broker |
| * send matching and non matching messages |
| * recreate and register the durable subscriber without a message selector |
| * verify ALL the sent messages are received |
| */ |
| @Test |
| public void testChangeSubscriberToHaveNoSelector() throws Exception |
| { |
| assumeTrue(getBrokerAdmin().supportsRestart()); |
| |
| final String subscriptionName = getTestName() + "_sub"; |
| Topic topic = createTopic(getTestName()); |
| String clientId = "testClientId"; |
| |
| //create and register a durable subscriber with selector then unsubscribe it |
| TopicConnection durConnection = (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| TopicSession session = durConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = |
| session.createDurableSubscriber(topic, subscriptionName, "testprop='true'", false); |
| |
| TopicPublisher publisher = session.createPublisher(topic); |
| TextMessage message = session.createTextMessage("Messag1"); |
| message.setStringProperty("testprop", "false"); |
| publisher.send(message); |
| message = session.createTextMessage("Message2"); |
| message.setStringProperty("testprop", "true"); |
| publisher.send(message); |
| |
| message = session.createTextMessage("Message3"); |
| message.setStringProperty("testprop", "true"); |
| publisher.send(message); |
| |
| durConnection.start(); |
| |
| Message receivedMessage = subscriber.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals("Message2", ((TextMessage) receivedMessage).getText(), |
| "Unexpected message content"); |
| |
| subscriber.close(); |
| session.close(); |
| } |
| finally |
| { |
| durConnection.close(); |
| } |
| |
| //create and register a durable subscriber without the message selector and then close it |
| TopicConnection subscriberConnection2 = |
| (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| TopicSession session = subscriberConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = session.createDurableSubscriber(topic, subscriptionName); |
| subscriberConnection2.start(); |
| subscriber.close(); |
| session.close(); |
| } |
| finally |
| { |
| subscriberConnection2.close(); |
| } |
| |
| //send messages matching and not matching the original used selector |
| TopicConnection publisherConnection = getTopicConnection(); |
| try |
| { |
| TopicSession session = publisherConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicPublisher publisher = session.createPublisher(topic); |
| for (int i = 1; i <= 10; i++) |
| { |
| Message message = session.createMessage(); |
| message.setStringProperty("testprop", String.valueOf(i % 2 == 0)); |
| publisher.publish(message); |
| } |
| publisher.close(); |
| session.close(); |
| } |
| finally |
| { |
| publisherConnection.close(); |
| } |
| |
| getBrokerAdmin().restart(); |
| |
| TopicConnection subscriberConnection3 = |
| (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| TopicSession session = (TopicSession) subscriberConnection3.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| TopicSubscriber subscriber = session.createDurableSubscriber(topic, subscriptionName); |
| subscriberConnection3.start(); |
| |
| for (int i = 1; i <= 10; i++) |
| { |
| Message message = subscriber.receive(2000); |
| if (message == null) |
| { |
| fail(String.format("Message %d was not received", i)); |
| } |
| } |
| |
| subscriber.close(); |
| session.unsubscribe(subscriptionName); |
| session.close(); |
| } |
| finally |
| { |
| subscriberConnection3.close(); |
| } |
| } |
| |
| @Test |
| public void testResubscribeWithChangedSelector() throws Exception |
| { |
| assumeTrue(getBrokerAdmin().supportsRestart()); |
| |
| String subscriptionName = getTestName() + "_sub"; |
| Topic topic = createTopic(getTestName()); |
| String clientId = "testClientId"; |
| |
| TopicConnection connection = (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| connection.start(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| MessageProducer producer = session.createProducer(topic); |
| |
| // Create durable subscriber that matches A |
| TopicSubscriber subscriberA = |
| session.createDurableSubscriber(topic, subscriptionName, "Match = True", false); |
| |
| // Send 1 non-matching message and 1 matching message |
| TextMessage message = session.createTextMessage("Message1"); |
| message.setBooleanProperty("Match", false); |
| producer.send(message); |
| message = session.createTextMessage("Message2"); |
| message.setBooleanProperty("Match", true); |
| producer.send(message); |
| |
| Message receivedMessage = subscriberA.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals("Message2", ((TextMessage) receivedMessage).getText(), "Unexpected message content"); |
| |
| // Send another 1 matching message and 1 non-matching message |
| message = session.createTextMessage("Message3"); |
| message.setBooleanProperty("Match", true); |
| producer.send(message); |
| message = session.createTextMessage("Message4"); |
| message.setBooleanProperty("Match", false); |
| producer.send(message); |
| |
| // Disconnect subscriber without receiving the message to |
| //leave it on the underlying queue |
| subscriberA.close(); |
| |
| // Reconnect with new selector that matches B |
| TopicSubscriber subscriberB = session.createDurableSubscriber(topic, |
| subscriptionName, |
| "Match = False", false); |
| |
| // Check that new messages are received properly |
| message = session.createTextMessage("Message5"); |
| message.setBooleanProperty("Match", true); |
| producer.send(message); |
| message = session.createTextMessage("Message6"); |
| message.setBooleanProperty("Match", false); |
| producer.send(message); |
| |
| // changing the selector should have cleared the queue so we expect message 6 instead of message 4 |
| receivedMessage = subscriberB.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals("Message6", ((TextMessage) receivedMessage).getText(), |
| "Unexpected message content"); |
| |
| // publish a message to be consumed after restart |
| message = session.createTextMessage("Message7"); |
| message.setBooleanProperty("Match", true); |
| producer.send(message); |
| message = session.createTextMessage("Message8"); |
| message.setBooleanProperty("Match", false); |
| producer.send(message); |
| session.close(); |
| } |
| finally |
| { |
| connection.close(); |
| } |
| |
| //now restart the server |
| getBrokerAdmin().restart(); |
| |
| // Reconnect to broker |
| TopicConnection connection2 = (TopicConnection) getConnectionBuilder().setClientId(clientId).build(); |
| try |
| { |
| connection2.start(); |
| Session session = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| |
| // Reconnect with new selector that matches B |
| TopicSubscriber subscriberC = |
| session.createDurableSubscriber(topic, subscriptionName, "Match = False", false); |
| |
| //check the dur sub's underlying queue now has msg count 1 |
| Message receivedMessage = subscriberC.receive(getReceiveTimeout()); |
| assertTrue(receivedMessage instanceof TextMessage); |
| assertEquals("Message8", ((TextMessage) receivedMessage).getText(), "Unexpected message content"); |
| |
| subscriberC.close(); |
| session.unsubscribe(subscriptionName); |
| |
| session.close(); |
| } |
| finally |
| { |
| connection2.close(); |
| } |
| } |
| } |
