org.springframework.web.reactive.socket.adapter

Class AbstractListenerWebSocketSession<T>

java.lang.Object

org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession<T>

org.springframework.web.reactive.socket.adapter.AbstractListenerWebSocketSession<T>

Type Parameters:

T - the native delegate type

All Implemented Interfaces:

org.reactivestreams.Subscriber<Void>, WebSocketSession

Direct Known Subclasses:

JettyWebSocketSession, StandardWebSocketSession, UndertowWebSocketSession

public abstract class AbstractListenerWebSocketSession<T>
extends AbstractWebSocketSession<T>
implements org.reactivestreams.Subscriber<Void>

Base class for WebSocketSession implementations that bridge between event-listener WebSocket APIs (e.g. Java WebSocket API JSR-356, Jetty, Undertow) and Reactive Streams.

Also an implementation of Subscriber<Void> so it can be used as the completion subscriber for session handling

Since:

5.0

Author:

Violeta Georgieva, Rossen Stoyanchev

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	AbstractListenerWebSocketSession.WebSocketSendProcessor Processor to send web socket messages.

Field Summary

Fields inherited from class org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession

logger

Constructor Summary

Constructors

Constructor and Description
AbstractListenerWebSocketSession(T delegate, String id, HandshakeInfo handshakeInfo, DataBufferFactory bufferFactory) Base constructor.
AbstractListenerWebSocketSession(T delegate, String id, HandshakeInfo info, DataBufferFactory bufferFactory, reactor.core.publisher.MonoProcessor<Void> completionMono) Alternative constructor with completion Mono&lt;Void&gt; to propagate the session completion (success or error) (for client-side use).

Method Summary

Modifier and Type	Method and Description
protected abstract boolean	canSuspendReceiving() Whether the underlying WebSocket API has flow control and can suspend and resume the receiving of messages.
protected AbstractListenerWebSocketSession.WebSocketSendProcessor	getSendProcessor()
void	onComplete()
void	onError(Throwable ex)
void	onNext(Void aVoid)
void	onSubscribe(org.reactivestreams.Subscription subscription)
reactor.core.publisher.Flux<WebSocketMessage>	receive() Provides access to the stream of inbound messages.
protected abstract void	resumeReceiving() Resume receiving new message(s) after demand is generated by the downstream Subscriber.
reactor.core.publisher.Mono<Void>	send(org.reactivestreams.Publisher<WebSocketMessage> messages) Give a source of outgoing messages, write the messages and return a Mono<Void> that completes when the source completes and writing is done.
protected abstract boolean	sendMessage(WebSocketMessage message) Send the given WebSocket message.
protected abstract void	suspendReceiving() Suspend receiving until received message(s) are processed and more demand is generated by the downstream Subscriber.

Methods inherited from class org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession

binaryMessage, bufferFactory, getAttributes, getDelegate, getHandshakeInfo, getId, getLogPrefix, pingMessage, pongMessage, textMessage, toString

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.springframework.web.reactive.socket.WebSocketSession

close, close

Constructor Detail

AbstractListenerWebSocketSession

public AbstractListenerWebSocketSession(T delegate,
                                        String id,
                                        HandshakeInfo handshakeInfo,
                                        DataBufferFactory bufferFactory)

Base constructor.

Parameters:

delegate - the native WebSocket session, channel, or connection

id - the session id

handshakeInfo - the handshake info

bufferFactory - the DataBuffer factor for the current connection

AbstractListenerWebSocketSession

public AbstractListenerWebSocketSession(T delegate,
                                        String id,
                                        HandshakeInfo info,
                                        DataBufferFactory bufferFactory,
                                        @Nullable
                                        reactor.core.publisher.MonoProcessor<Void> completionMono)

Alternative constructor with completion Mono<Void> to propagate the session completion (success or error) (for client-side use).

Method Detail

getSendProcessor

protected AbstractListenerWebSocketSession.WebSocketSendProcessor getSendProcessor()

receive

public reactor.core.publisher.Flux<WebSocketMessage> receive()

Description copied from interface: WebSocketSession

Provides access to the stream of inbound messages.

This stream receives a completion or error signal when the connection is closed. In a typical WebSocketHandler implementation this stream is composed into the overall processing flow, so that when the connection is closed, handling will end.

See the class-level doc of WebSocketHandler and the reference for more details and examples of how to handle the session.

Specified by:

receive in interface WebSocketSession

Specified by:

receive in class AbstractWebSocketSession<T>

send

public reactor.core.publisher.Mono<Void> send(org.reactivestreams.Publisher<WebSocketMessage> messages)

Description copied from interface: WebSocketSession

Give a source of outgoing messages, write the messages and return a Mono<Void> that completes when the source completes and writing is done.

See the class-level doc of WebSocketHandler and the reference for more details and examples of how to handle the session.

Specified by:

send in interface WebSocketSession

Specified by:

send in class AbstractWebSocketSession<T>

canSuspendReceiving

protected abstract boolean canSuspendReceiving()

Whether the underlying WebSocket API has flow control and can suspend and resume the receiving of messages.

Note: Sub-classes are encouraged to start out in suspended mode, if possible, and wait until demand is received.

suspendReceiving

protected abstract void suspendReceiving()

Suspend receiving until received message(s) are processed and more demand is generated by the downstream Subscriber.

Note: if the underlying WebSocket API does not provide flow control for receiving messages, this method should be a no-op and canSuspendReceiving() should return false.

resumeReceiving

protected abstract void resumeReceiving()

Resume receiving new message(s) after demand is generated by the downstream Subscriber.

Note: if the underlying WebSocket API does not provide flow control for receiving messages, this method should be a no-op and canSuspendReceiving() should return false.

sendMessage

protected abstract boolean sendMessage(WebSocketMessage message)
                                throws IOException

Send the given WebSocket message.

Note: Sub-classes are responsible for releasing the payload data buffer, once fully written, if pooled buffers apply to the underlying container.

Throws:

IOException

onSubscribe

public void onSubscribe(org.reactivestreams.Subscription subscription)

Specified by:

onSubscribe in interface org.reactivestreams.Subscriber<Void>

onNext

public void onNext(Void aVoid)

Specified by:

onNext in interface org.reactivestreams.Subscriber<Void>

onError

public void onError(Throwable ex)

Specified by:

onError in interface org.reactivestreams.Subscriber<Void>

onComplete

public void onComplete()

Specified by:

onComplete in interface org.reactivestreams.Subscriber<Void>