JMSMessageConsumer { List<Event> take() throws JMSException { List<Event> result = new ArrayList<Event>(batchSize); Message message; message = receive(); if (message != null) { result.addAll(messageConverter.convert(message)); int max = batchSize - 1; for (int i = 0; i < max; i++) { message = receiveNoWait(); if (message == null) { break; } result.addAll(messageConverter.convert(message)); } } if (logger.isDebugEnabled()) { logger.debug(String.format("Took batch of %s from %s", result.size(), destination)); } return result; } JMSMessageConsumer(InitialContext initialContext, ConnectionFactory connectionFactory,
                     String destinationName, JMSDestinationLocator destinationLocator,
                     JMSDestinationType destinationType, String messageSelector, int batchSize,
                     long pollTimeout, JMSMessageConverter messageConverter,
                     Optional<String> userName, Optional<String> password,
                     Optional<String> clientId, boolean createDurableSubscription,
                     String durableSubscriptionName);   }
@Test public void testQueue() throws Exception { destinationType = JMSDestinationType.QUEUE; when(session.createQueue(destinationName)).thenReturn(queue); consumer = create(); List<Event> events = consumer.take(); assertEquals(batchSize, events.size()); assertBodyIsExpected(events); verify(session, never()).createTopic(anyString()); }
@Test public void testTopic() throws Exception { destinationType = JMSDestinationType.TOPIC; when(session.createTopic(destinationName)).thenReturn(topic); consumer = create(); List<Event> events = consumer.take(); assertEquals(batchSize, events.size()); assertBodyIsExpected(events); verify(session, never()).createQueue(anyString()); }
@Test public void testUserPass() throws Exception { consumer = create(); List<Event> events = consumer.take(); assertEquals(batchSize, events.size()); assertBodyIsExpected(events); }
@Test public void testNoUserPass() throws Exception { userName = Optional.absent(); when(connectionFactory.createConnection(USERNAME, PASSWORD)).thenThrow(new AssertionError()); when(connectionFactory.createConnection()).thenReturn(connection); consumer = create(); List<Event> events = consumer.take(); assertEquals(batchSize, events.size()); assertBodyIsExpected(events); }