diff options
Diffstat (limited to 'qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java')
| -rw-r--r-- | qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java | 506 |
1 files changed, 506 insertions, 0 deletions
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java new file mode 100644 index 0000000000..989ac98747 --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java @@ -0,0 +1,506 @@ +/* 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.ct; + +import javax.jms.Connection; +import javax.jms.Message; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; +import javax.jms.Topic; +import javax.jms.TopicConnection; +import javax.jms.TopicConnectionFactory; +import javax.jms.TopicPublisher; +import javax.jms.TopicSession; +import javax.jms.TopicSubscriber; + +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.test.utils.QpidBrokerTestCase; + +/** + * Crash Recovery tests for durable subscription + * + */ +public class DurableSubscriberTest extends QpidBrokerTestCase +{ + private final String _topicName = "durableSubscriberTopic"; + + /** + * test strategy: + * create and register a durable subscriber then close it + * create a publisher and send a persistant message followed by a non persistant message + * crash and restart the broker + * recreate the durable subscriber and check that only the first message is received + */ + public void testDurSubRestoredAfterNonPersistentMessageSent() throws Exception + { + if (isBrokerStorePersistent() || !isBroker08()) + { + TopicConnectionFactory factory = getConnectionFactory(); + Topic topic = (Topic) getInitialContext().lookup(_topicName); + //create and register a durable subscriber then close it + TopicConnection durConnection = factory.createTopicConnection("guest", "guest"); + TopicSession durSession = durConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub1 = durSession.createDurableSubscriber(topic, "dursub"); + durConnection.start(); + durSub1.close(); + durSession.close(); + durConnection.stop(); + + //create a publisher and send a persistant message followed by a non persistant message + TopicConnection pubConnection = factory.createTopicConnection("guest", "guest"); + TopicSession pubSession = pubConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicPublisher publisher = pubSession.createPublisher(topic); + Message message = pubSession.createMessage(); + message.setIntProperty("count", 1); + publisher.publish(message, javax.jms.DeliveryMode.PERSISTENT, javax.jms.Message.DEFAULT_PRIORITY, + javax.jms.Message.DEFAULT_TIME_TO_LIVE); + message.setIntProperty("count", 2); + publisher.publish(message, javax.jms.DeliveryMode.NON_PERSISTENT, javax.jms.Message.DEFAULT_PRIORITY, + javax.jms.Message.DEFAULT_TIME_TO_LIVE); + publisher.close(); + pubSession.close(); + //now stop the server + try + { + restartBroker(); + } + catch (Exception e) + { + _logger.error("problems restarting broker: " + e); + throw e; + } + //now recreate the durable subscriber and check the received messages + factory = getConnectionFactory(); + topic = (Topic) getInitialContext().lookup(_topicName); + TopicConnection durConnection2 = factory.createTopicConnection("guest", "guest"); + TopicSession durSession2 = durConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub2 = durSession2.createDurableSubscriber(topic, "dursub"); + durConnection2.start(); + Message m1 = durSub2.receive(1000); + if (m1 == null) + { + assertTrue("testDurSubRestoredAfterNonPersistentMessageSent test failed. no message was returned", + false); + } + assertTrue("testDurSubRestoredAfterNonPersistentMessageSent test failed. Wrong message was returned.", + m1.getIntProperty("count") == 1); + durSession2.unsubscribe("dursub"); + durConnection2.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 + */ + public void testDurSubRestoresMessageSelector() throws Exception + { + if (isBrokerStorePersistent() || !isBroker08()) + { + TopicConnectionFactory factory = getConnectionFactory(); + Topic topic = (Topic) getInitialContext().lookup(_topicName); + //create and register a durable subscriber with a message selector and then close it + TopicConnection durConnection = factory.createTopicConnection("guest", "guest"); + TopicSession durSession = durConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub1 = durSession.createDurableSubscriber(topic, "dursub", "testprop='true'", false); + durConnection.start(); + durSub1.close(); + durSession.close(); + durConnection.stop(); + //now stop the server + try + { + restartBroker(); + } + catch (Exception e) + { + _logger.error("problems restarting broker: " + e); + throw e; + } + topic = (Topic) getInitialContext().lookup(_topicName); + factory = getConnectionFactory(); + TopicConnection pubConnection = factory.createTopicConnection("guest", "guest"); + TopicSession pubSession = pubConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicPublisher publisher = pubSession.createPublisher(topic); + for (int i = 0; i < 5; i++) + { + Message message = pubSession.createMessage(); + message.setStringProperty("testprop", "true"); + publisher.publish(message); + message = pubSession.createMessage(); + message.setStringProperty("testprop", "false"); + publisher.publish(message); + } + publisher.close(); + pubSession.close(); + + //now recreate the durable subscriber and check the received messages + TopicConnection durConnection2 = factory.createTopicConnection("guest", "guest"); + TopicSession durSession2 = durConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub2 = durSession2.createDurableSubscriber(topic, "dursub", "testprop='true'", false); + durConnection2.start(); + for (int i = 0; i < 5; i++) + { + Message message = durSub2.receive(1000); + if (message == null) + { + assertTrue("testDurSubRestoresMessageSelector test failed. no message was returned", false); + } + else + { + assertTrue("testDurSubRestoresMessageSelector test failed. message selector not reset", + message.getStringProperty("testprop").equals("true")); + } + } + durSession2.unsubscribe("dursub"); + durConnection2.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 + */ + public void testDurSubChangedToHaveSelectorThenRestart() throws Exception + { + if (! isBrokerStorePersistent()) + { + _logger.warn("Test skipped due to requirement of a persistent store"); + return; + } + + final String SUB_NAME=getTestQueueName(); + + TopicConnectionFactory factory = getConnectionFactory(); + Topic topic = (Topic) getInitialContext().lookup(_topicName); + + //create and register a durable subscriber then unsubscribe it + TopicConnection durConnection = factory.createTopicConnection("guest", "guest"); + TopicSession durSession = durConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub1 = durSession.createDurableSubscriber(topic, SUB_NAME); + durConnection.start(); + durSub1.close(); + durSession.unsubscribe(SUB_NAME); + durSession.close(); + durConnection.close(); + + //create and register a durable subscriber with a message selector and then close it + TopicConnection durConnection2 = factory.createTopicConnection("guest", "guest"); + TopicSession durSession2 = durConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub2 = durSession2.createDurableSubscriber(topic, SUB_NAME, "testprop='true'", false); + durConnection2.start(); + durSub2.close(); + durSession2.close(); + durConnection2.close(); + + //now restart the server + try + { + restartBroker(); + } + catch (Exception e) + { + _logger.error("problems restarting broker: " + e); + throw e; + } + + //send messages matching and not matching the selector + TopicConnection pubConnection = factory.createTopicConnection("guest", "guest"); + TopicSession pubSession = pubConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicPublisher publisher = pubSession.createPublisher(topic); + for (int i = 0; i < 5; i++) + { + Message message = pubSession.createMessage(); + message.setStringProperty("testprop", "true"); + publisher.publish(message); + message = pubSession.createMessage(); + message.setStringProperty("testprop", "false"); + publisher.publish(message); + } + publisher.close(); + pubSession.close(); + + //now recreate the durable subscriber with selector to check there are no exceptions generated + //and then verify the messages are received correctly + TopicConnection durConnection3 = (TopicConnection) factory.createConnection("guest", "guest"); + TopicSession durSession3 = (TopicSession) durConnection3.createSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub3 = durSession3.createDurableSubscriber(topic, SUB_NAME, "testprop='true'", false); + durConnection3.start(); + + for (int i = 0; i < 5; i++) + { + Message message = durSub3.receive(2000); + if (message == null) + { + fail("testDurSubChangedToHaveSelectorThenRestart test failed. Expected message " + i + " was not returned"); + } + else + { + assertTrue("testDurSubChangedToHaveSelectorThenRestart test failed. Got message not matching selector", + message.getStringProperty("testprop").equals("true")); + } + } + + durSub3.close(); + durSession3.unsubscribe(SUB_NAME); + durSession3.close(); + durConnection3.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 + */ + public void testDurSubChangedToNotHaveSelectorThenRestart() throws Exception + { + if (! isBrokerStorePersistent()) + { + _logger.warn("Test skipped due to requirement of a persistent store"); + return; + } + + final String SUB_NAME=getTestQueueName(); + + TopicConnectionFactory factory = getConnectionFactory(); + Topic topic = (Topic) getInitialContext().lookup(_topicName); + + //create and register a durable subscriber with selector then unsubscribe it + TopicConnection durConnection = factory.createTopicConnection("guest", "guest"); + TopicSession durSession = durConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub1 = durSession.createDurableSubscriber(topic, SUB_NAME, "testprop='true'", false); + durConnection.start(); + durSub1.close(); + durSession.unsubscribe(SUB_NAME); + durSession.close(); + durConnection.close(); + + //create and register a durable subscriber without the message selector and then close it + TopicConnection durConnection2 = factory.createTopicConnection("guest", "guest"); + TopicSession durSession2 = durConnection2.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub2 = durSession2.createDurableSubscriber(topic, SUB_NAME); + durConnection2.start(); + durSub2.close(); + durSession2.close(); + durConnection2.close(); + + //now restart the server + try + { + restartBroker(); + } + catch (Exception e) + { + _logger.error("problems restarting broker: " + e); + throw e; + } + + //send messages matching and not matching the original used selector + TopicConnection pubConnection = factory.createTopicConnection("guest", "guest"); + TopicSession pubSession = pubConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); + TopicPublisher publisher = pubSession.createPublisher(topic); + for (int i = 1; i <= 5; i++) + { + Message message = pubSession.createMessage(); + message.setStringProperty("testprop", "true"); + publisher.publish(message); + message = pubSession.createMessage(); + message.setStringProperty("testprop", "false"); + publisher.publish(message); + } + publisher.close(); + pubSession.close(); + + //now recreate the durable subscriber without selector to check there are no exceptions generated + //then verify ALL messages sent are received + TopicConnection durConnection3 = (TopicConnection) factory.createConnection("guest", "guest"); + TopicSession durSession3 = (TopicSession) durConnection3.createSession(false, Session.AUTO_ACKNOWLEDGE); + TopicSubscriber durSub3 = durSession3.createDurableSubscriber(topic, SUB_NAME); + durConnection3.start(); + + for (int i = 1; i <= 10; i++) + { + Message message = durSub3.receive(2000); + if (message == null) + { + fail("testDurSubChangedToNotHaveSelectorThenRestart test failed. Expected message " + i + " was not received"); + } + } + + durSub3.close(); + durSession3.unsubscribe(SUB_NAME); + durSession3.close(); + durConnection3.close(); + } + + + public void testResubscribeWithChangedSelectorAndRestart() throws Exception + { + if (! isBrokerStorePersistent()) + { + _logger.warn("Test skipped due to requirement of a persistent store"); + return; + } + + Connection conn = getConnection(); + conn.start(); + Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); + AMQTopic topic = new AMQTopic((AMQConnection) conn, "testResubscribeWithChangedSelectorAndRestart"); + 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 + TextMessage msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1"); + msg.setBooleanProperty("Match", true); + producer.send(msg); + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart2"); + msg.setBooleanProperty("Match", false); + producer.send(msg); + + Message rMsg = subA.receive(1000); + assertNotNull(rMsg); + assertEquals("Content was wrong", + "testResubscribeWithChangedSelectorAndRestart1", + ((TextMessage) rMsg).getText()); + + // Queue has no messages left + AMQQueue subQueueTmp = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorAndRestart"); + assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueueTmp)); + + rMsg = subA.receive(1000); + assertNull(rMsg); + + // Send another 1 matching message and 1 non-matching message + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1"); + msg.setBooleanProperty("Match", true); + producer.send(msg); + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart2"); + msg.setBooleanProperty("Match", false); + producer.send(msg); + + // Disconnect subscriber without receiving the message to + //leave it on the underlying queue + subA.close(); + + // Reconnect with new selector that matches B + TopicSubscriber subB = session.createDurableSubscriber(topic, + "testResubscribeWithChangedSelectorAndRestart", + "Match = false", false); + + //verify no messages are now present on the queue as changing selector should have issued + //an unsubscribe and thus deleted the previous durable backing queue for the subscription. + //check the dur sub's underlying queue now has msg count 0 + AMQQueue subQueue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorAndRestart"); + assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue)); + + // Check that new messages are received properly + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1"); + msg.setBooleanProperty("Match", true); + producer.send(msg); + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart2"); + msg.setBooleanProperty("Match", false); + producer.send(msg); + + rMsg = subB.receive(1000); + assertNotNull(rMsg); + assertEquals("Content was wrong", + "testResubscribeWithChangedSelectorAndRestart2", + ((TextMessage) rMsg).getText()); + + rMsg = subB.receive(1000); + assertNull(rMsg); + + //check the dur sub's underlying queue now has msg count 0 + subQueue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorAndRestart"); + assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue)); + + //now restart the server + try + { + restartBroker(); + } + catch (Exception e) + { + _logger.error("problems restarting broker: " + e); + throw e; + } + + // Reconnect to broker + Connection connection = getConnectionFactory().createConnection("guest", "guest"); + connection.start(); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + topic = new AMQTopic((AMQConnection) connection, "testResubscribeWithChangedSelectorAndRestart"); + producer = session.createProducer(topic); + + //verify no messages now present on the queue after we restart the broker + //check the dur sub's underlying queue now has msg count 0 + subQueue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorAndRestart"); + assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue)); + + // Reconnect with new selector that matches B + TopicSubscriber subC = session.createDurableSubscriber(topic, + "testResubscribeWithChangedSelectorAndRestart", + "Match = False", false); + + // Check that new messages are still sent and recieved properly + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1"); + msg.setBooleanProperty("Match", true); + producer.send(msg); + msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart2"); + msg.setBooleanProperty("Match", false); + producer.send(msg); + + //check the dur sub's underlying queue now has msg count 1 + subQueue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorAndRestart"); + assertEquals("Msg count should be 1", 1, ((AMQSession<?, ?>) session).getQueueDepth(subQueue)); + + rMsg = subC.receive(1000); + assertNotNull(rMsg); + assertEquals("Content was wrong", + "testResubscribeWithChangedSelectorAndRestart2", + ((TextMessage) rMsg).getText()); + + rMsg = subC.receive(1000); + assertNull(rMsg); + + session.unsubscribe("testResubscribeWithChangedSelectorAndRestart"); + + subC.close(); + session.close(); + connection.close(); + } +} + |