You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2017/01/24 10:14:48 UTC
svn commit: r1780052 [3/3] - in /qpid/java/trunk:
broker-core/src/main/java/org/apache/qpid/server/model/
broker-core/src/main/java/org/apache/qpid/server/queue/
broker-core/src/main/java/org/apache/qpid/server/transport/
broker-core/src/main/java/org/...
Added: qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/subscription/SharedSubscriptionTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/subscription/SharedSubscriptionTest.java?rev=1780052&view=auto
==============================================================================
--- qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/subscription/SharedSubscriptionTest.java (added)
+++ qpid/java/trunk/systests/qpid-systests-jms_2.0/src/test/java/org/apache/qpid/systests/jms_2_0/subscription/SharedSubscriptionTest.java Tue Jan 24 10:14:48 2017
@@ -0,0 +1,307 @@
+/*
+ * 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_2_0.subscription;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.apache.qpid.systest.rest.RestTestHelper;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+public class SharedSubscriptionTest extends QpidBrokerTestCase
+{
+
+ private RestTestHelper _restTestHelper;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ getDefaultBrokerConfiguration().addHttpManagementConfiguration();
+ super.setUp();
+ _restTestHelper = new RestTestHelper(getDefaultBroker().getHttpPort());
+ }
+
+ public void testSharedNonDurableSubscription() throws Exception
+ {
+ Connection connection = getConnectionWithPrefetch(0);
+
+ Session publishingSession = connection.createSession();
+ Session subscriber1Session = connection.createSession();
+ Session subscriber2Session = connection.createSession();
+
+ String topicName = getTestName();
+ Topic topic = publishingSession.createTopic("amq.direct/" + topicName);
+
+ MessageConsumer consumer1 = subscriber1Session.createSharedConsumer(topic, "subscription");
+ MessageConsumer consumer2 = subscriber2Session.createSharedConsumer(topic, "subscription");
+
+ MessageProducer producer = publishingSession.createProducer(topic);
+
+ sendMessage(publishingSession, topic, 2);
+
+ connection.start();
+
+ Message message1 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message2 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull("Message 1 was not received", message1);
+ assertNotNull("Message 2 was not received", message2);
+
+ assertEquals("Unexpected index for message 1", 0, message1.getIntProperty(INDEX));
+ assertEquals("Unexpected index for message 2", 1, message2.getIntProperty(INDEX));
+
+ Message message3 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message4 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNull("Unexpected message received by first shared consumer", message3);
+ assertNull("Unexpected message received by second shared consumer", message4);
+ }
+
+ public void testSharedDurableSubscription() throws Exception
+ {
+ Connection connection = getConnectionWithPrefetch(0);
+
+ Session publishingSession = connection.createSession();
+ Session subscriber1Session = connection.createSession();
+ Session subscriber2Session = connection.createSession();
+
+ String topicName = getTestName();
+ Topic topic = publishingSession.createTopic("amq.direct/" + topicName);
+
+ MessageConsumer consumer1 = subscriber1Session.createSharedDurableConsumer(topic, "subscription");
+ MessageConsumer consumer2 = subscriber2Session.createSharedDurableConsumer(topic, "subscription");
+
+ MessageProducer producer = publishingSession.createProducer(topic);
+
+ sendMessage(publishingSession, topic, 4);
+
+ connection.start();
+
+ Message message1 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message2 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull("Message 1 was not received", message1);
+ assertNotNull("Message 2 was not received", message2);
+
+ assertEquals("Unexpected index for message 1", 0, message1.getIntProperty(INDEX));
+ assertEquals("Unexpected index for message 2", 1, message2.getIntProperty(INDEX));
+
+ connection.close();
+
+ if (isBrokerStorePersistent())
+ {
+ restartDefaultBroker();
+ }
+
+ connection = getConnectionWithPrefetch(0);
+ subscriber1Session = connection.createSession();
+ subscriber2Session = connection.createSession();
+
+ consumer1 = subscriber1Session.createSharedDurableConsumer(topic, "subscription");
+ consumer2 = subscriber2Session.createSharedDurableConsumer(topic, "subscription");
+
+ connection.start();
+
+ Message message3 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message4 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull("Message 3 was not received", message3);
+ assertNotNull("Message 4 was not received", message4);
+
+ assertEquals("Unexpected index for message 3", 2, message3.getIntProperty(INDEX));
+ assertEquals("Unexpected index for message 4", 3, message4.getIntProperty(INDEX));
+
+ Message message5 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message6 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNull("Unexpected message received by first shared consumer", message5);
+ assertNull("Unexpected message received by second shared consumer", message6);
+
+ }
+
+ public void testUnsubscribe() throws Exception
+ {
+ Connection connection = getConnectionWithPrefetch(0);
+ Session session = connection.createSession();
+
+ connection.start();
+
+ String topicName = getTestName();
+ Topic topic = session.createTopic("amq.direct/" + topicName);
+ String subscriptionName = "testSharedSubscription";
+ MessageConsumer consumer = session.createSharedDurableConsumer(topic, subscriptionName);
+
+ Map<String, Object> statistics = _restTestHelper.getJsonAsMap("virtualhost/test/test/getStatistics?statistics=[\"queueCount\"]");
+ int numberOfQueuesBeforeUnsubscribe = (int) statistics.get("queueCount");
+ assertEquals("Unexpected number of Queues", 1, numberOfQueuesBeforeUnsubscribe);
+
+ consumer.close();
+ session.close();
+ connection.close();
+
+ if (isBrokerStorePersistent())
+ {
+ restartDefaultBroker();
+ }
+
+ connection = getConnectionWithPrefetch(0);
+ session = connection.createSession();
+ session.unsubscribe(subscriptionName);
+
+ statistics = _restTestHelper.getJsonAsMap("virtualhost/test/test/getStatistics?statistics=[\"queueCount\"]");
+ int numberOfQueuesAfterUnsubscribe = (int) statistics.get("queueCount");
+ assertEquals("Queue should be deleted", 0, numberOfQueuesAfterUnsubscribe);
+ }
+
+ public void testDurableSharedAndNonDurableSharedCanUseTheSameSubscriptionName() throws Exception
+ {
+ Connection connection = getConnectionWithPrefetch(0);
+
+ Session publishingSession = connection.createSession();
+ Session subscriberSession = connection.createSession();
+
+ String topicName = getTestName();
+ Topic topic = publishingSession.createTopic("amq.direct/" + topicName);
+ MessageConsumer consumer1 = subscriberSession.createSharedDurableConsumer(topic, "testSharedSubscription");
+ MessageConsumer consumer2 = subscriberSession.createSharedConsumer(topic, "testSharedSubscription");
+ connection.start();
+
+ sendMessage(publishingSession, topic, 1);
+
+ Message message1 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message2 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull("Message 1 was not received", message1);
+ assertNotNull("Message 2 was not received", message2);
+
+ assertEquals("Unexpected index for message 1", 0, message1.getIntProperty(INDEX));
+ assertEquals("Unexpected index for message 2", 0, message2.getIntProperty(INDEX));
+ }
+
+ public void testGlobalAndNotGlobalCanUseTheSameSubscriptionName() throws Exception
+ {
+ Connection connection = getClientConnection(GUEST_USERNAME, GUEST_PASSWORD, "testClientId");
+ Connection connection2 = getClientConnection(GUEST_USERNAME, GUEST_PASSWORD, null);
+
+ Session publishingSession = connection.createSession();
+ Session subscriber1Session = connection.createSession();
+ Session subscriber2Session = connection2.createSession();
+
+ String topicName = getTestName();
+ Topic topic = publishingSession.createTopic("amq.direct/" + topicName);
+ MessageConsumer consumer1 = subscriber1Session.createSharedConsumer(topic, "testSharedSubscription");
+ MessageConsumer consumer2 = subscriber2Session.createSharedConsumer(topic, "testSharedSubscription");
+ connection.start();
+ connection2.start();
+
+ sendMessage(publishingSession, topic, 1);
+
+ Message message1 = consumer1.receive(RECEIVE_TIMEOUT);
+ Message message2 = consumer2.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull("Message 1 was not received", message1);
+ assertNotNull("Message 2 was not received", message2);
+
+ assertEquals("Unexpected index for message 1", 0, message1.getIntProperty(INDEX));
+ assertEquals("Unexpected index for message 2", 0, message2.getIntProperty(INDEX));
+ }
+
+ public void testTopicOrSelectorChange() throws Exception
+ {
+ final Map<String, String> options = new HashMap<>();
+ options.put("jms.prefetchPolicy.all", "0");
+ options.put("jms.clientID", null);
+ Connection connection = getConnectionWithOptions(options);
+ Connection connection2 = getConnectionWithOptions(options);
+
+ Session publishingSession = connection.createSession();
+ Session subscriber1Session = connection.createSession();
+ Session subscriber2Session = connection2.createSession();
+
+ String topicName = getTestName();
+ Topic topic = publishingSession.createTopic("amq.direct/" + topicName);
+
+ MessageConsumer consumer1 = subscriber1Session.createSharedDurableConsumer(topic, "subscription", "index>1");
+
+ MessageProducer producer = publishingSession.createProducer(topic);
+
+ sendMessage(publishingSession, topic, 4);
+
+ connection.start();
+ connection2.start();
+
+ Message message1 = consumer1.receive(RECEIVE_TIMEOUT);
+ assertNotNull("Message 1 was not received", message1);
+ assertEquals("Unexpected index for message 1", 2, message1.getIntProperty(INDEX));
+
+ try
+ {
+ subscriber2Session.createSharedDurableConsumer(topic, "subscription", "index>2");
+ fail("Consumer should not be allowed to join shared subscription with different filter when there is an active subscriber");
+ }
+ catch (JMSException e)
+ {
+ // pass
+ }
+ Topic topic2 = publishingSession.createTopic("amq.direct/" + topicName + "2");
+ try
+ {
+ subscriber2Session.createSharedDurableConsumer(topic2, "subscription", "index>1");
+ fail("Consumer should not be allowed to join shared subscription with different topic when there is an active subscriber");
+ }
+ catch (JMSException e)
+ {
+ // pass
+ }
+ consumer1.close();
+ MessageConsumer consumer2 = subscriber2Session.createSharedDurableConsumer(topic, "subscription", "index>2");
+
+ Message message2 = consumer2.receive(RECEIVE_TIMEOUT);
+ assertNull("No message should be received as re-subscribing with different topic or selector is equivalent to unsubscribe/subscribe", message2);
+
+ sendMessage(publishingSession, topic, 4);
+
+ Message message3 = consumer2.receive(RECEIVE_TIMEOUT);
+ assertNotNull("Should receive message 3", message3);
+ assertEquals("Unexpected index for message 3", 3, message3.getIntProperty(INDEX));
+ consumer2.close();
+
+ MessageConsumer consumer3 = subscriber2Session.createSharedDurableConsumer(topic2, "subscription", "index>2");
+ Message message4 = consumer3.receive(RECEIVE_TIMEOUT);
+
+ assertNull("No message should be received as re-subscribing with different topic or selector is equivalent to unsubscribe/subscribe", message4);
+
+ sendMessage(publishingSession, topic2, 4);
+
+ Message message5 = consumer3.receive(RECEIVE_TIMEOUT);
+ assertEquals("Unexpected index for message 5", 3, message5.getIntProperty(INDEX));
+
+ }
+
+}
Modified: qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/AmqpManagementFacade.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/AmqpManagementFacade.java?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/AmqpManagementFacade.java (original)
+++ qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/AmqpManagementFacade.java Tue Jan 24 10:14:48 2017
@@ -20,7 +20,9 @@
package org.apache.qpid.test.utils;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -123,7 +125,7 @@ public class AmqpManagementFacade
}
}
- public List managementQueryObjects(final Session session, final String type) throws JMSException
+ public List<Map<String, Object>> managementQueryObjects(final Session session, final String type) throws JMSException
{
MessageProducer producer = session.createProducer(session.createQueue("$management"));
final TemporaryQueue responseQ = session.createTemporaryQueue();
@@ -143,14 +145,20 @@ public class AmqpManagementFacade
{
if (response instanceof MapMessage)
{
- return (List) ((MapMessage) response).getObject("results");
+ MapMessage bodyMap = (MapMessage) response;
+ List<String> attributeNames = (List<String>) bodyMap.getObject("attributeNames");
+ List<List<Object>> attributeValues = (List<List<Object>>) bodyMap.getObject("results");
+ return getResultsAsMap(attributeNames, attributeValues);
}
else if (response instanceof ObjectMessage)
{
Object body = ((ObjectMessage) response).getObject();
if (body instanceof Map)
{
- return (List) ((Map) body).get("results");
+ Map<String, ?> bodyMap = (Map<String, ?>) body;
+ List<String> attributeNames = (List<String>) bodyMap.get("attributeNames");
+ List<List<Object>> attributeValues = (List<List<Object>>) bodyMap.get("results");
+ return getResultsAsMap(attributeNames, attributeValues);
}
}
throw new IllegalArgumentException("Cannot parse the results from a management query");
@@ -161,4 +169,19 @@ public class AmqpManagementFacade
responseQ.delete();
}
}
+
+ private List<Map<String, Object>> getResultsAsMap(final List<String> attributeNames, final List<List<Object>> attributeValues)
+ {
+ List<Map<String, Object>> results = new ArrayList<>();
+ for (List<Object> resultObject : attributeValues)
+ {
+ Map<String, Object> result = new HashMap<>();
+ for (int i = 0; i < attributeNames.size(); ++i)
+ {
+ result.put(attributeNames.get(i), resultObject.get(i));
+ }
+ results.add(result);
+ }
+ return results;
+ }
}
\ No newline at end of file
Modified: qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java (original)
+++ qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java Tue Jan 24 10:14:48 2017
@@ -372,7 +372,7 @@ public class QpidBrokerTestCase extends
_managementFacade.performOperationUsingAmqpManagement(name, operation, session, type, arguments);
}
- protected List managementQueryObjects(final Session session, final String type) throws JMSException
+ protected List<Map<String, Object>> managementQueryObjects(final Session session, final String type) throws JMSException
{
return _managementFacade.managementQueryObjects(session, type);
Modified: qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidJmsClientProvider.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidJmsClientProvider.java?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidJmsClientProvider.java (original)
+++ qpid/java/trunk/systests/src/main/java/org/apache/qpid/test/utils/QpidJmsClientProvider.java Tue Jan 24 10:14:48 2017
@@ -117,11 +117,15 @@ public class QpidJmsClientProvider imple
options = new HashMap<>(options);
options.put("amqp.vhost", "test");
}
- if (!options.containsKey("amqp.clientID"))
+ if (!options.containsKey("jms.clientID"))
{
options = new HashMap<>(options);
options.put("jms.clientID", CLIENTID);
}
+ else if (options.get("jms.clientID") == null)
+ {
+ options.remove("jms.clientID");
+ }
if (!options.containsKey("amqp.forceSyncSend"))
{
options = new HashMap<>(options);
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java Tue Jan 24 10:14:48 2017
@@ -51,7 +51,7 @@ public class NoLocalAfterRecoveryTest ex
Connection connection = getConnection();
Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
- Topic topic = session.createTopic(MY_TOPIC_SUBSCRIPTION_NAME);
+ Topic topic = createTopic(connection, MY_TOPIC_SUBSCRIPTION_NAME);
TopicSubscriber noLocalSubscriber = session.
createDurableSubscriber(topic, MY_TOPIC_SUBSCRIPTION_NAME + "-NoLocal",
@@ -113,7 +113,7 @@ public class NoLocalAfterRecoveryTest ex
Connection connection = getConnection();
Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
- Topic topic = session.createTopic(MY_TOPIC_SUBSCRIPTION_NAME);
+ Topic topic = createTopic(connection, MY_TOPIC_SUBSCRIPTION_NAME);
TopicSubscriber noLocalSubscriber =
session.createDurableSubscriber(topic, MY_TOPIC_SUBSCRIPTION_NAME + "-NoLocal", null, true);
@@ -146,7 +146,7 @@ public class NoLocalAfterRecoveryTest ex
connection2.start();
Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
- Topic topic2 = session2.createTopic(MY_TOPIC_SUBSCRIPTION_NAME);
+ Topic topic2 = createTopic(connection2, MY_TOPIC_SUBSCRIPTION_NAME);
TopicSubscriber noLocalSubscriber2 =
session2.createDurableSubscriber(topic2, MY_TOPIC_SUBSCRIPTION_NAME + "-NoLocal",null, true);
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/server/queue/TimeToLiveTest.java Tue Jan 24 10:14:48 2017
@@ -367,7 +367,7 @@ public class TimeToLiveTest extends Qpid
// check Queue depth for up to TIMEOUT seconds after the Queue Depth hasn't changed for 100ms.
long messageCount = MSG_COUNT;
long lastPass;
- Queue subcriptionQueue = isBroker10() ? producerSession.createQueue("qpid_/clientid_/MyDurableTTLSubscription") : new AMQQueue("amq.topic","clientid" + ":" + "MyDurableTTLSubscription");
+ Queue subcriptionQueue = isBroker10() ? producerSession.createQueue("qpidsub_/clientid_/MyDurableTTLSubscription_/durable") : new AMQQueue("amq.topic","clientid" + ":" + "MyDurableTTLSubscription");
do
{
lastPass = messageCount;
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/test/unit/ct/DurableSubscriberTest.java Tue Jan 24 10:14:48 2017
@@ -40,70 +40,6 @@ public class DurableSubscriberTest exten
private final String _topicName = "durableSubscriberTopic";
/**
- * test strategy:
- * create and register a durable subscriber then close it
- * create a publisher and send a persistent 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())
- {
- TopicConnection durConnection = (TopicConnection) getConnection();
- Topic topic = createTopic(durConnection, _topicName);
- //create and register a durable subscriber then close it
- 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 = (TopicConnection) getConnection();
- 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
- {
- restartDefaultBroker();
- }
- catch (Exception e)
- {
- _logger.error("problems restarting broker: " + e);
- throw e;
- }
- //now recreate the durable subscriber and check the received messages
- TopicConnection durConnection2 = (TopicConnection) getConnection();
- 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
@@ -165,6 +101,7 @@ public class DurableSubscriberTest exten
message.getStringProperty("testprop").equals("true"));
}
}
+ durSub2.close();
durSession2.unsubscribe("dursub");
durConnection2.close();
}
@@ -378,17 +315,14 @@ public class DurableSubscriberTest exten
msg.setBooleanProperty("Match", false);
producer.send(msg);
- Message rMsg = subA.receive(1000);
+ Message rMsg = subA.receive(getReceiveTimeout());
assertNotNull(rMsg);
assertEquals("Content was wrong",
"testResubscribeWithChangedSelectorAndRestart1",
((TextMessage) rMsg).getText());
// Queue has no messages left
- AMQQueue subQueue = new AMQQueue("amq.topic", "clientid" + ":" + "testResubscribeWithChangedSelectorAndRestart");
- assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue, true));
-
- rMsg = subA.receive(1000);
+ rMsg = subA.receive(getReceiveTimeout());
assertNull(rMsg);
// Send another 1 matching message and 1 non-matching message
@@ -411,7 +345,8 @@ public class DurableSubscriberTest exten
//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
- assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue, true));
+ rMsg = subB.receive(getReceiveTimeout());
+ assertNull(rMsg);
// Check that new messages are received properly
msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1");
@@ -421,17 +356,16 @@ public class DurableSubscriberTest exten
msg.setBooleanProperty("Match", false);
producer.send(msg);
- rMsg = subB.receive(1000);
+ rMsg = subB.receive(getReceiveTimeout());
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
- assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue, true));
+ rMsg = subB.receive(getReceiveTimeout());
+ assertNull(rMsg);
+
conn.close();
//now restart the server
@@ -452,15 +386,16 @@ public class DurableSubscriberTest exten
topic = createTopic(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
- assertEquals("Msg count should be 0", 0, ((AMQSession<?, ?>) session).getQueueDepth(subQueue, true));
-
// Reconnect with new selector that matches B
TopicSubscriber subC = session.createDurableSubscriber(topic,
"testResubscribeWithChangedSelectorAndRestart",
"Match = False", false);
-
+
+ //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
+ rMsg = subC.receive(getReceiveTimeout());
+ assertNull(rMsg);
+
// Check that new messages are still sent and recieved properly
msg = session.createTextMessage("testResubscribeWithChangedSelectorAndRestart1");
msg.setBooleanProperty("Match", true);
@@ -470,20 +405,18 @@ public class DurableSubscriberTest exten
producer.send(msg);
//check the dur sub's underlying queue now has msg count 1
- assertEquals("Msg count should be 1", 1, ((AMQSession<?, ?>) session).getQueueDepth(subQueue, true));
-
- rMsg = subC.receive(1000);
+ rMsg = subC.receive(getReceiveTimeout());
assertNotNull(rMsg);
assertEquals("Content was wrong",
"testResubscribeWithChangedSelectorAndRestart2",
((TextMessage) rMsg).getText());
- rMsg = subC.receive(1000);
+ rMsg = subC.receive(getReceiveTimeout());
assertNull(rMsg);
-
- session.unsubscribe("testResubscribeWithChangedSelectorAndRestart");
-
+
subC.close();
+ session.unsubscribe("testResubscribeWithChangedSelectorAndRestart");
+
session.close();
connection.close();
}
Modified: qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes
URL: http://svn.apache.org/viewvc/qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes?rev=1780052&r1=1780051&r2=1780052&view=diff
==============================================================================
--- qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes (original)
+++ qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes Tue Jan 24 10:14:48 2017
@@ -66,5 +66,3 @@ org.apache.qpid.test.unit.topic.Temporar
org.apache.qpid.server.queue.ConsumerPriorityTest#*
org.apache.qpid.server.queue.ArrivalTimeFilterTest#*
-// Theses tests require durable subscriptions
-org.apache.qpid.server.persistent.NoLocalAfterRecoveryTest#*
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org