danielkec

AtomicInteger sum = new AtomicInteger();

Flux<Integer> flux = Flux.just("1", "2", "3", "4", "5")
       .map(Integer::parseInt);

ReactiveStreams.fromPublisher(flux)
       .limit(3)
       .forEach(sum::addAndGet)
       .run()
       .whenComplete((r, t) -> System.out.println("Sum: " + sum.get()));

danielkec

@Outgoing("int-channel")
public Publisher<Integer> produceInts() {
   return Flowable.just(1, 2, 3);
}

@Incoming("int-channel")
public void consumeInt(Integer value) {
   System.out.println("Consuming Integer: " + value);
}

danielkec

<dependency>
   <groupId>io.helidon.microprofile.reactive-streams</groupId>
   <artifactId>helidon-microprofile-reactive-streams</artifactId>
   <version>${helidon.version}</version>
</dependency>

danielkec

Multi.just(1, 2, 3, 4, 5)
       .flatMap(i -> Multi.just(i, i))
       .limit(4)
       .distinct()
       .forEach(System.out::println);

> 1
> 2

danielkec

package io.helidon.reactive.jmh;

import java.util.List;
import java.util.concurrent.Flow;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import io.helidon.common.reactive.BufferedEmittingPublisher;
import io.helidon.common.reactive.Multi;
import io.helidon.common.reactive.OriginThreadPublisher;

danielkec

https://github.com/danielkec/helidon-jmh/tree/emitter-lock-comparison/src/main/java/io/helidon

Single thread test

• WIN 10
• AMD Ryzen 7 3700X 8-Core Processor, 3593 Mhz
• JMH version: 1.19
• VM version: JDK 14.0.1, VM 14.0.1+7

Benchmark                                Mode  Cnt   Score   Error  Units

danielkec

mp.messaging:

  incoming.from-kafka:
    connector: helidon-kafka
    topic: messaging-test-topic-1
    auto.offset.reset: latest
    enable.auto.commit: true

  outgoing.to-kafka:
    connector: helidon-kafka

danielkec

mp.messaging:

 incoming.from-kafka:
   connector: helidon-kafka
   topic: messaging-test-topic-1
   auto.offset.reset: latest
   enable.auto.commit: true
   group.id: example-group-1

 outgoing.to-kafka:

danielkec

<dependency>
   <groupId>io.helidon.messaging</groupId>
   <artifactId>helidon-messaging</artifactId>
</dependency>
<dependency>
   <groupId>io.helidon.messaging.kafka</groupId>
   <artifactId>helidon-messaging-kafka</artifactId>
</dependency>

danielkec

@ApplicationScoped
public class ExampleBean {

   @Incoming("from-kafka")
   public void broadcast(String payload) {
       System.out.println("Kafka says: " + payload);
   }