### Question:
RabbitDirectAsyncGateway implements DirectAsyncGateway { @Override public <T> Mono<Void> sendCommand(Command<T> command, String targetName) { return sender.sendWithConfirm(command, exchange, targetName, Collections.emptyMap(), persistentCommands); } RabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender sender, String exchange, MessageConverter converter); @Override Mono<Void> sendCommand(Command<T> command, String targetName); Flux<OutboundMessageResult> sendCommands(Flux<Command<T>> commands, String targetName); @Override Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class<R> type);  }### Answer:
@Test public void shouldSendInOptimalTime() throws InterruptedException { int messageCount = 40000; final Flux<Command<DummyMessage>> messages = createMessagesHot(messageCount); final Flux<Void> target = messages.flatMap(dummyMessageCommand -> asyncGateway.sendCommand(dummyMessageCommand, "testTarget") .doOnSuccess(aVoid -> semaphore.release())); final long init = System.currentTimeMillis(); target.subscribe(); semaphore.acquire(messageCount); final long end = System.currentTimeMillis(); final long total = end - init; final double microsPerMessage = ((total+0.0)/messageCount)*1000; System.out.println("Message count: " + messageCount); System.out.println("Total Execution Time: " + total + "ms"); System.out.println("Microseconds per message: " + microsPerMessage + "us"); Assertions.assertThat(microsPerMessage).isLessThan(150); }