Connecting WebSocketSessions

ChatSocketHandler.java

@Override
public Mono<Void> handle(WebSocketSession session) {
    WebSocketMessageSubscriber subscriber = new WebSocketMessageSubscriber(eventPublisher);
    session.receive()
            .map(WebSocketMessage::getPayloadAsText)
            .map(this::toEvent)
            .subscribe(subscriber::onNext, subscriber::onError, subscriber::onComplete);
    return session.send(outputEvents.map(session::textMessage));
}

private static class WebSocketMessageSubscriber {
    private UnicastProcessor<Event> eventPublisher;
    private Optional<Event> lastReceivedEvent = Optional.empty();

    public WebSocketMessageSubscriber(UnicastProcessor<Event> eventPublisher) {
        this.eventPublisher = eventPublisher;
    }

    public void onNext(Event event) {
        lastReceivedEvent = Optional.of(event);
        eventPublisher.onNext(event);
    }

    public void onError(Throwable error) {
        //TODO log error
        error.printStackTrace();
    }

    public void onComplete() {
        lastReceivedEvent.ifPresent(event -> eventPublisher.onNext(
                Event.type(USER_LEFT)
                        .withPayload()
                        .user(event.getUser())
                        .build()));
    }

}