Reactive Streams Support

[glassfishrobot]

Reactive Streams, also known as the JDK9 juc.Flow API, allows developers to plumb asynchronous streams of messages between libraries, in a way that safely ensures that backpressure, error and completion handling are correctly propagated through the streams. Providing support for Reactive Streams in the WebSocket spec would allow JSR356 WebSocket implementations to participate in this rich and growing ecosystem of asynchronous streaming libraries.

Use cases include plumbing WebSocket connections together, plumbing message broker subscribers and publishers into WebSocket streams, and integration with dedicated streaming libraries such as Akka, Reactor and RxJava, to allow developers to create complex graphs of streams.

As an example of what reactive streams support could look like, here's what it might look like to build a chat room using Kafka (using a Kafka reactive streams client such as https://github.com/unicredit/kafka-reactive-streams) as the backend for publishing and subscribing to messages:

private Flow.Subscriber<ChatMessage> createKafkaSubscriber(String room) {
  // Kafka specific code to create subscriber for consuming messages to send to Kafka
  ...
}

private Flow.Publisher<ChatMessage> createKafkaPublisher(String room) {
  // Kafka specific code to create publisher for producing messages received from Kafka
  ...
}

@OnStream
public Flow.Publisher<ChatMessage> connectToRoom(
    @PathParam("room") String room,
    Flow.Publisher<ChatMessage> incomingMessages) {
  incomingMessages.subscribe(createKafkaSubscriber(room));
  return createKafkaPublisher(room);
}

All the WebSocket specific code relating to plumbing messages, back pressure, and stream error handling can be seen in the connectToRoom method, which is invoked when the WebSocket is connected. The passed in incomingMessages publisher is the incoming WebSocket stream, each message being a chat message to publish to the room, while the returned Publisher is a stream of all the messages published to the chat room.

[glassfishrobot]

• Issue Imported From: https://github.com/javaee/websocket-spec/issues/244
• Original Issue Raised By:@jroper
• Original Issue Assigned To: Unassigned