### Question:
PumpShuttle implements Shuttle { @Override public void send(Collection<Message> messages) { Validate.notNull(messages); Validate.noNullElements(messages); if (messages.isEmpty()) { return; } queue.add(messages); } PumpShuttle(Shuttle backingShuttle, LinkedBlockingQueue<Collection<Message>> queue); static PumpShuttleController create(Shuttle backingShuttle); static PumpShuttleController create(Shuttle backingShuttle, int warnThreshold); @Override String getPrefix(); @Override void send(Collection<Message> messages);  }### Answer:
@Test public void mustPumpMessagesToGateway() throws InterruptedException { List<Message> messages = Arrays.asList( new Message("direct:src:1", "direct:dst:1", "payload1"), new Message("direct:src:2", "direct:dst:2", "payload2"), new Message("direct:src:3", "direct:dst:3", "payload3") ); directGateway.listen("direct:dst"); pumpShuttleController.getPumpShuttle().send(messages); Message message; message = directGateway.readMessage("direct:dst"); assertEquals(messages.get(0), message); message = directGateway.readMessage("direct:dst"); assertEquals(messages.get(1), message); message = directGateway.readMessage("direct:dst"); assertEquals(messages.get(2), message); }
@Test public void mustNotPumpMessagesIfClosed() throws InterruptedException { List<Message> messages = Arrays.asList( new Message("direct:src:1", "direct:dst:1", "payload1"), new Message("direct:src:2", "direct:dst:2", "payload2"), new Message("direct:src:3", "direct:dst:3", "payload3") ); directGateway.listen("direct:dst"); pumpShuttleController.close(); pumpShuttleController.join(); pumpShuttleController.getPumpShuttle().send(messages); Message message; message = directGateway.readMessage("direct:dst", 300L, TimeUnit.MILLISECONDS); assertNull(message); }