@Test @SuppressWarnings("unchecked") public void shouldFlush() { given(mockMutinyProducer.flush(any())) .will(a -> { ((Consumer<Void>) a.getArgument(0)).accept(null); return mockMutinyProducer; }); StepVerifier.create(producer.flush()) .verifyComplete(); }
@Override public Mono<Void> flush() { return Mono.create(sink -> { try { delegate.flush(v -> sink.success()); } catch (Throwable t) { sink.error(t); } }); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> flush() { return Mono.create(sink -> { try { delegate.flush(v -> sink.success()); } catch (Throwable t) { sink.error(t); } }); } }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> flush() { return Mono.create(sink -> { try { delegate.flush(v -> sink.success()); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaProducer(io.vertx.mutiny.kafka.client.producer.KafkaProducer<K, V> delegate); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> flush() { return Mono.create(sink -> { try { delegate.flush(v -> sink.success()); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaProducer(io.vertx.mutiny.kafka.client.producer.KafkaProducer<K, V> delegate); @Override Mono<RecordMetadata> send(ProducerRecord<K, V> record); @Override Flux<PartitionInfo> partitionsFor(String topic); @Override Mono<Void> flush(); @Override Mono<Void> close(); @Override Mono<Void> close(long timeout); @Override @SuppressWarnings("unchecked") // Mutiny API returns KafkaProducer without generics Mono<T> doOnVertxProducer(Function<io.vertx.kafka.client.producer.KafkaProducer<K, V>, T> function); @Override KafkaProducer<K, V> exceptionHandler(Consumer<Throwable> handler); @Override KafkaProducer<K, V> drainHandler(Consumer<Void> handler); @Override KafkaProducer<K, V> setWriteQueueMaxSize(int maxSize); @Override boolean writeQueueFull(); @Override Mono<Void> write(ProducerRecord<K, V> record); @Override Mono<Void> end(); @Override WriteStream vertxWriteStream(); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> flush() { return Mono.create(sink -> { try { delegate.flush(v -> sink.success()); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaProducer(io.vertx.mutiny.kafka.client.producer.KafkaProducer<K, V> delegate); @Override Mono<RecordMetadata> send(ProducerRecord<K, V> record); @Override Flux<PartitionInfo> partitionsFor(String topic); @Override Mono<Void> flush(); @Override Mono<Void> close(); @Override Mono<Void> close(long timeout); @Override @SuppressWarnings("unchecked") // Mutiny API returns KafkaProducer without generics Mono<T> doOnVertxProducer(Function<io.vertx.kafka.client.producer.KafkaProducer<K, V>, T> function); @Override KafkaProducer<K, V> exceptionHandler(Consumer<Throwable> handler); @Override KafkaProducer<K, V> drainHandler(Consumer<Void> handler); @Override KafkaProducer<K, V> setWriteQueueMaxSize(int maxSize); @Override boolean writeQueueFull(); @Override Mono<Void> write(ProducerRecord<K, V> record); @Override Mono<Void> end(); @Override WriteStream vertxWriteStream();  }