Scaling reactive APIs - www.razorcodes.com

code5_EventStream.java

@Value
public class EventStream {
    private final String eventStreamId;
    private final Long offsetTimeMs;
    private final Long offsetCount;
    private final Set<FluxSink<Object>> fluxSinks;

    public void addStream(FluxSink<Object> newFluxSinkObject) {
        synchronized(this) {
            fluxSinks.add(newFluxSinkObject);
        }
    }

    public void removeStream(FluxSink<Object> existingFluxSinkObject) {
        synchronized(this) {
            fluxSinks.remove(existingFluxSinkObject);
        }
    }

    public boolean hasListeners() {
        return !fluxSinks.isEmpty();
    }

    //closes all stream connections with an error message.
    public void sendError(String message) {
       fluxSinks.forEach(fluxSink -> fluxSink.error(new ConnectionClosedException(message)));
    }

    //publish an event to all streams
    public void publishEvent(String event) {
        fluxSinks.forEach(fluxSink -> fluxSink.next(event));
    }

    //creates a new object with updated offsets
    public EventStream withOffset(Long offsetMs, Long offsetCount) {
        return new EventStream(this.eventStreamId, offsetMs, offsetCount, this.fluxSinks);
    }
}