CollectorJobScheduler implements Runnable { void add(Queue<MessageWithDestination> messagesQueue, String correlationID) { if (messagesMap.putIfAbsent(correlationID, messagesQueue) == null) { availableMessages.release(); LOGGER.debug( "New collection message with correlationID: {} (Semaphore availableMessages released). Currently waiting for {} packages ({} slots available).", correlationID, receivedMessagesCounter.size(), getAvailableSlots()); } else { LOGGER.error("Message {} already in messages map!", correlationID); throw new IllegalStateException(); } } CollectorJobScheduler(JmsConnection jmsConnection, Integer maxMessagesInCollectorQueue,
      MessagesManager messagesManager); @Override void run();  }
@Test(timeout = 2000l) public void add_expectMessageSent() throws Exception { Message mockedMessage = Mockito.mock(Message.class); Destination destination = Mockito.mock(Destination.class); String correlationID = "01234567890"; Queue<MessageWithDestination> messagesQueue = mockMessagesQueue(mockedMessage, destination, correlationID); tested.add(messagesQueue, correlationID); Thread.sleep(500); verify(producer, times(1)).send(Matchers.<Destination>any(), Matchers.<Message>any()); }