| /** |
| * 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.activemq.broker; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.jms.DeliveryMode; |
| |
| import junit.framework.Test; |
| |
| import org.apache.activemq.command.ActiveMQDestination; |
| import org.apache.activemq.command.ActiveMQQueue; |
| import org.apache.activemq.command.ActiveMQTopic; |
| import org.apache.activemq.command.ConnectionInfo; |
| import org.apache.activemq.command.ConsumerInfo; |
| import org.apache.activemq.command.LocalTransactionId; |
| import org.apache.activemq.command.Message; |
| import org.apache.activemq.command.MessageAck; |
| import org.apache.activemq.command.ProducerInfo; |
| import org.apache.activemq.command.SessionInfo; |
| |
| public class BrokerTest extends BrokerTestSupport { |
| |
| public ActiveMQDestination destination; |
| public int deliveryMode; |
| public int prefetch; |
| public byte destinationType; |
| public boolean durableConsumer; |
| protected static final int MAX_NULL_WAIT=500; |
| |
| public void initCombosForTestQueueOnlyOnceDeliveryWith2Consumers() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| } |
| |
| public void testQueueOnlyOnceDeliveryWith2Consumers() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQQueue("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(1); |
| connection1.request(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(1); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.request(consumerInfo2); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| for (int i = 0; i < 2; i++) { |
| Message m1 = receiveMessage(connection1); |
| Message m2 = receiveMessage(connection2); |
| |
| assertNotNull("m1 is null for index: " + i, m1); |
| assertNotNull("m2 is null for index: " + i, m2); |
| |
| assertNotSame(m1.getMessageId(), m2.getMessageId()); |
| connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| connection2.send(createAck(consumerInfo2, m2, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| assertNoMessagesLeft(connection1); |
| assertNoMessagesLeft(connection2); |
| } |
| |
| public void initCombosForTestQueueBrowserWith2Consumers() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| } |
| |
| public void testQueueBrowserWith2Consumers() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQQueue("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(10); |
| connection1.request(consumerInfo1); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| //as the messages are sent async - need to synchronize the last |
| //one to ensure they arrive in the order we want |
| connection1.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Setup a second connection with a queue browser. |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(1); |
| consumerInfo2.setBrowser(true); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.request(consumerInfo2); |
| |
| List<Message> messages = new ArrayList<Message>(); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| messages.add(m1); |
| } |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = messages.get(i); |
| Message m2 = receiveMessage(connection2); |
| assertNotNull("m2 is null for index: " + i, m2); |
| assertEquals(m1.getMessageId(), m2.getMessageId()); |
| connection2.send(createAck(consumerInfo2, m2, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| } |
| |
| assertNoMessagesLeft(connection1); |
| assertNoMessagesLeft(connection2); |
| } |
| |
| |
| /* |
| * change the order of the above test |
| */ |
| public void testQueueBrowserWith2ConsumersBrowseFirst() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQQueue("TEST"); |
| deliveryMode = DeliveryMode.NON_PERSISTENT; |
| |
| |
| // Setup a second connection with a queue browser. |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(10); |
| consumerInfo2.setBrowser(true); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.request(consumerInfo2); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(10); |
| connection1.request(consumerInfo1); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| //as the messages are sent async - need to synchronize the last |
| //one to ensure they arrive in the order we want |
| connection1.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| |
| List<Message> messages = new ArrayList<Message>(); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| messages.add(m1); |
| } |
| |
| // no messages present in queue browser as there were no messages when it |
| // was created |
| assertNoMessagesLeft(connection1); |
| assertNoMessagesLeft(connection2); |
| } |
| |
| public void testQueueBrowserWith2ConsumersInterleaved() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQQueue("TEST"); |
| deliveryMode = DeliveryMode.NON_PERSISTENT; |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(10); |
| connection1.request(consumerInfo1); |
| |
| // Send the messages |
| connection1.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Setup a second connection with a queue browser. |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(1); |
| consumerInfo2.setBrowser(true); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.request(consumerInfo2); |
| |
| |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| //as the messages are sent async - need to synchronize the last |
| //one to ensure they arrive in the order we want |
| connection1.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| |
| List<Message> messages = new ArrayList<Message>(); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| messages.add(m1); |
| } |
| |
| for (int i = 0; i < 1; i++) { |
| Message m1 = messages.get(i); |
| Message m2 = receiveMessage(connection2); |
| assertNotNull("m2 is null for index: " + i, m2); |
| assertEquals(m1.getMessageId(), m2.getMessageId()); |
| connection2.send(createAck(consumerInfo2, m2, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| } |
| |
| assertNoMessagesLeft(connection1); |
| assertNoMessagesLeft(connection2); |
| } |
| |
| |
| public void initCombosForTestConsumerPrefetchAndStandardAck() { |
| addCombinationValues("deliveryMode", new Object[] { |
| // Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testConsumerPrefetchAndStandardAck() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| consumerInfo.setPrefetchSize(1); |
| connection.send(consumerInfo); |
| |
| // Send 3 messages to the broker. |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Make sure only 1 message was delivered. |
| Message m1 = receiveMessage(connection); |
| assertNotNull(m1); |
| assertNoMessagesLeft(connection); |
| |
| // Acknowledge the first message. This should cause the next message to |
| // get dispatched. |
| connection.send(createAck(consumerInfo, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| |
| Message m2 = receiveMessage(connection); |
| assertNotNull(m2); |
| connection.send(createAck(consumerInfo, m2, 1, MessageAck.STANDARD_ACK_TYPE)); |
| |
| Message m3 = receiveMessage(connection); |
| assertNotNull(m3); |
| connection.send(createAck(consumerInfo, m3, 1, MessageAck.STANDARD_ACK_TYPE)); |
| |
| connection.send(closeConnectionInfo(connectionInfo)); |
| } |
| |
| public void initCombosForTestTransactedAckWithPrefetchOfOne() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testTransactedAckWithPrefetchOfOne() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| destination = createDestinationInfo(connection1, connectionInfo1, destinationType); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(1); |
| connection1.send(consumerInfo1); |
| |
| // Send the messages |
| for (int i = 0; i < 4; i++) { |
| Message message = createMessage(producerInfo1, destination, deliveryMode); |
| connection1.send(message); |
| } |
| |
| |
| |
| // Now get the messages. |
| for (int i = 0; i < 4; i++) { |
| // Begin the transaction. |
| LocalTransactionId txid = createLocalTransaction(sessionInfo1); |
| connection1.send(createBeginTransaction(connectionInfo1, txid)); |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| MessageAck ack = createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE); |
| ack.setTransactionId(txid); |
| connection1.send(ack); |
| // Commit the transaction. |
| connection1.send(createCommitTransaction1Phase(connectionInfo1, txid)); |
| } |
| assertNoMessagesLeft(connection1); |
| } |
| |
| public void initCombosForTestTransactedSend() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testTransactedSend() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| destination = createDestinationInfo(connection1, connectionInfo1, destinationType); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| // Begin the transaction. |
| LocalTransactionId txid = createLocalTransaction(sessionInfo1); |
| connection1.send(createBeginTransaction(connectionInfo1, txid)); |
| |
| // Send the messages |
| for (int i = 0; i < 4; i++) { |
| Message message = createMessage(producerInfo1, destination, deliveryMode); |
| message.setTransactionId(txid); |
| connection1.request(message); |
| } |
| |
| // The point of this test is that message should not be delivered until |
| // send is committed. |
| assertNull(receiveMessage(connection1,MAX_NULL_WAIT)); |
| |
| // Commit the transaction. |
| connection1.send(createCommitTransaction1Phase(connectionInfo1, txid)); |
| |
| // Now get the messages. |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| } |
| |
| assertNoMessagesLeft(connection1); |
| } |
| |
| public void initCombosForTestQueueTransactedAck() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)}); |
| } |
| |
| public void testQueueTransactedAck() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| destination = createDestinationInfo(connection1, connectionInfo1, destinationType); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| // Send the messages |
| for (int i = 0; i < 4; i++) { |
| Message message = createMessage(producerInfo1, destination, deliveryMode); |
| connection1.send(message); |
| } |
| |
| // Begin the transaction. |
| LocalTransactionId txid = createLocalTransaction(sessionInfo1); |
| connection1.send(createBeginTransaction(connectionInfo1, txid)); |
| |
| // Acknowledge the first 2 messages. |
| for (int i = 0; i < 2; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| MessageAck ack = createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE); |
| ack.setTransactionId(txid); |
| connection1.request(ack); |
| } |
| |
| // Commit the transaction. |
| connection1.send(createCommitTransaction1Phase(connectionInfo1, txid)); |
| |
| // The queue should now only have the remaining 2 messages |
| assertEquals(2, countMessagesInQueue(connection1, connectionInfo1, destination)); |
| } |
| |
| public void initCombosForTestConsumerCloseCausesRedelivery() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destination", new Object[] {new ActiveMQQueue("TEST")}); |
| } |
| |
| public void testConsumerCloseCausesRedelivery() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.request(consumerInfo1); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| |
| // Receive the messages. |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| assertFalse(m1.isRedelivered()); |
| } |
| |
| // Close the consumer without acking.. this should cause re-delivery of |
| // the messages. |
| connection1.send(consumerInfo1.createRemoveCommand()); |
| |
| // Create another consumer that should get the messages again. |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo2.setPrefetchSize(100); |
| connection1.request(consumerInfo2); |
| |
| // Receive the messages. |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| assertTrue(m1.isRedelivered()); |
| } |
| assertNoMessagesLeft(connection1); |
| |
| } |
| |
| public void testTopicDurableSubscriptionCanBeRestored() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQTopic("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| connectionInfo1.setClientId("clientid1"); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| consumerInfo1.setSubscriptionName("test"); |
| connection1.send(consumerInfo1); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT)); |
| connection1.send(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT)); |
| connection1.send(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT)); |
| connection1.request(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT)); |
| |
| // Get the messages |
| Message m = null; |
| for (int i = 0; i < 2; i++) { |
| m = receiveMessage(connection1); |
| assertNotNull(m); |
| } |
| // Ack the last message. |
| connection1.send(createAck(consumerInfo1, m, 2, MessageAck.STANDARD_ACK_TYPE)); |
| // Close the connection. |
| connection1.request(closeConnectionInfo(connectionInfo1)); |
| connection1.stop(); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| connectionInfo2.setClientId("clientid1"); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(100); |
| consumerInfo2.setSubscriptionName("test"); |
| |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.send(consumerInfo2); |
| |
| // Get the rest of the messages |
| for (int i = 0; i < 2; i++) { |
| Message m1 = receiveMessage(connection2); |
| assertNotNull("m1 is null for index: " + i, m1); |
| } |
| assertNoMessagesLeft(connection2); |
| } |
| |
| public void initCombosForTestGroupedMessagesDeliveredToOnlyOneConsumer() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| } |
| |
| public void testGroupedMessagesDeliveredToOnlyOneConsumer() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQQueue("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(1); |
| connection1.send(consumerInfo1); |
| |
| // Send the messages. |
| for (int i = 0; i < 4; i++) { |
| Message message = createMessage(producerInfo, destination, deliveryMode); |
| message.setGroupID("TEST-GROUP"); |
| message.setGroupSequence(i + 1); |
| connection1.request(message); |
| } |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(1); |
| connection2.send(consumerInfo2); |
| |
| // All the messages should have been sent down connection 1.. just get |
| // the first 3 |
| for (int i = 0; i < 3; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull("m1 is null for index: " + i, m1); |
| connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| // Close the first consumer. |
| connection1.request(closeConsumerInfo(consumerInfo1)); |
| |
| // The last messages should now go the the second consumer. |
| for (int i = 0; i < 1; i++) { |
| Message m1 = receiveMessage(connection2); |
| assertNotNull("m1 is null for index: " + i, m1); |
| connection2.request(createAck(consumerInfo2, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| assertNoMessagesLeft(connection2); |
| } |
| |
| public void initCombosForTestTopicConsumerOnlySeeMessagesAfterCreation() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("durableConsumer", new Object[] {Boolean.TRUE, Boolean.FALSE}); |
| } |
| |
| public void testTopicConsumerOnlySeeMessagesAfterCreation() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQTopic("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| connectionInfo1.setClientId("A"); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| // Send the 1st message |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| |
| // Create the durable subscription. |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| if (durableConsumer) { |
| consumerInfo1.setSubscriptionName("test"); |
| } |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| Message m = createMessage(producerInfo1, destination, deliveryMode); |
| connection1.send(m); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| |
| // Subscription should skip over the first message |
| Message m2 = receiveMessage(connection1); |
| assertNotNull(m2); |
| assertEquals(m.getMessageId(), m2.getMessageId()); |
| m2 = receiveMessage(connection1); |
| assertNotNull(m2); |
| |
| assertNoMessagesLeft(connection1); |
| } |
| |
| public void initCombosForTestTopicRetroactiveConsumerSeeMessagesBeforeCreation() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("durableConsumer", new Object[] {Boolean.TRUE, Boolean.FALSE}); |
| } |
| |
| public void testTopicRetroactiveConsumerSeeMessagesBeforeCreation() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQTopic("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| connectionInfo1.setClientId("A"); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| // Send the messages |
| Message m = createMessage(producerInfo1, destination, deliveryMode); |
| connection1.send(m); |
| |
| // Create the durable subscription. |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| if (durableConsumer) { |
| consumerInfo1.setSubscriptionName("test"); |
| } |
| consumerInfo1.setPrefetchSize(100); |
| consumerInfo1.setRetroactive(true); |
| connection1.send(consumerInfo1); |
| |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.request(createMessage(producerInfo1, destination, deliveryMode)); |
| |
| // the behavior is VERY dependent on the recovery policy used. |
| // But the default broker settings try to make it as consistent as |
| // possible |
| |
| // Subscription should see all messages sent. |
| Message m2 = receiveMessage(connection1); |
| assertNotNull(m2); |
| assertEquals(m.getMessageId(), m2.getMessageId()); |
| for (int i = 0; i < 2; i++) { |
| m2 = receiveMessage(connection1); |
| assertNotNull(m2); |
| } |
| |
| assertNoMessagesLeft(connection1); |
| } |
| |
| // |
| // TODO: need to reimplement this since we don't fail when we send to a |
| // non-existant |
| // destination. But if we can access the Region directly then we should be |
| // able to |
| // check that if the destination was removed. |
| // |
| // public void initCombosForTestTempDestinationsRemovedOnConnectionClose() { |
| // addCombinationValues( "deliveryMode", new Object[]{ |
| // Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| // Integer.valueOf(DeliveryMode.PERSISTENT)} ); |
| // addCombinationValues( "destinationType", new Object[]{ |
| // Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| // Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} ); |
| // } |
| // |
| // public void testTempDestinationsRemovedOnConnectionClose() throws |
| // Exception { |
| // |
| // // Setup a first connection |
| // StubConnection connection1 = createConnection(); |
| // ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| // SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| // ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| // connection1.send(connectionInfo1); |
| // connection1.send(sessionInfo1); |
| // connection1.send(producerInfo1); |
| // |
| // destination = createDestinationInfo(connection1, connectionInfo1, |
| // destinationType); |
| // |
| // StubConnection connection2 = createConnection(); |
| // ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| // SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| // ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2); |
| // connection2.send(connectionInfo2); |
| // connection2.send(sessionInfo2); |
| // connection2.send(producerInfo2); |
| // |
| // // Send from connection2 to connection1's temp destination. Should |
| // succeed. |
| // connection2.send(createMessage(producerInfo2, destination, |
| // deliveryMode)); |
| // |
| // // Close connection 1 |
| // connection1.request(closeConnectionInfo(connectionInfo1)); |
| // |
| // try { |
| // // Send from connection2 to connection1's temp destination. Should not |
| // succeed. |
| // connection2.request(createMessage(producerInfo2, destination, |
| // deliveryMode)); |
| // fail("Expected JMSException."); |
| // } catch ( JMSException success ) { |
| // } |
| // |
| // } |
| |
| // public void initCombosForTestTempDestinationsAreNotAutoCreated() { |
| // addCombinationValues( "deliveryMode", new Object[]{ |
| // Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| // Integer.valueOf(DeliveryMode.PERSISTENT)} ); |
| // addCombinationValues( "destinationType", new Object[]{ |
| // Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| // Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} ); |
| // } |
| // |
| // |
| |
| // We create temp destination on demand now so this test case is no longer |
| // valid. |
| // |
| // public void testTempDestinationsAreNotAutoCreated() throws Exception { |
| // |
| // // Setup a first connection |
| // StubConnection connection1 = createConnection(); |
| // ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| // SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| // ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| // connection1.send(connectionInfo1); |
| // connection1.send(sessionInfo1); |
| // connection1.send(producerInfo1); |
| // |
| // destination = |
| // ActiveMQDestination.createDestination(connectionInfo1.getConnectionId()+":1", |
| // destinationType); |
| // |
| // // Should not be able to send to a non-existant temp destination. |
| // try { |
| // connection1.request(createMessage(producerInfo1, destination, |
| // deliveryMode)); |
| // fail("Expected JMSException."); |
| // } catch ( JMSException success ) { |
| // } |
| // |
| // } |
| |
| |
| public void initCombosForTestExclusiveQueueDeliversToOnlyOneConsumer() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| } |
| |
| public void testExclusiveQueueDeliversToOnlyOneConsumer() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQQueue("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(1); |
| consumerInfo1.setExclusive(true); |
| connection1.send(consumerInfo1); |
| |
| // Send a message.. this should make consumer 1 the exclusive owner. |
| connection1.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setPrefetchSize(1); |
| consumerInfo2.setExclusive(true); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.request(consumerInfo2); |
| |
| // Second message should go to consumer 1 even though consumer 2 is |
| // ready |
| // for dispatch. |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Acknowledge the first 2 messages |
| for (int i = 0; i < 2; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| // Close the first consumer. |
| connection1.send(closeConsumerInfo(consumerInfo1)); |
| |
| // The last two messages should now go the the second consumer. |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| for (int i = 0; i < 2; i++) { |
| Message m1 = receiveMessage(connection2); |
| assertNotNull(m1); |
| connection2.send(createAck(consumerInfo2, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| assertNoMessagesLeft(connection2); |
| } |
| |
| public void initCombosForTestWildcardConsume() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)}); |
| } |
| |
| public void testWildcardConsume() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| // setup the wildcard consumer. |
| ActiveMQDestination compositeDestination = ActiveMQDestination.createDestination("WILD.*.TEST", |
| destinationType); |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, compositeDestination); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| // These two message should NOT match the wild card. |
| connection1.send(createMessage(producerInfo1, ActiveMQDestination.createDestination("WILD.CARD", |
| destinationType), |
| deliveryMode)); |
| connection1.send(createMessage(producerInfo1, ActiveMQDestination.createDestination("WILD.TEST", |
| destinationType), |
| deliveryMode)); |
| |
| // These two message should match the wild card. |
| ActiveMQDestination d1 = ActiveMQDestination.createDestination("WILD.CARD.TEST", destinationType); |
| connection1.send(createMessage(producerInfo1, d1, deliveryMode)); |
| |
| Message m = receiveMessage(connection1); |
| assertNotNull(m); |
| assertEquals(d1, m.getDestination()); |
| |
| ActiveMQDestination d2 = ActiveMQDestination.createDestination("WILD.FOO.TEST", destinationType); |
| connection1.request(createMessage(producerInfo1, d2, deliveryMode)); |
| m = receiveMessage(connection1); |
| assertNotNull(m); |
| assertEquals(d2, m.getDestination()); |
| |
| assertNoMessagesLeft(connection1); |
| connection1.send(closeConnectionInfo(connectionInfo1)); |
| } |
| |
| public void initCombosForTestCompositeConsume() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)}); |
| } |
| |
| public void testCompositeConsume() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| // setup the composite consumer. |
| ActiveMQDestination compositeDestination = ActiveMQDestination.createDestination("A,B", |
| destinationType); |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, compositeDestination); |
| consumerInfo1.setRetroactive(true); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| // Publish to the two destinations |
| ActiveMQDestination destinationA = ActiveMQDestination.createDestination("A", destinationType); |
| ActiveMQDestination destinationB = ActiveMQDestination.createDestination("B", destinationType); |
| |
| // Send a message to each destination . |
| connection1.send(createMessage(producerInfo1, destinationA, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destinationB, deliveryMode)); |
| |
| // The consumer should get both messages. |
| for (int i = 0; i < 2; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| } |
| |
| assertNoMessagesLeft(connection1); |
| connection1.send(closeConnectionInfo(connectionInfo1)); |
| } |
| |
| public void initCombosForTestCompositeSend() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)}); |
| } |
| |
| public void testCompositeSend() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| ActiveMQDestination destinationA = ActiveMQDestination.createDestination("A", destinationType); |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destinationA); |
| consumerInfo1.setRetroactive(true); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.request(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| |
| ActiveMQDestination destinationB = ActiveMQDestination.createDestination("B", destinationType); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destinationB); |
| consumerInfo2.setRetroactive(true); |
| consumerInfo2.setPrefetchSize(100); |
| connection2.request(consumerInfo2); |
| |
| // Send the messages to the composite destination. |
| ActiveMQDestination compositeDestination = ActiveMQDestination.createDestination("A,B", |
| destinationType); |
| for (int i = 0; i < 4; i++) { |
| connection1.request(createMessage(producerInfo1, compositeDestination, deliveryMode)); |
| } |
| |
| // The messages should have been delivered to both the A and B |
| // destination. |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| Message m2 = receiveMessage(connection2); |
| |
| assertNotNull(m1); |
| assertNotNull(m2); |
| |
| assertEquals(m1.getMessageId(), m2.getMessageId()); |
| assertEquals(compositeDestination, m1.getOriginalDestination()); |
| assertEquals(compositeDestination, m2.getOriginalDestination()); |
| |
| connection1.request(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| connection2.request(createAck(consumerInfo2, m2, 1, MessageAck.STANDARD_ACK_TYPE)); |
| |
| } |
| |
| assertNoMessagesLeft(connection1); |
| assertNoMessagesLeft(connection2); |
| |
| connection1.send(closeConnectionInfo(connectionInfo1)); |
| connection2.send(closeConnectionInfo(connectionInfo2)); |
| } |
| |
| public void initCombosForTestConnectionCloseCascades() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destination", new Object[] {new ActiveMQTopic("TEST"), |
| new ActiveMQQueue("TEST")}); |
| } |
| |
| public void testConnectionCloseCascades() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| consumerInfo1.setNoLocal(true); |
| connection1.request(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.send(producerInfo2); |
| |
| // Send the messages |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| // give the async ack a chance to perculate and validate all are currently consumed |
| Message msg = receiveMessage(connection1, MAX_NULL_WAIT); |
| assertNull("all messages were received " + msg, msg); |
| |
| // Close the connection, this should in turn close the consumer. |
| connection1.request(closeConnectionInfo(connectionInfo1)); |
| |
| // Send another message, connection1 should not get the message. |
| connection2.request(createMessage(producerInfo2, destination, deliveryMode)); |
| |
| assertNull("no message received", receiveMessage(connection1, MAX_NULL_WAIT)); |
| } |
| |
| public void initCombosForTestSessionCloseCascades() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destination", new Object[] {new ActiveMQTopic("TEST"), |
| new ActiveMQQueue("TEST")}); |
| } |
| |
| public void testSessionCloseCascades() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| consumerInfo1.setNoLocal(true); |
| connection1.request(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.send(producerInfo2); |
| |
| // Send the messages |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| // Close the session, this should in turn close the consumer. |
| connection1.request(closeSessionInfo(sessionInfo1)); |
| |
| // Send another message, connection1 should not get the message. |
| connection2.request(createMessage(producerInfo2, destination, deliveryMode)); |
| |
| Message msg = receiveMessage(connection1,MAX_NULL_WAIT); |
| assertNull("no message received from connection1 after session close", msg); |
| } |
| |
| public void initCombosForTestConsumerClose() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destination", new Object[] {new ActiveMQTopic("TEST"), |
| new ActiveMQQueue("TEST")}); |
| } |
| |
| public void testConsumerClose() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| consumerInfo1.setNoLocal(true); |
| connection1.request(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.send(producerInfo2); |
| |
| // Send the messages |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| connection2.send(createMessage(producerInfo2, destination, deliveryMode)); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE)); |
| } |
| |
| // give the async ack a chance to perculate and validate all are currently consumed |
| // use receive rather than poll as broker info is sent async and may still need to be dequeued |
| Message result = receiveMessage(connection1, MAX_NULL_WAIT); |
| assertNull("no more messages " + result, result); |
| |
| // Close the consumer. |
| connection1.request(closeConsumerInfo(consumerInfo1)); |
| |
| // Send another message, connection1 should not get the message. |
| connection2.request(createMessage(producerInfo2, destination, deliveryMode)); |
| |
| result = receiveMessage(connection1, MAX_NULL_WAIT); |
| assertNull("no message received after close " + result, result); |
| } |
| |
| public void initCombosForTestTopicNoLocal() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| } |
| |
| public void testTopicNoLocal() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQTopic("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setRetroactive(true); |
| consumerInfo1.setPrefetchSize(100); |
| consumerInfo1.setNoLocal(true); |
| connection1.send(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.send(producerInfo2); |
| |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setRetroactive(true); |
| consumerInfo2.setPrefetchSize(100); |
| consumerInfo2.setNoLocal(true); |
| connection2.send(consumerInfo2); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| |
| // The 2nd connection should get the messages. |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection2); |
| assertNotNull(m1); |
| } |
| |
| // Send a message with the 2nd connection |
| Message message = createMessage(producerInfo2, destination, deliveryMode); |
| connection2.send(message); |
| |
| // The first connection should not see the initial 4 local messages sent |
| // but should |
| // see the messages from connection 2. |
| Message m = receiveMessage(connection1); |
| assertNotNull(m); |
| assertEquals(message.getMessageId(), m.getMessageId()); |
| |
| assertNoMessagesLeft(connection1); |
| assertNoMessagesLeft(connection2); |
| } |
| |
| public void initCombosForTopicDispatchIsBroadcast() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| } |
| |
| public void testTopicDispatchIsBroadcast() throws Exception { |
| |
| ActiveMQDestination destination = new ActiveMQTopic("TEST"); |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo1); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setRetroactive(true); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| // Setup a second connection |
| StubConnection connection2 = createConnection(); |
| ConnectionInfo connectionInfo2 = createConnectionInfo(); |
| SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2); |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination); |
| consumerInfo2.setRetroactive(true); |
| consumerInfo2.setPrefetchSize(100); |
| connection2.send(connectionInfo2); |
| connection2.send(sessionInfo2); |
| connection2.send(consumerInfo2); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo1, destination, deliveryMode)); |
| |
| // Get the messages |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| m1 = receiveMessage(connection2); |
| assertNotNull(m1); |
| } |
| } |
| |
| public void initCombosForTestQueueDispatchedAreRedeliveredOnConsumerClose() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)}); |
| } |
| |
| public void testQueueDispatchedAreRedeliveredOnConsumerClose() throws Exception { |
| |
| // Setup a first connection |
| StubConnection connection1 = createConnection(); |
| ConnectionInfo connectionInfo1 = createConnectionInfo(); |
| SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo1); |
| connection1.send(connectionInfo1); |
| connection1.send(sessionInfo1); |
| connection1.send(producerInfo); |
| |
| destination = createDestinationInfo(connection1, connectionInfo1, destinationType); |
| |
| ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo1.setPrefetchSize(100); |
| connection1.send(consumerInfo1); |
| |
| // Send the messages |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection1.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Get the messages |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| assertFalse(m1.isRedelivered()); |
| } |
| // Close the consumer without sending any ACKS. |
| connection1.send(closeConsumerInfo(consumerInfo1)); |
| |
| // Drain any in flight messages.. |
| while (connection1.getDispatchQueue().poll(0, TimeUnit.MILLISECONDS) != null) { |
| } |
| |
| // Add the second consumer |
| ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo1, destination); |
| consumerInfo2.setPrefetchSize(100); |
| connection1.send(consumerInfo2); |
| |
| // Make sure the messages were re delivered to the 2nd consumer. |
| for (int i = 0; i < 4; i++) { |
| Message m1 = receiveMessage(connection1); |
| assertNotNull(m1); |
| assertTrue(m1.isRedelivered()); |
| } |
| } |
| |
| public void initCombosForTestQueueBrowseMessages() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)}); |
| } |
| |
| public void testQueueBrowseMessages() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Use selector to skip first message. |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| consumerInfo.setBrowser(true); |
| connection.send(consumerInfo); |
| |
| for (int i = 0; i < 4; i++) { |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| connection.send(createAck(consumerInfo, m, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| } |
| |
| assertNoMessagesLeft(connection); |
| } |
| |
| public void initCombosForTestQueueSendThenAddConsumer() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)}); |
| } |
| |
| public void testQueueSendThenAddConsumer() throws Exception { |
| |
| // Start a producer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| // Send a message to the broker. |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Start the consumer |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| connection.send(consumerInfo); |
| |
| // Make sure the message was delivered. |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| |
| } |
| |
| public void initCombosForTestQueueAckRemovesMessage() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)}); |
| } |
| |
| public void testQueueAckRemovesMessage() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| Message message1 = createMessage(producerInfo, destination, deliveryMode); |
| Message message2 = createMessage(producerInfo, destination, deliveryMode); |
| connection.send(message1); |
| connection.send(message2); |
| |
| // Make sure the message was delivered. |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| connection.request(consumerInfo); |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| assertEquals(m.getMessageId(), message1.getMessageId()); |
| |
| assertTrue(countMessagesInQueue(connection, connectionInfo, destination) == 2); |
| connection.send(createAck(consumerInfo, m, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| assertTrue(countMessagesInQueue(connection, connectionInfo, destination) == 2); |
| connection.send(createAck(consumerInfo, m, 1, MessageAck.STANDARD_ACK_TYPE)); |
| assertTrue(countMessagesInQueue(connection, connectionInfo, destination) == 1); |
| |
| } |
| |
| public void initCombosForTestSelectorSkipsMessages() { |
| addCombinationValues("destination", new Object[] {new ActiveMQTopic("TEST_TOPIC"), |
| new ActiveMQQueue("TEST_QUEUE")}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testSelectorSkipsMessages() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| consumerInfo.setSelector("JMSType='last'"); |
| connection.send(consumerInfo); |
| |
| Message message1 = createMessage(producerInfo, destination, deliveryMode); |
| message1.setType("first"); |
| Message message2 = createMessage(producerInfo, destination, deliveryMode); |
| message2.setType("last"); |
| connection.send(message1); |
| connection.send(message2); |
| |
| // Use selector to skip first message. |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| assertEquals(m.getMessageId(), message2.getMessageId()); |
| connection.send(createAck(consumerInfo, m, 1, MessageAck.STANDARD_ACK_TYPE)); |
| connection.send(closeConsumerInfo(consumerInfo)); |
| |
| assertNoMessagesLeft(connection); |
| } |
| |
| public void initCombosForTestAddConsumerThenSend() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testAddConsumerThenSend() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| connection.send(consumerInfo); |
| |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Make sure the message was delivered. |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| } |
| |
| public void initCombosForTestConsumerPrefetchAtOne() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testConsumerPrefetchAtOne() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| consumerInfo.setPrefetchSize(1); |
| connection.send(consumerInfo); |
| |
| // Send 2 messages to the broker. |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Make sure only 1 message was delivered. |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| assertNoMessagesLeft(connection); |
| |
| } |
| |
| public void initCombosForTestConsumerPrefetchAtTwo() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testConsumerPrefetchAtTwo() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| consumerInfo.setPrefetchSize(2); |
| connection.send(consumerInfo); |
| |
| // Send 3 messages to the broker. |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Make sure only 1 message was delivered. |
| Message m = receiveMessage(connection); |
| assertNotNull(m); |
| m = receiveMessage(connection); |
| assertNotNull(m); |
| assertNoMessagesLeft(connection); |
| |
| } |
| |
| public void initCombosForTestConsumerPrefetchAndDeliveredAck() { |
| addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), |
| Integer.valueOf(DeliveryMode.PERSISTENT)}); |
| addCombinationValues("destinationType", |
| new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), |
| Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)}); |
| } |
| |
| public void testConsumerPrefetchAndDeliveredAck() throws Exception { |
| |
| // Start a producer and consumer |
| StubConnection connection = createConnection(); |
| ConnectionInfo connectionInfo = createConnectionInfo(); |
| SessionInfo sessionInfo = createSessionInfo(connectionInfo); |
| ProducerInfo producerInfo = createProducerInfo(sessionInfo); |
| connection.send(connectionInfo); |
| connection.send(sessionInfo); |
| connection.send(producerInfo); |
| |
| destination = createDestinationInfo(connection, connectionInfo, destinationType); |
| |
| ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination); |
| consumerInfo.setPrefetchSize(1); |
| connection.request(consumerInfo); |
| |
| // Send 3 messages to the broker. |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.send(createMessage(producerInfo, destination, deliveryMode)); |
| connection.request(createMessage(producerInfo, destination, deliveryMode)); |
| |
| // Make sure only 1 message was delivered. |
| Message m1 = receiveMessage(connection); |
| assertNotNull(m1); |
| |
| assertNoMessagesLeft(connection); |
| |
| // Acknowledge the first message. This should cause the next message to |
| // get dispatched. |
| connection.request(createAck(consumerInfo, m1, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| |
| Message m2 = receiveMessage(connection); |
| assertNotNull(m2); |
| connection.request(createAck(consumerInfo, m2, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| |
| Message m3 = receiveMessage(connection); |
| assertNotNull(m3); |
| connection.request(createAck(consumerInfo, m3, 1, MessageAck.DELIVERED_ACK_TYPE)); |
| } |
| |
| public void testGetServices() throws Exception { |
| assertTrue(broker.getServices().length != 0); |
| } |
| |
| public static Test suite() { |
| return suite(BrokerTest.class); |
| } |
| |
| public static void main(String[] args) { |
| junit.textui.TestRunner.run(suite()); |
| } |
| |
| } |
