@Test public void testTopic() throws Exception { destinationType = JMSDestinationType.TOPIC; when(session.createQueue(destinationName)).thenThrow(new AssertionError()); when(session.createTopic(destinationName)).thenReturn(topic); consumer = create(); List<Event> events = consumer.take(); assertEquals(batchSize, events.size()); assertBodyIsExpected(events); }
List<Event> take() throws JMSException { List<Event> result = new ArrayList<Event>(batchSize); Message message; message = messageConsumer.receive(pollTimeout); if(message != null) { result.addAll(messageConverter.convert(message)); int max = batchSize - 1; for (int i = 0; i < max; i++) { message = messageConsumer.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 { List<Event> take() throws JMSException { List<Event> result = new ArrayList<Event>(batchSize); Message message; message = messageConsumer.receive(pollTimeout); if(message != null) { result.addAll(messageConverter.convert(message)); int max = batchSize - 1; for (int i = 0; i < max; i++) { message = messageConsumer.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 { List<Event> take() throws JMSException { List<Event> result = new ArrayList<Event>(batchSize); Message message; message = messageConsumer.receive(pollTimeout); if(message != null) { result.addAll(messageConverter.convert(message)); int max = batchSize - 1; for (int i = 0; i < max; i++) { message = messageConsumer.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(ConnectionFactory connectionFactory,
      String destinationName, JMSDestinationType destinationType,
      String messageSelector, int batchSize, long pollTimeout,
      JMSMessageConverter messageConverter, Optional<String> userName,
      Optional<String> password); }
JMSMessageConsumer { List<Event> take() throws JMSException { List<Event> result = new ArrayList<Event>(batchSize); Message message; message = messageConsumer.receive(pollTimeout); if(message != null) { result.addAll(messageConverter.convert(message)); int max = batchSize - 1; for (int i = 0; i < max; i++) { message = messageConsumer.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(ConnectionFactory connectionFactory,
      String destinationName, JMSDestinationType destinationType,
      String messageSelector, int batchSize, long pollTimeout,
      JMSMessageConverter messageConverter, Optional<String> userName,
      Optional<String> password);  }
JMSMessageConsumer { List<Event> take() throws JMSException { List<Event> result = new ArrayList<Event>(batchSize); Message message; message = messageConsumer.receive(pollTimeout); if(message != null) { result.addAll(messageConverter.convert(message)); int max = batchSize - 1; for (int i = 0; i < max; i++) { message = messageConsumer.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(ConnectionFactory connectionFactory,
      String destinationName, JMSDestinationType destinationType,
      String messageSelector, int batchSize, long pollTimeout,
      JMSMessageConverter messageConverter, Optional<String> userName,
      Optional<String> password);   }