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:

Subscriber<Void>, WebSocketSession

Direct Known Subclasses:

JettyWebSocketSession, StandardWebSocketSession, UndertowWebSocketSession

public abstract class AbstractListenerWebSocketSession<T>
extends AbstractWebSocketSession<T>
implements 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 implements Subscriber<Void> so it can be used to subscribe to the completion of WebSocketHandler.handle(WebSocketSession).

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 info, DataBufferFactory bufferFactory) Base constructor.
AbstractListenerWebSocketSession(T delegate, String id, HandshakeInfo info, DataBufferFactory bufferFactory, reactor.core.publisher.MonoProcessor<Void> handlerCompletion) Deprecated. as of 5.3 in favor of AbstractListenerWebSocketSession(Object, String, HandshakeInfo, DataBufferFactory, Sinks.Empty)
AbstractListenerWebSocketSession(T delegate, String id, HandshakeInfo info, DataBufferFactory bufferFactory, reactor.core.publisher.Sinks.Empty<Void> handlerCompletionSink) Alternative constructor with completion sink to use to signal when the handling of the session is complete, with success or error.

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.
reactor.core.publisher.Mono<CloseStatus>	closeStatus() Provides access to the CloseStatus with which the session is closed either locally or remotely, or completes empty if the session ended without a status.
protected AbstractListenerWebSocketSession.WebSocketSendProcessor	getSendProcessor()
void	onComplete()
void	onError(Throwable ex)
void	onNext(Void aVoid)
void	onSubscribe(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(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, isOpen

Constructor Detail

AbstractListenerWebSocketSession

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

Base constructor.

Parameters:

delegate - the native WebSocket session, channel, or connection

id - the session id

info - 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.Sinks.Empty<Void> handlerCompletionSink)

Alternative constructor with completion sink to use to signal when the handling of the session is complete, with success or error.

Primarily for use with WebSocketClient to be able to communicate the end of handling.

AbstractListenerWebSocketSession

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

Deprecated. as of 5.3 in favor of AbstractListenerWebSocketSession(Object, String, HandshakeInfo, DataBufferFactory, Sinks.Empty)

Alternative constructor with completion MonoProcessor to use to signal when the handling of the session is complete, with success or error.

Primarily for use with WebSocketClient to be able to communicate the end of handling.

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(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>

closeStatus

public reactor.core.publisher.Mono<CloseStatus> closeStatus()

Description copied from interface: WebSocketSession

Provides access to the CloseStatus with which the session is closed either locally or remotely, or completes empty if the session ended without a status.

Specified by:

closeStatus in interface WebSocketSession

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(Subscription subscription)

Specified by:

onSubscribe in interface Subscriber<Void>

onNext

public void onNext(Void aVoid)

Specified by:

onNext in interface Subscriber<Void>

onError

public void onError(Throwable ex)

Specified by:

onError in interface Subscriber<Void>

onComplete

public void onComplete()

Specified by:

onComplete in interface Subscriber<Void>