@Test public void shouldGetPartition() { PartitionInfo firstPartitionInfo = mock(PartitionInfo.class); PartitionInfo secondPartitionInfo = mock(PartitionInfo.class); given(mockMutinyProducer.partitionsFor("test-topic")) .willReturn(Uni.createFrom().item(Arrays.asList(firstPartitionInfo, secondPartitionInfo))); StepVerifier.create(producer.partitionsFor("test-topic")) .expectNext(new SnowdropPartitionInfo(firstPartitionInfo)) .expectNext(new SnowdropPartitionInfo(secondPartitionInfo)) .verifyComplete(); }
@Override public Flux<PartitionInfo> partitionsFor(String topic) { if (StringUtils.isEmpty(topic)) { throw new IllegalArgumentException("Topic cannot be empty"); } return delegate.partitionsFor(topic) .convert() .with(UniReactorConverters.toMono()) .flatMapMany(Flux::fromIterable) .map(SnowdropPartitionInfo::new); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Flux<PartitionInfo> partitionsFor(String topic) { if (StringUtils.isEmpty(topic)) { throw new IllegalArgumentException("Topic cannot be empty"); } return delegate.partitionsFor(topic) .convert() .with(UniReactorConverters.toMono()) .flatMapMany(Flux::fromIterable) .map(SnowdropPartitionInfo::new); } }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Flux<PartitionInfo> partitionsFor(String topic) { if (StringUtils.isEmpty(topic)) { throw new IllegalArgumentException("Topic cannot be empty"); } return delegate.partitionsFor(topic) .convert() .with(UniReactorConverters.toMono()) .flatMapMany(Flux::fromIterable) .map(SnowdropPartitionInfo::new); } SnowdropKafkaProducer(io.vertx.mutiny.kafka.client.producer.KafkaProducer<K, V> delegate); }
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Flux<PartitionInfo> partitionsFor(String topic) { if (StringUtils.isEmpty(topic)) { throw new IllegalArgumentException("Topic cannot be empty"); } return delegate.partitionsFor(topic) .convert() .with(UniReactorConverters.toMono()) .flatMapMany(Flux::fromIterable) .map(SnowdropPartitionInfo::new); } 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 Flux<PartitionInfo> partitionsFor(String topic) { if (StringUtils.isEmpty(topic)) { throw new IllegalArgumentException("Topic cannot be empty"); } return delegate.partitionsFor(topic) .convert() .with(UniReactorConverters.toMono()) .flatMapMany(Flux::fromIterable) .map(SnowdropPartitionInfo::new); } 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();  }