The equivilent class of Reactor Sinks

[hantsy]

In Spring project, I used Sinks to emit events into a SSE endpoint, it worked well, check:

https://github.com/hantsy/spring-graphql-sample/blob/master/dgs-subscription-sse/src/main/kotlin/com/example/demo/DataFetchers.kt#L33

But when I tried to MultiEmitterProcessor to archive the same purpose, it failed.

The example project is https://github.com/hantsy/quarkus-sandbox/tree/master/jms

MultiEmitterProcessor<Message> emitterProcessor = MultiEmitterProcessor.create();


    @Scheduled(delay = 500L, delayUnit = TimeUnit.MILLISECONDS, every = "1s")
    void receive() {
        var consumer = jmsContext.createConsumer(helloQueue);
//        consumer.setMessageListener(
//                msg -> {
//                    try {
//                        var received = jsonb.fromJson(msg.getBody(String.class), Message.class);
//                        LOGGER.log(Level.INFO, "consuming message: {0}", received);
//                        emitterProcessor.emit(received);
//                    } catch (JMSException e) {
//                        throw new RuntimeException(e);
//                    }
//                }
//        );
        var message = consumer.receiveBody(String.class, 500);
        if (message != null) {
            var received = jsonb.fromJson(message, Message.class);
            LOGGER.log(Level.INFO, "received message: {0}", received);
            emitterProcessor.emit(received);
        }
    }

And in the Resource class,

    @GET
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @RestStreamElementType(MediaType.APPLICATION_JSON)
    public Multi<Message> stream() {
        // see: https://github.com/quarkusio/quarkus/issues/35220
        return handler.emitterProcessor.toMulti().toHotStream();
    }

1. I am not sure MultiEmitterProcessor is good to work as a hot stream?

2. or there are something like Reactor ConnectableFlux to connect to a hot stream manually?

[jponge]

You might look into https://smallrye.io/smallrye-mutiny/2.3.1/guides/integrate-a-non-reactive-source/

[hantsy]

Tried it, https://github.com/hantsy/quarkus-sandbox/blob/master/jms/src/main/java/com/example/MessageConsumer.java#L33

Did not work.

[hantsy]

Is there an example of demonstrating the connectable hot stream usage, I have tried the approach in docs, it does not work in my case.

[jponge]

1. First and foremost Mutiny is Mutiny, Reactor is Reactor, so don't expect a 1-1 match for Reactor concepts.
2. What exactly is not working?
3. The proper way to push events is through an emitter: https://smallrye.io/smallrye-mutiny/2.5.1/tutorials/creating-multi-pipelines/#creating-multis-from-ticks-advanced (Mutiny) and https://smallrye.io/smallrye-reactive-messaging/4.9.0/concepts/emitter/ (Reactive Messaging).

[hantsy]

OK, let's forget the 1-1 mapping, how to connect to a hotstream that can emit items as needed in my example.

From the API docs, it seems MultiEmitterProcessor or UnicastProcessor are good matches.

What exactly is not working?

The SSE endpoint is not kept connecting(when using curl to access, it quits immediately) and show the new items emitted into the stream.

[jponge]

Are you running curl with the -N flag? (curl -N http://localhost:8080/my-sse-endpoint)?

Are you using RestEasy Reactive in a similar way as https://quarkus.io/guides/resteasy-reactive#server-sent-event-sse-support for doing server-side events?

Now if you really want to use MultiProcessorEmitter here's how you could do it:

package org.acme;

import org.jboss.resteasy.reactive.RestStreamElementType;

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.helpers.MultiEmitterProcessor;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

@Path("/hello")
public class GreetingResource {

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    public String hello() {
        emitter.emit("Yolo");
        return "Hello from RESTEasy Reactive";
    }

    MultiEmitterProcessor<String> emitter = MultiEmitterProcessor.create();
    Multi<String> sseBroadcast = emitter.toMulti().broadcast().toAllSubscribers();

    @Path("/sse")
    @RestStreamElementType(MediaType.APPLICATION_JSON)
    public Multi<String> stream() {
        return sseBroadcast;
    }
}

The trick is to make sure there's 1 Multi created out of the emitter, and that Multi is a broadcaster so it can be subscribed to multiple times. If you call .toMulti() for each new SSE connection request then only the first subscription will be correct, the next ones will be cancelled.

You can run this example with multiple SSE consumers (e.g., http --stream :8080/hello/sse) and each time you make a request (e.g., http :8080/hello) a new SSE event is sent to all subscribers.

[hantsy]

Thanks for this comprehensive explanation.

I change my original version which declared emitter.toMulti() as a property in MessageConsumer bean (I really do not why it did not work when calling a bean's property in resource) and used a Kotlin fun to wrap the emitter.toMulti(), it works.

https://github.com/hantsy/quarkus-sandbox/blob/master/jms/src/main/java/com/example/MessageConsumer.java#L36-L38

Of course, as shown in your codes, broadcasting to multiple subscribers is also working.