### Question:
ReactiveMessageSender { public <T> Mono<Void> sendWithConfirm(T message, String exchange, String routingKey, Map<String, Object> headers, boolean persistent) { return Mono.create(monoSink -> { Consumer<Boolean> notifier = new AckNotifier(monoSink); final MyOutboundMessage outboundMessage = toOutboundMessage(message, exchange, routingKey, headers, notifier, persistent); executorService2.submit(() -> fluxSinkConfirm.get((int) (System.currentTimeMillis()%numberOfSenderSubscriptions)).next(outboundMessage)); }); } ReactiveMessageSender(Sender sender, String sourceApplication, MessageConverter messageConverter, TopologyCreator topologyCreator); Mono<Void> sendWithConfirm(T message, String exchange, String routingKey, Map<String, Object> headers, boolean persistent); Mono<Void> sendNoConfirm(T message, String exchange, String routingKey, Map<String, Object> headers, boolean persistent); Flux<OutboundMessageResult> sendWithConfirmBatch(Flux<T> messages, String exchange, String routingKey, Map<String, Object> headers, boolean persistent); reactor.rabbitmq.Sender getSender(); TopologyCreator getTopologyCreator();  }### Answer:
@Test public void sendWithConfirmEmptyNullMessage() { final Mono<Void> voidMono = messageSender.sendWithConfirm(null, "exchange", "rkey", new HashMap<>(), true); StepVerifier.create(voidMono).verifyComplete(); }
@Test public void sendWithConfirmSomeMessage() { SomeClass some = new SomeClass("42", "Daniel", new Date()); final Mono<Void> voidMono = messageSender.sendWithConfirm(some, "exchange", "rkey", new HashMap<>(), true); StepVerifier.create(voidMono).verifyComplete(); }