ServerSentEventsWordCountTopology.java

@Component
public class ServerSentEventsWordCountTopology 
        // (O) Extend EventStreamSupport or implement the EventStreamProvider interface
        extends EventStreamSupport 
        implements TopologyProvider {

    private EventStream<String, Long> wordCountStream;

    public ServerSentEventsWordCountTopology() {
        // (1) Create a new EventStream for word counts updates.
        wordCountStream = new EventStream.Builder("word-count")
            .withQueueSize(10_000)
            .withQueueLimitHandler(LimitHandlers.dropHeadOnLimitReached())
            .build();
        // (2) Register the EventStream
        addEventStream(wordCountStream);
    }

    @Override
    public String version() {
        return Version.getVersion();
    }

    @Override
    public Topology topology() {
        var builder = new StreamsBuilder();
        builder.<String, String>stream("text-lines")
            .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
            .groupBy((key, value) -> value)
            .count(Materialized.as("WordCount"))
            .toStream()
            // (3) send record to the EventStream      
            .foreach( (k, v) -> wordCountStream.send(k, v));  

        return builder.build();
    }
}