@Test public void testGetLastMessageFromQueue() throws Exception { final String queueName = "queue"; doAnswer( call -> { Consumer<Cdr> listener = call.getArgument(1); listener.accept(aCdr("M1").build()); listener.accept(aCdr("M2").build()); return null; } ).when(transport).subscribeToQueue(eq(queueName), any(Consumer.class)); client.subscribeToQueue(queueName); Optional<Cdr> lastMessage = client.getLastMessageFromQueue(queueName); assertTrue(lastMessage.isPresent()); assertEquals("M2", lastMessage.get().getType()); }
public Optional<Cdr> getLastMessageFromQueue(final String queueName) { return Optional.ofNullable( messagesByQueue.getOrDefault(queueName, new LinkedList<>()) .peekLast() ); }
ExchangeBrokerClient extends Client { public Optional<Cdr> getLastMessageFromQueue(final String queueName) { return Optional.ofNullable( messagesByQueue.getOrDefault(queueName, new LinkedList<>()) .peekLast() ); } }
ExchangeBrokerClient extends Client { public Optional<Cdr> getLastMessageFromQueue(final String queueName) { return Optional.ofNullable( messagesByQueue.getOrDefault(queueName, new LinkedList<>()) .peekLast() ); } ExchangeBrokerClient(String clientConfig);  ExchangeBrokerClient(String primaryClientConfig, String secondaryClientConfig);  ExchangeBrokerClient(String primaryClientConfig, String secondaryClientConfig, ProtocolConfig protocol); }
ExchangeBrokerClient extends Client { public Optional<Cdr> getLastMessageFromQueue(final String queueName) { return Optional.ofNullable( messagesByQueue.getOrDefault(queueName, new LinkedList<>()) .peekLast() ); } ExchangeBrokerClient(String clientConfig);  ExchangeBrokerClient(String primaryClientConfig, String secondaryClientConfig);  ExchangeBrokerClient(String primaryClientConfig, String secondaryClientConfig, ProtocolConfig protocol); @Override boolean isLoggedOn(); @Override ExchangeTransport getTransport(); void sendMessageToExchange(final Cdr cdr, final String exchangeName); void sendMessageToExchange(final Cdr cdr, final String exchangeName, final String routingKey); void subscribeToQueue(final String queueName); Cdr retrieveMessageFromQueue(final String queueName, Duration timeout); List<Cdr> getReceivedMessagesFromQueue(final String queueName); Optional<Cdr> getLastMessageFromQueue(final String queueName); }
ExchangeBrokerClient extends Client { public Optional<Cdr> getLastMessageFromQueue(final String queueName) { return Optional.ofNullable( messagesByQueue.getOrDefault(queueName, new LinkedList<>()) .peekLast() ); } ExchangeBrokerClient(String clientConfig);  ExchangeBrokerClient(String primaryClientConfig, String secondaryClientConfig);  ExchangeBrokerClient(String primaryClientConfig, String secondaryClientConfig, ProtocolConfig protocol); @Override boolean isLoggedOn(); @Override ExchangeTransport getTransport(); void sendMessageToExchange(final Cdr cdr, final String exchangeName); void sendMessageToExchange(final Cdr cdr, final String exchangeName, final String routingKey); void subscribeToQueue(final String queueName); Cdr retrieveMessageFromQueue(final String queueName, Duration timeout); List<Cdr> getReceivedMessagesFromQueue(final String queueName); Optional<Cdr> getLastMessageFromQueue(final String queueName);  }