@Test @SuppressWarnings("unchecked") public void shouldWrite() { given(mockMutinyProducer.write(any())) .willReturn(Uni.createFrom().voidItem()); ProducerRecord<Integer, String> record = ProducerRecord .builder("topic", "value", 1) .withTimestamp(2) .withPartition(3) .withHeader(Header.create("h1", "v1")) .build(); StepVerifier.create(producer.write(record)) .verifyComplete(); ArgumentCaptor<io.vertx.mutiny.kafka.client.producer.KafkaProducerRecord<Integer, String>> mutinyRecordCaptor = ArgumentCaptor.forClass(io.vertx.mutiny.kafka.client.producer.KafkaProducerRecord.class); verify(mockMutinyProducer).write(mutinyRecordCaptor.capture()); io.vertx.mutiny.kafka.client.producer.KafkaProducerRecord<Integer, String> mutinyRecord = mutinyRecordCaptor.getValue(); assertThat(mutinyRecord.topic()).isEqualTo("topic"); assertThat(mutinyRecord.value()).isEqualTo("value"); assertThat(mutinyRecord.key()).isEqualTo(1); assertThat(mutinyRecord.timestamp()).isEqualTo(2); assertThat(mutinyRecord.partition()).isEqualTo(3); assertThat(mutinyRecord.headers()).hasSize(1); assertThat(mutinyRecord.headers().get(0).key()).isEqualTo("h1"); assertThat(mutinyRecord.headers().get(0).value().toString()).isEqualTo("v1"); }
@Override public Mono<Void> write(ProducerRecord<K, V> record) { Objects.requireNonNull(record, "Record cannot be null"); return delegate.write(toMutinyProducerRecord(record)).convert().with(UniReactorConverters.toMono()); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> write(ProducerRecord<K, V> record) { Objects.requireNonNull(record, "Record cannot be null"); return delegate.write(toMutinyProducerRecord(record)).convert().with(UniReactorConverters.toMono()); } }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> write(ProducerRecord<K, V> record) { Objects.requireNonNull(record, "Record cannot be null"); return delegate.write(toMutinyProducerRecord(record)).convert().with(UniReactorConverters.toMono()); } SnowdropKafkaProducer(io.vertx.mutiny.kafka.client.producer.KafkaProducer<K, V> delegate); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> write(ProducerRecord<K, V> record) { Objects.requireNonNull(record, "Record cannot be null"); return delegate.write(toMutinyProducerRecord(record)).convert().with(UniReactorConverters.toMono()); } 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> write(ProducerRecord<K, V> record) { Objects.requireNonNull(record, "Record cannot be null"); return delegate.write(toMutinyProducerRecord(record)).convert().with(UniReactorConverters.toMono()); } 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();  }