| /* |
| * |
| * 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.test.unit.topic; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.client.AMQConnection; |
| import org.apache.qpid.client.AMQQueue; |
| import org.apache.qpid.client.AMQSession; |
| import org.apache.qpid.client.AMQTopic; |
| import org.apache.qpid.management.common.JMXConnnectionFactory; |
| import org.apache.qpid.test.utils.QpidBrokerTestCase; |
| |
| import javax.jms.Connection; |
| import javax.jms.InvalidDestinationException; |
| import javax.jms.InvalidSelectorException; |
| 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.TopicSubscriber; |
| import javax.management.MBeanServerConnection; |
| import javax.management.ObjectName; |
| import javax.management.remote.JMXConnector; |
| import java.io.IOException; |
| import java.util.Set; |
| |
| /** |
| * @todo Code to check that a consumer gets only one particular method could be factored into a re-usable method (as |
| * a static on a base test helper class, e.g. TestUtils. |
| * |
| * @todo Code to create test end-points using session per connection, or all sessions on one connection, to be factored |
| * out to make creating this test variation simpler. Want to make this variation available through LocalCircuit, |
| * driven by the test model. |
| */ |
| public class DurableSubscriptionTest extends QpidBrokerTestCase |
| { |
| private static final Logger _logger = LoggerFactory.getLogger(DurableSubscriptionTest.class); |
| |
| /** Timeout for receive() if we are expecting a message */ |
| private static final long POSITIVE_RECEIVE_TIMEOUT = 2000; |
| |
| /** Timeout for receive() if we are not expecting a message */ |
| private static final long NEGATIVE_RECEIVE_TIMEOUT = 1000; |
| |
| private JMXConnector _jmxc; |
| private MBeanServerConnection _mbsc; |
| private static final String USER = "admin"; |
| private static final String PASSWORD = "admin"; |
| private boolean _jmxConnected; |
| |
| public void setUp() throws Exception |
| { |
| setConfigurationProperty("management.enabled", "true"); |
| _jmxConnected=false; |
| super.setUp(); |
| } |
| |
| public void tearDown() throws Exception |
| { |
| if(_jmxConnected) |
| { |
| try |
| { |
| _jmxc.close(); |
| } |
| catch (IOException e) |
| { |
| e.printStackTrace(); |
| } |
| } |
| |
| super.tearDown(); |
| } |
| |
| public void testUnsubscribe() throws Exception |
| { |
| AMQConnection con = (AMQConnection) getConnection("guest", "guest"); |
| AMQTopic topic = new AMQTopic(con, "MyDurableSubscriptionTestTopic"); |
| _logger.info("Create Session 1"); |
| Session session1 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE); |
| _logger.info("Create Consumer on Session 1"); |
| MessageConsumer consumer1 = session1.createConsumer(topic); |
| _logger.info("Create Producer on Session 1"); |
| MessageProducer producer = session1.createProducer(topic); |
| |
| _logger.info("Create Session 2"); |
| Session session2 = con.createSession(false, AMQSession.NO_ACKNOWLEDGE); |
| _logger.info("Create Durable Subscriber on Session 2"); |
| TopicSubscriber consumer2 = session2.createDurableSubscriber(topic, "MySubscription"); |
| |
| _logger.info("Starting connection"); |
| con.start(); |
| |
| _logger.info("Producer sending message A"); |
| producer.send(session1.createTextMessage("A")); |
| |
| //check the dur sub's underlying queue now has msg count 1 |
| AMQQueue subQueue = new AMQQueue("amq.topic", "clientid" + ":" + "MySubscription"); |
| assertEquals("Msg count should be 1", 1, ((AMQSession<?, ?>) session1).getQueueDepth(subQueue, true)); |
| |
| Message msg; |
| _logger.info("Receive message on consumer 1:expecting A"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should have been received",msg); |
| assertEquals("A", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| msg = consumer1.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertEquals(null, msg); |
| |
| _logger.info("Receive message on consumer 2:expecting A"); |
| msg = consumer2.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should have been received",msg); |
| assertEquals("A", ((TextMessage) msg).getText()); |
| msg = consumer2.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| assertEquals(null, msg); |
| |
| //check the dur sub's underlying queue now has msg count 0 |
| assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session2).getQueueDepth(subQueue, true)); |
| |
| consumer2.close(); |
| _logger.info("Unsubscribe session2/consumer2"); |
| session2.unsubscribe("MySubscription"); |
| |
| ((AMQSession<?, ?>) session2).sync(); |
| |
| if(isJavaBroker()) |
| { |
| //Verify that the queue was deleted by querying for its JMX MBean |
| _jmxc = JMXConnnectionFactory.getJMXConnection(5000, "127.0.0.1", |
| getManagementPort(getPort()), USER, PASSWORD); |
| |
| _jmxConnected = true; |
| _mbsc = _jmxc.getMBeanServerConnection(); |
| |
| //must replace the occurrence of ':' in queue name with '-' |
| String queueObjectNameText = "clientid" + "-" + "MySubscription"; |
| |
| ObjectName objName = new ObjectName("org.apache.qpid:type=VirtualHost.Queue,name=" |
| + queueObjectNameText + ",*"); |
| |
| Set<ObjectName> objectInstances = _mbsc.queryNames(objName, null); |
| |
| if(objectInstances.size() != 0) |
| { |
| fail("Queue MBean was found. Expected queue to have been deleted"); |
| } |
| else |
| { |
| _logger.info("Underlying dueue for the durable subscription was confirmed deleted."); |
| } |
| } |
| |
| //verify unsubscribing the durable subscriber did not affect the non-durable one |
| _logger.info("Producer sending message B"); |
| producer.send(session1.createTextMessage("B")); |
| |
| _logger.info("Receive message on consumer 1 :expecting B"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should have been received",msg); |
| assertEquals("B", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| msg = consumer1.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertEquals(null, msg); |
| |
| _logger.info("Close connection"); |
| con.close(); |
| } |
| |
| public void testDurabilityNOACK() throws Exception |
| { |
| durabilityImpl(AMQSession.NO_ACKNOWLEDGE, false); |
| } |
| |
| public void testDurabilityAUTOACK() throws Exception |
| { |
| durabilityImpl(Session.AUTO_ACKNOWLEDGE, false); |
| } |
| |
| public void testDurabilityAUTOACKwithRestartIfPersistent() throws Exception |
| { |
| if(!isBrokerStorePersistent()) |
| { |
| System.out.println("The broker store is not persistent, skipping this test."); |
| return; |
| } |
| |
| durabilityImpl(Session.AUTO_ACKNOWLEDGE, true); |
| } |
| |
| public void testDurabilityNOACKSessionPerConnection() throws Exception |
| { |
| durabilityImplSessionPerConnection(AMQSession.NO_ACKNOWLEDGE); |
| } |
| |
| public void testDurabilityAUTOACKSessionPerConnection() throws Exception |
| { |
| durabilityImplSessionPerConnection(Session.AUTO_ACKNOWLEDGE); |
| } |
| |
| private void durabilityImpl(int ackMode, boolean restartBroker) throws Exception |
| { |
| AMQConnection con = (AMQConnection) getConnection("guest", "guest"); |
| AMQTopic topic = new AMQTopic(con, "MyTopic"); |
| Session session1 = con.createSession(false, ackMode); |
| MessageConsumer consumer1 = session1.createConsumer(topic); |
| |
| Session sessionProd = con.createSession(false, ackMode); |
| MessageProducer producer = sessionProd.createProducer(topic); |
| |
| Session session2 = con.createSession(false, ackMode); |
| TopicSubscriber consumer2 = session2.createDurableSubscriber(topic, "MySubscription"); |
| |
| con.start(); |
| |
| //send message A and check both consumers receive |
| producer.send(session1.createTextMessage("A")); |
| |
| Message msg; |
| _logger.info("Receive message on consumer 1 :expecting A"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should have been received",msg); |
| assertEquals("A", ((TextMessage) msg).getText()); |
| msg = consumer1.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertEquals(null, msg); |
| |
| _logger.info("Receive message on consumer 2 :expecting A"); |
| msg = consumer2.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should have been received",msg); |
| assertEquals("A", ((TextMessage) msg).getText()); |
| msg = consumer2.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertEquals(null, msg); |
| |
| //send message B, receive with consumer 1, and disconnect consumer 2 to leave the message behind (if not NO_ACK) |
| producer.send(session1.createTextMessage("B")); |
| |
| _logger.info("Receive message on consumer 1 :expecting B"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Consumer 1 should get message 'B'.", msg); |
| assertEquals("Incorrect Message received on consumer1.", "B", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| msg = consumer1.receive(500); |
| assertNull("There should be no more messages for consumption on consumer1.", msg); |
| |
| consumer2.close(); |
| session2.close(); |
| |
| //Send message C, then connect consumer 3 to durable subscription and get |
| //message B if not using NO_ACK, then receive C with consumer 1 and 3 |
| producer.send(session1.createTextMessage("C")); |
| |
| Session session3 = con.createSession(false, ackMode); |
| MessageConsumer consumer3 = session3.createDurableSubscriber(topic, "MySubscription"); |
| |
| if(ackMode == AMQSession.NO_ACKNOWLEDGE) |
| { |
| //Do nothing if NO_ACK was used, as prefetch means the message was dropped |
| //when we didn't call receive() to get it before closing consumer 2 |
| } |
| else |
| { |
| _logger.info("Receive message on consumer 3 :expecting B"); |
| msg = consumer3.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Consumer 3 should get message 'B'.", msg); |
| assertEquals("Incorrect Message received on consumer3.", "B", ((TextMessage) msg).getText()); |
| } |
| |
| _logger.info("Receive message on consumer 1 :expecting C"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Consumer 1 should get message 'C'.", msg); |
| assertEquals("Incorrect Message received on consumer1.", "C", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| msg = consumer1.receive(500); |
| assertNull("There should be no more messages for consumption on consumer1.", msg); |
| |
| _logger.info("Receive message on consumer 3 :expecting C"); |
| msg = consumer3.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Consumer 3 should get message 'C'.", msg); |
| assertEquals("Incorrect Message received on consumer3.", "C", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 3 :expecting null"); |
| msg = consumer3.receive(500); |
| assertNull("There should be no more messages for consumption on consumer3.", msg); |
| |
| consumer1.close(); |
| consumer3.close(); |
| |
| session3.unsubscribe("MySubscription"); |
| |
| con.close(); |
| |
| if(restartBroker) |
| { |
| try |
| { |
| restartBroker(); |
| } |
| catch (Exception e) |
| { |
| fail("Error restarting the broker"); |
| } |
| } |
| } |
| |
| private void durabilityImplSessionPerConnection(int ackMode) throws Exception |
| { |
| Message msg; |
| // Create producer. |
| AMQConnection con0 = (AMQConnection) getConnection("guest", "guest"); |
| con0.start(); |
| Session session0 = con0.createSession(false, ackMode); |
| |
| AMQTopic topic = new AMQTopic(con0, "MyTopic"); |
| |
| Session sessionProd = con0.createSession(false, ackMode); |
| MessageProducer producer = sessionProd.createProducer(topic); |
| |
| // Create consumer 1. |
| AMQConnection con1 = (AMQConnection) getConnection("guest", "guest"); |
| con1.start(); |
| Session session1 = con1.createSession(false, ackMode); |
| |
| MessageConsumer consumer1 = session1.createConsumer(topic); |
| |
| // Create consumer 2. |
| AMQConnection con2 = (AMQConnection) getConnection("guest", "guest"); |
| con2.start(); |
| Session session2 = con2.createSession(false, ackMode); |
| |
| TopicSubscriber consumer2 = session2.createDurableSubscriber(topic, "MySubscription"); |
| |
| // Send message and check that both consumers get it and only it. |
| producer.send(session0.createTextMessage("A")); |
| |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should be available", msg); |
| assertEquals("Message Text doesn't match", "A", ((TextMessage) msg).getText()); |
| msg = consumer1.receive(500); |
| assertNull("There should be no more messages for consumption on consumer1.", msg); |
| |
| msg = consumer2.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Message should have been received",msg); |
| assertEquals("Consumer 2 should also received the first msg.", "A", ((TextMessage) msg).getText()); |
| msg = consumer2.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("There should be no more messages for consumption on consumer2.", msg); |
| |
| // Send message and receive on consumer 1. |
| producer.send(session0.createTextMessage("B")); |
| |
| _logger.info("Receive message on consumer 1 :expecting B"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertEquals("B", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| msg = consumer1.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertEquals(null, msg); |
| |
| // Detach the durable subscriber. |
| consumer2.close(); |
| session2.close(); |
| con2.close(); |
| |
| // Send message C and receive on consumer 1 |
| producer.send(session0.createTextMessage("C")); |
| |
| _logger.info("Receive message on consumer 1 :expecting C"); |
| msg = consumer1.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertEquals("C", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 1 :expecting null"); |
| msg = consumer1.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertEquals(null, msg); |
| |
| // Re-attach a new consumer to the durable subscription, and check that it gets message B it left (if not NO_ACK) |
| // and also gets message C sent after it was disconnected. |
| AMQConnection con3 = (AMQConnection) getConnection("guest", "guest"); |
| con3.start(); |
| Session session3 = con3.createSession(false, ackMode); |
| |
| TopicSubscriber consumer3 = session3.createDurableSubscriber(topic, "MySubscription"); |
| |
| if(ackMode == AMQSession.NO_ACKNOWLEDGE) |
| { |
| //Do nothing if NO_ACK was used, as prefetch means the message was dropped |
| //when we didn't call receive() to get it before closing consumer 2 |
| } |
| else |
| { |
| _logger.info("Receive message on consumer 3 :expecting B"); |
| msg = consumer3.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull(msg); |
| assertEquals("B", ((TextMessage) msg).getText()); |
| } |
| |
| _logger.info("Receive message on consumer 3 :expecting C"); |
| msg = consumer3.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Consumer 3 should get message 'C'.", msg); |
| assertEquals("Incorrect Message recevied on consumer3.", "C", ((TextMessage) msg).getText()); |
| _logger.info("Receive message on consumer 3 :expecting null"); |
| msg = consumer3.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("There should be no more messages for consumption on consumer3.", msg); |
| |
| consumer1.close(); |
| consumer3.close(); |
| |
| session3.unsubscribe("MySubscription"); |
| |
| con0.close(); |
| con1.close(); |
| con3.close(); |
| } |
| |
| /** |
| * This tests the fix for QPID-1085 |
| * Creates a durable subscriber with an invalid selector, checks that the |
| * exception is thrown correctly and that the subscription is not created. |
| * @throws Exception |
| */ |
| public void testDurableWithInvalidSelector() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "MyTestDurableWithInvalidSelectorTopic"); |
| MessageProducer producer = session.createProducer(topic); |
| producer.send(session.createTextMessage("testDurableWithInvalidSelector1")); |
| try |
| { |
| TopicSubscriber deadSubscriber = session.createDurableSubscriber(topic, "testDurableWithInvalidSelectorSub", |
| "=TEST 'test", true); |
| assertNull("Subscriber should not have been created", deadSubscriber); |
| } |
| catch (JMSException e) |
| { |
| assertTrue("Wrong type of exception thrown", e instanceof InvalidSelectorException); |
| } |
| TopicSubscriber liveSubscriber = session.createDurableSubscriber(topic, "testDurableWithInvalidSelectorSub"); |
| assertNotNull("Subscriber should have been created", liveSubscriber); |
| |
| producer.send(session.createTextMessage("testDurableWithInvalidSelector2")); |
| |
| Message msg = liveSubscriber.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull ("Message should have been received", msg); |
| assertEquals ("testDurableWithInvalidSelector2", ((TextMessage) msg).getText()); |
| assertNull("Should not receive subsequent message", liveSubscriber.receive(200)); |
| liveSubscriber.close(); |
| session.unsubscribe("testDurableWithInvalidSelectorSub"); |
| } |
| |
| /** |
| * This tests the fix for QPID-1085 |
| * Creates a durable subscriber with an invalid destination, checks that the |
| * exception is thrown correctly and that the subscription is not created. |
| * @throws Exception |
| */ |
| public void testDurableWithInvalidDestination() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "testDurableWithInvalidDestinationTopic"); |
| try |
| { |
| TopicSubscriber deadSubscriber = session.createDurableSubscriber(null, "testDurableWithInvalidDestinationsub"); |
| assertNull("Subscriber should not have been created", deadSubscriber); |
| } |
| catch (InvalidDestinationException e) |
| { |
| // This was expected |
| } |
| MessageProducer producer = session.createProducer(topic); |
| producer.send(session.createTextMessage("testDurableWithInvalidSelector1")); |
| |
| TopicSubscriber liveSubscriber = session.createDurableSubscriber(topic, "testDurableWithInvalidDestinationsub"); |
| assertNotNull("Subscriber should have been created", liveSubscriber); |
| |
| producer.send(session.createTextMessage("testDurableWithInvalidSelector2")); |
| Message msg = liveSubscriber.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull ("Message should have been received", msg); |
| assertEquals ("testDurableWithInvalidSelector2", ((TextMessage) msg).getText()); |
| assertNull("Should not receive subsequent message", liveSubscriber.receive(200)); |
| |
| session.unsubscribe("testDurableWithInvalidDestinationsub"); |
| } |
| |
| /** |
| * Creates a durable subscription with a selector, then changes that selector on resubscription |
| * <p> |
| * QPID-1202, QPID-2418 |
| */ |
| public void testResubscribeWithChangedSelector() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "testResubscribeWithChangedSelector"); |
| MessageProducer producer = session.createProducer(topic); |
| |
| // Create durable subscriber that matches A |
| TopicSubscriber subA = session.createDurableSubscriber(topic, |
| "testResubscribeWithChangedSelector", |
| "Match = True", false); |
| |
| // Send 1 matching message and 1 non-matching message |
| sendMatchingAndNonMatchingMessage(session, producer); |
| |
| Message rMsg = subA.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNotNull(rMsg); |
| assertEquals("Content was wrong", |
| "testResubscribeWithChangedSelector1", |
| ((TextMessage) rMsg).getText()); |
| |
| rMsg = subA.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull(rMsg); |
| |
| // Disconnect subscriber |
| subA.close(); |
| |
| // Reconnect with new selector that matches B |
| TopicSubscriber subB = session.createDurableSubscriber(topic, |
| "testResubscribeWithChangedSelector","Match = False", false); |
| |
| //verify no messages are now recieved. |
| rMsg = subB.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Should not have received message as the selector was changed", rMsg); |
| |
| // Check that new messages are received properly |
| sendMatchingAndNonMatchingMessage(session, producer); |
| rMsg = subB.receive(POSITIVE_RECEIVE_TIMEOUT); |
| |
| assertNotNull("Message should have been received", rMsg); |
| assertEquals("Content was wrong", |
| "testResubscribeWithChangedSelector2", |
| ((TextMessage) rMsg).getText()); |
| |
| |
| rMsg = subB.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Message should not have been received",rMsg); |
| session.unsubscribe("testResubscribeWithChangedSelector"); |
| } |
| |
| public void testDurableSubscribeWithTemporaryTopic() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session ssn = conn.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 |
| } |
| } |
| |
| private void sendMatchingAndNonMatchingMessage(Session session, MessageProducer producer) throws JMSException |
| { |
| TextMessage msg = session.createTextMessage("testResubscribeWithChangedSelector1"); |
| msg.setBooleanProperty("Match", true); |
| producer.send(msg); |
| msg = session.createTextMessage("testResubscribeWithChangedSelector2"); |
| msg.setBooleanProperty("Match", false); |
| producer.send(msg); |
| } |
| |
| |
| /** |
| * create and register a durable subscriber with a message selector and then close it |
| * create a publisher and send 5 right messages and 5 wrong messages |
| * create another durable subscriber with the same selector and name |
| * check messages are still there |
| * <p> |
| * QPID-2418 |
| */ |
| public void testDurSubSameMessageSelector() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(true, Session.SESSION_TRANSACTED); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "sameMessageSelector"); |
| |
| //create and register a durable subscriber with a message selector and then close it |
| TopicSubscriber subOne = session.createDurableSubscriber(topic, "sameMessageSelector", "testprop = TRUE", false); |
| subOne.close(); |
| |
| MessageProducer producer = session.createProducer(topic); |
| for (int i = 0; i < 5; i++) |
| { |
| Message message = session.createMessage(); |
| message.setBooleanProperty("testprop", true); |
| producer.send(message); |
| message = session.createMessage(); |
| message.setBooleanProperty("testprop", false); |
| producer.send(message); |
| } |
| session.commit(); |
| producer.close(); |
| |
| // should be 5 or 10 messages on queue now |
| // (5 for the java broker due to use of server side selectors, and 10 for the cpp broker due to client side selectors only) |
| AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "sameMessageSelector"); |
| assertEquals("Queue depth is wrong", isJavaBroker() ? 5 : 10, ((AMQSession<?, ?>) session).getQueueDepth(queue, true)); |
| |
| // now recreate the durable subscriber and check the received messages |
| TopicSubscriber subTwo = session.createDurableSubscriber(topic, "sameMessageSelector", "testprop = TRUE", false); |
| |
| for (int i = 0; i < 5; i++) |
| { |
| Message message = subTwo.receive(1000); |
| if (message == null) |
| { |
| fail("sameMessageSelector test failed. no message was returned"); |
| } |
| else |
| { |
| assertEquals("sameMessageSelector test failed. message selector not reset", |
| "true", message.getStringProperty("testprop")); |
| } |
| } |
| |
| session.commit(); |
| |
| // Check queue has no messages |
| if (isJavaBroker()) |
| { |
| assertEquals("Queue should be empty", 0, ((AMQSession<?, ?>) session).getQueueDepth(queue)); |
| } |
| else |
| { |
| assertTrue("At most the queue should have only 1 message", ((AMQSession<?, ?>) session).getQueueDepth(queue) <= 1); |
| } |
| |
| // Unsubscribe |
| session.unsubscribe("sameMessageSelector"); |
| |
| conn.close(); |
| } |
| |
| /** |
| * <ul> |
| * <li>create and register a durable subscriber with a message selector |
| * <li>create another durable subscriber with a different selector and same name |
| * <li>check first subscriber is now closed |
| * <li>create a publisher and send messages |
| * <li>check messages are received correctly |
| * </ul> |
| * <p> |
| * QPID-2418 |
| */ |
| public void testResubscribeWithChangedSelectorNoClose() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "testResubscribeWithChangedSelectorNoClose"); |
| |
| // Create durable subscriber that matches A |
| TopicSubscriber subA = session.createDurableSubscriber(topic, |
| "testResubscribeWithChangedSelectorNoClose", |
| "Match = True", false); |
| |
| // Reconnect with new selector that matches B |
| TopicSubscriber subB = session.createDurableSubscriber(topic, |
| "testResubscribeWithChangedSelectorNoClose", |
| "Match = false", false); |
| |
| // First subscription has been closed |
| try |
| { |
| subA.receive(1000); |
| fail("First subscription was not closed"); |
| } |
| catch (Exception e) |
| { |
| e.printStackTrace(); |
| } |
| |
| conn.stop(); |
| |
| // Send 1 matching message and 1 non-matching message |
| MessageProducer producer = session.createProducer(topic); |
| TextMessage msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1"); |
| msg.setBooleanProperty("Match", true); |
| producer.send(msg); |
| msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart2"); |
| msg.setBooleanProperty("Match", false); |
| producer.send(msg); |
| |
| // should be 1 or 2 messages on queue now |
| // (1 for the java broker due to use of server side selectors, and 2 for the cpp broker due to client side selectors only) |
| AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorNoClose"); |
| assertEquals("Queue depth is wrong", isJavaBroker() ? 1 : 2, ((AMQSession<?, ?>) session).getQueueDepth(queue, true)); |
| |
| conn.start(); |
| |
| Message rMsg = subB.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull(rMsg); |
| assertEquals("Content was wrong", |
| "testResubscribeWithChangedSelectorAndRestart2", |
| ((TextMessage) rMsg).getText()); |
| |
| rMsg = subB.receive(1000); |
| assertNull(rMsg); |
| |
| // Check queue has no messages |
| assertEquals("Queue should be empty", 0, ((AMQSession<?, ?>) session).getQueueDepth(queue, true)); |
| |
| conn.close(); |
| } |
| |
| /** |
| * <ul> |
| * <li>create and register a durable subscriber with no message selector |
| * <li>create another durable subscriber with a selector and same name |
| * <li>check first subscriber is now closed |
| * <li>create a publisher and send messages |
| * <li>check messages are recieved correctly |
| * </ul> |
| * <p> |
| * QPID-2418 |
| */ |
| public void testDurSubAddMessageSelectorNoClose() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "subscriptionName"); |
| |
| // create and register a durable subscriber with no message selector |
| TopicSubscriber subOne = session.createDurableSubscriber(topic, "subscriptionName", null, false); |
| |
| // now create a durable subscriber with a selector |
| TopicSubscriber subTwo = session.createDurableSubscriber(topic, "subscriptionName", "testprop = TRUE", false); |
| |
| // First subscription has been closed |
| try |
| { |
| subOne.receive(1000); |
| fail("First subscription was not closed"); |
| } |
| catch (Exception e) |
| { |
| e.printStackTrace(); |
| } |
| |
| conn.stop(); |
| |
| // Send 1 matching message and 1 non-matching message |
| MessageProducer producer = session.createProducer(topic); |
| TextMessage msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1"); |
| msg.setBooleanProperty("testprop", true); |
| producer.send(msg); |
| msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart2"); |
| msg.setBooleanProperty("testprop", false); |
| producer.send(msg); |
| |
| // should be 1 or 2 messages on queue now |
| // (1 for the java broker due to use of server side selectors, and 2 for the cpp broker due to client side selectors only) |
| AMQQueue queue = new AMQQueue("amq.topic", "clientid" + ":" + "subscriptionName"); |
| assertEquals("Queue depth is wrong", isJavaBroker() ? 1 : 2, ((AMQSession<?, ?>) session).getQueueDepth(queue, true)); |
| |
| conn.start(); |
| |
| Message rMsg = subTwo.receive(POSITIVE_RECEIVE_TIMEOUT); |
| assertNotNull(rMsg); |
| assertEquals("Content was wrong", |
| "testResubscribeWithChangedSelectorAndRestart1", |
| ((TextMessage) rMsg).getText()); |
| |
| rMsg = subTwo.receive(1000); |
| assertNull(rMsg); |
| |
| // Check queue has no messages |
| assertEquals("Queue should be empty", 0, ((AMQSession<?, ?>) session).getQueueDepth(queue, true)); |
| |
| conn.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 |
| */ |
| public void testDurSubNoSelectorResubscribeNoClose() throws Exception |
| { |
| Connection conn = getConnection(); |
| conn.start(); |
| Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| AMQTopic topic = new AMQTopic((AMQConnection) conn, "subscriptionName"); |
| |
| // 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 (Exception e) |
| { |
| e.printStackTrace(); |
| } |
| } |
| |
| /** |
| * Tests that a subscriber created on a same <i>session</i> as producer with |
| * no local true does not receive messages. |
| */ |
| public void testNoLocalOnSameSession() throws Exception |
| { |
| Connection connection = getConnection(); |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Topic topic = session.createTopic(getTestQueueName()); |
| MessageProducer producer = session.createProducer(topic); |
| TopicSubscriber subscriber = null; |
| try |
| { |
| subscriber = session.createDurableSubscriber(topic, getTestName(), null, true); |
| connection.start(); |
| |
| producer.send(createNextMessage(session, 1)); |
| |
| Message m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Unexpected message received", m); |
| } |
| finally |
| { |
| session.unsubscribe(getTestName()); |
| } |
| } |
| |
| |
| /** |
| * Tests that a subscriber created on a same <i>connection</i> but separate |
| * <i>sessionM</i> as producer with no local true does not receive messages. |
| */ |
| public void testNoLocalOnSameConnection() throws Exception |
| { |
| Connection connection = getConnection(); |
| |
| Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Topic topic = consumerSession.createTopic(getTestQueueName()); |
| MessageProducer producer = producerSession.createProducer(topic); |
| |
| TopicSubscriber subscriber = null; |
| try |
| { |
| subscriber = consumerSession.createDurableSubscriber(topic, getTestName(), null, true); |
| connection.start(); |
| |
| producer.send(createNextMessage(producerSession, 1)); |
| |
| Message m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Unexpected message received", m); |
| } |
| finally |
| { |
| consumerSession.unsubscribe(getTestName()); |
| } |
| } |
| |
| /** |
| * Tests that if no-local is in use, that the messages are delivered when |
| * the client reconnects. |
| * |
| * Currently fails on the Java Broker due to QPID-3605. |
| */ |
| public void testNoLocalMessagesNotDeliveredAfterReconnection() throws Exception |
| { |
| Connection connection = getConnection(); |
| |
| Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Session producerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Topic topic = consumerSession.createTopic(getTestQueueName()); |
| MessageProducer producer = producerSession.createProducer(topic); |
| |
| TopicSubscriber subscriber = null; |
| try |
| { |
| subscriber = consumerSession.createDurableSubscriber(topic, getTestName(), null, true); |
| connection.start(); |
| |
| producer.send(createNextMessage(producerSession, 1)); |
| |
| Message m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Unexpected message received", m); |
| |
| connection.close(); |
| |
| connection = getConnection(); |
| |
| consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| subscriber = consumerSession.createDurableSubscriber(topic, getTestName(), null, true); |
| connection.start(); |
| m = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Message should not be received on a new connection", m); |
| } |
| finally |
| { |
| consumerSession.unsubscribe(getTestName()); |
| } |
| } |
| |
| /** |
| * 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. |
| */ |
| public void testNoLocalSubscriberAndSubscriberOnSeparateConnection() throws Exception |
| { |
| Connection noLocalConnection = getConnection(); |
| Connection connection = getConnection(); |
| |
| String noLocalSubId1 = getTestName() + "subId1"; |
| String subId = getTestName() + "subId2"; |
| |
| Session noLocalSession = noLocalConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Topic noLocalTopic = noLocalSession.createTopic(getTestQueueName()); |
| |
| Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| Topic topic = consumerSession.createTopic(getTestQueueName()); |
| |
| TopicSubscriber noLocalSubscriber = null; |
| TopicSubscriber subscriber = null; |
| try |
| { |
| MessageProducer producer = noLocalSession.createProducer(noLocalTopic); |
| noLocalSubscriber = noLocalSession.createDurableSubscriber(noLocalTopic, noLocalSubId1, null, true); |
| subscriber = consumerSession.createDurableSubscriber(topic, subId, null, true); |
| noLocalConnection.start(); |
| connection.start(); |
| |
| producer.send(createNextMessage(noLocalSession, 1)); |
| |
| Message m1 = noLocalSubscriber.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNull("Subscriber on nolocal connection should not receive message", m1); |
| |
| Message m2 = subscriber.receive(NEGATIVE_RECEIVE_TIMEOUT); |
| assertNotNull("Subscriber on non-nolocal connection should receive message", m2); |
| } |
| finally |
| { |
| noLocalSession.unsubscribe(noLocalSubId1); |
| consumerSession.unsubscribe(subId); |
| } |
| } |
| } |
