@Test public void shouldDoOnVertxConsumer() { given(mockMutinyConsumer.getDelegate()) .willReturn(mockVertxConsumer); AtomicReference<io.vertx.kafka.client.consumer.KafkaConsumer<Integer, String>> vertxConsumer = new AtomicReference<>(); Function<io.vertx.kafka.client.consumer.KafkaConsumer<Integer, String>, Boolean> function = vc -> { vertxConsumer.set(vc); return true; }; StepVerifier.create(consumer.doOnVertxConsumer(function)) .expectNext(true) .verifyComplete(); assertThat(vertxConsumer.get()).isEqualTo(mockVertxConsumer); }
@Override @SuppressWarnings("unchecked") public <T> Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function) { Objects.requireNonNull(function, "Function cannot be null"); return Mono.create(sink -> { try { T result = function.apply((io.vertx.kafka.client.consumer.KafkaConsumer<K, V>) delegate.getDelegate()); sink.success(result); } catch (Throwable t) { sink.error(t); } }); }
SnowdropKafkaConsumer implements KafkaConsumer<K, V> { @Override @SuppressWarnings("unchecked") public <T> Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function) { Objects.requireNonNull(function, "Function cannot be null"); return Mono.create(sink -> { try { T result = function.apply((io.vertx.kafka.client.consumer.KafkaConsumer<K, V>) delegate.getDelegate()); sink.success(result); } catch (Throwable t) { sink.error(t); } }); } }
SnowdropKafkaConsumer implements KafkaConsumer<K, V> { @Override @SuppressWarnings("unchecked") public <T> Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function) { Objects.requireNonNull(function, "Function cannot be null"); return Mono.create(sink -> { try { T result = function.apply((io.vertx.kafka.client.consumer.KafkaConsumer<K, V>) delegate.getDelegate()); sink.success(result); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaConsumer(io.vertx.mutiny.kafka.client.consumer.KafkaConsumer<K, V> delegate); }
SnowdropKafkaConsumer implements KafkaConsumer<K, V> { @Override @SuppressWarnings("unchecked") public <T> Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function) { Objects.requireNonNull(function, "Function cannot be null"); return Mono.create(sink -> { try { T result = function.apply((io.vertx.kafka.client.consumer.KafkaConsumer<K, V>) delegate.getDelegate()); sink.success(result); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaConsumer(io.vertx.mutiny.kafka.client.consumer.KafkaConsumer<K, V> delegate); @Override Mono<Void> subscribe(String topic); @Override Mono<Void> subscribe(Collection<String> topics); @Override Mono<Void> assign(Partition partition); @Override Mono<Void> assign(Collection<Partition> partitions); @Override Mono<Void> unsubscribe(); @Override Flux<String> subscriptions(); @Override Flux<Partition> assignments(); @Override Flux<PartitionInfo> partitionsFor(String topic); @Override void partitionsRevokedHandler(Consumer<Set<Partition>> handler); @Override void partitionsAssignedHandler(Consumer<Set<Partition>> handler); @Override Mono<Void> seek(Partition partition, long offset); @Override Mono<Void> seekToBeginning(Partition partition); @Override Mono<Void> seekToBeginning(Collection<Partition> partitions); @Override Mono<Void> seekToEnd(Partition partition); @Override Mono<Void> seekToEnd(Collection<Partition> partitions); @Override Mono<Long> position(Partition partition); @Override Mono<Long> committed(Partition partition); @Override Mono<Long> beginningOffset(Partition partition); @Override Mono<Long> endOffset(Partition partition); @Override Mono<Long> timeOffset(Partition partition, long timestamp); @Override Mono<Void> commit(); @Override Mono<Void> close(); @Override @SuppressWarnings("unchecked") // SmallRye API returns KafkaConsumer without generics Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function); @Override Mono<ConsumerRecord<K, V>> mono(); @Override Flux<ConsumerRecord<K, V>> flux(); @Override ReadStream vertxReadStream(); }
SnowdropKafkaConsumer implements KafkaConsumer<K, V> { @Override @SuppressWarnings("unchecked") public <T> Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function) { Objects.requireNonNull(function, "Function cannot be null"); return Mono.create(sink -> { try { T result = function.apply((io.vertx.kafka.client.consumer.KafkaConsumer<K, V>) delegate.getDelegate()); sink.success(result); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaConsumer(io.vertx.mutiny.kafka.client.consumer.KafkaConsumer<K, V> delegate); @Override Mono<Void> subscribe(String topic); @Override Mono<Void> subscribe(Collection<String> topics); @Override Mono<Void> assign(Partition partition); @Override Mono<Void> assign(Collection<Partition> partitions); @Override Mono<Void> unsubscribe(); @Override Flux<String> subscriptions(); @Override Flux<Partition> assignments(); @Override Flux<PartitionInfo> partitionsFor(String topic); @Override void partitionsRevokedHandler(Consumer<Set<Partition>> handler); @Override void partitionsAssignedHandler(Consumer<Set<Partition>> handler); @Override Mono<Void> seek(Partition partition, long offset); @Override Mono<Void> seekToBeginning(Partition partition); @Override Mono<Void> seekToBeginning(Collection<Partition> partitions); @Override Mono<Void> seekToEnd(Partition partition); @Override Mono<Void> seekToEnd(Collection<Partition> partitions); @Override Mono<Long> position(Partition partition); @Override Mono<Long> committed(Partition partition); @Override Mono<Long> beginningOffset(Partition partition); @Override Mono<Long> endOffset(Partition partition); @Override Mono<Long> timeOffset(Partition partition, long timestamp); @Override Mono<Void> commit(); @Override Mono<Void> close(); @Override @SuppressWarnings("unchecked") // SmallRye API returns KafkaConsumer without generics Mono<T> doOnVertxConsumer(Function<io.vertx.kafka.client.consumer.KafkaConsumer<K, V>, T> function); @Override Mono<ConsumerRecord<K, V>> mono(); @Override Flux<ConsumerRecord<K, V>> flux(); @Override ReadStream vertxReadStream();  }