MsgProcessingBean.java

@ApplicationScoped
public class MsgProcessingBean {
    private final EmittingPublisher<String> emittingPublisher = new EmittingPublisher<>();
    private SseBroadcaster sseBroadcaster;

    @Outgoing("multiplyVariants")
    public Publisher<String> preparePublisher() {
        // Create new publisher for emitting to by this::process
        return ReactiveStreams
                .fromPublisher(emittingPublisher)
                .buildRs();
    }

    @Incoming("multiplyVariants")
    @Outgoing("wrapSseEvent")
    public ProcessorBuilder<String, String> multiply() {
        // Multiply to 3 variants of same message
        return ReactiveStreams.<String>builder()
                .flatMap(o ->
                        ReactiveStreams.of(
                                // upper case variant
                                o.toUpperCase(),
                                // repeat twice variant
                                o.repeat(2),
                                // reverse chars 'tnairav'
                                new StringBuilder(o).reverse().toString())
                );
    }

    @Incoming("wrapSseEvent")
    @Outgoing("broadcast")
    public OutboundSseEvent wrapSseEvent(String msg) {
        // Map every message to sse event
        return new OutboundEvent.Builder().data(msg).build();
    }

    @Incoming("broadcast")
    public void broadcast(OutboundSseEvent sseEvent) {
        // Broadcast to all sse sinks
        this.sseBroadcaster.broadcast(sseEvent);
    }

    public void addSink(final SseEventSink eventSink, final Sse sse) {
        if (this.sseBroadcaster == null) {
            this.sseBroadcaster = sse.newBroadcaster();
        }
        this.sseBroadcaster.register(eventSink);
    }

    public void process(final String msg) {
        emittingPublisher.emit(msg);
    }

}