-
Notifications
You must be signed in to change notification settings - Fork 49
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Showing
5 changed files
with
234 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
169 changes: 169 additions & 0 deletions
169
src/test/java/com/rabbitmq/integration/tests/RabbitMqBasicQosIT.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,169 @@ | ||
| /* Copyright (c) 2013 Pivotal Software, Inc. All rights reserved. */ | ||
| package com.rabbitmq.integration.tests; | ||
|
|
||
| import com.rabbitmq.jms.admin.RMQConnectionFactory; | ||
| import com.rabbitmq.jms.client.RMQConnection; | ||
| import org.awaitility.Awaitility; | ||
| import org.awaitility.Duration; | ||
| import org.hamcrest.Matcher; | ||
| import org.hamcrest.Matchers; | ||
| import org.junit.Test; | ||
|
|
||
| import javax.jms.*; | ||
| import java.lang.IllegalStateException; | ||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| import static org.awaitility.Awaitility.*; | ||
| import static org.awaitility.Awaitility.waitAtMost; | ||
| import static org.awaitility.Duration.*; | ||
| import static org.hamcrest.Matchers.*; | ||
| import static org.junit.Assert.*; | ||
|
|
||
| /** | ||
| * Asynchronous Consumer integration test. | ||
| */ | ||
| public class RabbitMqBasicQosIT extends AbstractITQueue { | ||
| private static final String QUEUE_NAME = "test.queue." + RabbitMqBasicQosIT.class.getCanonicalName(); | ||
| private static final String MESSAGE = "Hello " + RabbitMqBasicQosIT.class.getName(); | ||
|
|
||
| @Test | ||
| public void messagesAreDispatchedToMultipleListenersWithQos() throws Exception { | ||
| sendMessages(); | ||
| QueueConnection connection = null; | ||
| try { | ||
| connection = connection(1); | ||
| QueueSession queueSession = connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE); | ||
| Queue queue = queueSession.createQueue(QUEUE_NAME); | ||
| QueueReceiver queueReceiver = queueSession.createReceiver(queue); | ||
|
|
||
| CountDownLatch latch = new CountDownLatch(1); | ||
| CountDownLatch ackLatch = new CountDownLatch(1); | ||
|
|
||
| MessageListener listener = new MessageListener(latch, ackLatch); | ||
| queueReceiver.setMessageListener(listener); | ||
|
|
||
| // waiting for message to be received | ||
| latch.await(1000, TimeUnit.MILLISECONDS); | ||
| // message is in first listener, but not yet ack-ed | ||
| // no other message should be delivered to first listener | ||
|
|
||
| // start another listener | ||
| QueueSession queueSession2 = connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE); | ||
| QueueReceiver queueReceiver2 = queueSession2.createReceiver(queue); | ||
| CountDownLatch latch2 = new CountDownLatch(1); | ||
| CountDownLatch ackLatch2 = new CountDownLatch(1); | ||
| MessageListener listener2 = new MessageListener(latch2, ackLatch2); | ||
| queueReceiver2.setMessageListener(listener2); | ||
|
|
||
| // wait for second message in second listener | ||
| latch2.await(1000L, TimeUnit.MILLISECONDS); | ||
| // let's ack first message in first listener | ||
| ackLatch.countDown(); | ||
| // let's ack second message in second listener | ||
| ackLatch2.countDown(); | ||
|
|
||
| assertEquals(1, listener.getMessageCount()); | ||
| assertEquals(1, listener2.getMessageCount()); | ||
| } finally { | ||
| if(connection != null) { | ||
| connection.close(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void messagesAreNotDispatchedToMultipleListenersWithoutQos() throws Exception { | ||
| sendMessages(); | ||
| QueueConnection connection = null; | ||
| try { | ||
| connection = connection(RMQConnection.NO_CHANNEL_QOS); | ||
| QueueSession queueSession = connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE); | ||
| Queue queue = queueSession.createQueue(QUEUE_NAME); | ||
| QueueReceiver queueReceiver = queueSession.createReceiver(queue); | ||
|
|
||
| CountDownLatch latch = new CountDownLatch(1); | ||
| CountDownLatch ackLatch = new CountDownLatch(1); | ||
|
|
||
| MessageListener listener = new MessageListener(latch, ackLatch); | ||
| queueReceiver.setMessageListener(listener); | ||
|
|
||
| // waiting for message to be received | ||
| latch.await(1000, TimeUnit.MILLISECONDS); | ||
| // message is in first listener, but not yet ack-ed | ||
| // second message should arrive in first listener internal queue | ||
|
|
||
| // start another listener, but it should see anything | ||
| QueueSession queueSession2 = connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE); | ||
| QueueReceiver queueReceiver2 = queueSession2.createReceiver(queue); | ||
| CountDownLatch latch2 = new CountDownLatch(1); | ||
| CountDownLatch ackLatch2 = new CountDownLatch(1); | ||
| MessageListener listener2 = new MessageListener(latch2, ackLatch2); | ||
| queueReceiver2.setMessageListener(listener2); | ||
|
|
||
| // ack first message | ||
| ackLatch.countDown(); | ||
| // second message should be delivered to first listener | ||
| waitAtMost(ONE_SECOND).untilCall(to(listener).getMessageCount(), equalTo(2)); | ||
|
|
||
| assertEquals(2, listener.getMessageCount()); | ||
| assertEquals(0, listener2.getMessageCount()); | ||
| } finally { | ||
| if(connection != null) { | ||
| connection.close(); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| } | ||
|
|
||
| private void sendMessages() throws Exception { | ||
| try { | ||
| queueConn.start(); | ||
| QueueSession queueSession = queueConn.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE); | ||
| Queue queue = queueSession.createQueue(QUEUE_NAME); | ||
| QueueSender queueSender = queueSession.createSender(queue); | ||
| queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT); | ||
| TextMessage message = queueSession.createTextMessage(MESSAGE); | ||
| queueSender.send(message); | ||
| queueSender.send(message); | ||
| } finally { | ||
| reconnect(); | ||
| } | ||
| } | ||
|
|
||
| private QueueConnection connection(int qos) throws Exception { | ||
| RMQConnectionFactory connectionFactory = (RMQConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory().getConnectionFactory(); | ||
| connectionFactory.setChannelsQos(qos); | ||
| QueueConnection queueConnection = connectionFactory.createQueueConnection(); | ||
| queueConnection.start(); | ||
| return queueConnection; | ||
| } | ||
|
|
||
| private static class MessageListener implements javax.jms.MessageListener { | ||
| private final AtomicInteger messageCount = new AtomicInteger(0); | ||
| private final CountDownLatch latch; | ||
| private final CountDownLatch ackLatch; | ||
| public MessageListener(CountDownLatch latch, CountDownLatch ackLatch) { | ||
| this.latch = latch; | ||
| this.ackLatch = ackLatch; | ||
| } | ||
| @Override | ||
| public void onMessage(Message message) { | ||
| messageCount.incrementAndGet(); | ||
| this.latch.countDown(); | ||
| try { | ||
| this.ackLatch.await(1000L, TimeUnit.MILLISECONDS); | ||
| message.acknowledge(); | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Error in message processing", e); | ||
| } | ||
|
|
||
| } | ||
| public int getMessageCount() { | ||
| return this.messageCount.get(); | ||
| } | ||
| } | ||
| } |