Top Description Fields Constructors Methods
io.rsocket.core

public final Class RSocketServer

extends Object
Class Inheritance
Static Imports
io.rsocket.core.FragmentationUtils.assertMtu, .PayloadValidationUtils.assertValidateSetup, .ReassemblyUtils.assertInboundPayloadSize, io.rsocket.frame.FrameLengthCodec.FRAME_LENGTH_MASK
The main class for starting an RSocket server.

For example: 

CloseableChannel closeable =
        RSocketServer.create(SocketAcceptor.with(new RSocket() {...}))
                .bind(TcpServerTransport.create("localhost", 7000))
                .block();

Field Summary

Modifier and TypeField and Description
private SocketAcceptor
acceptor

private InitializingInterceptorRegistry
interceptors

private Consumer<LeaseSpec>
leaseConfigurer

private int
maxInboundPayloadSize

private int
mtu

private PayloadDecoder
payloadDecoder

private Resume
resume

private static final String
SERVER_TAG

private Duration
timeout

Constructor Summary

AccessConstructor and Description
private
RSocketServer()

Method Summary

Modifier and TypeMethod and Description
private Mono<Void>
accept(ServerSetup serverSetup, ByteBuf startFrame, DuplexConnection clientServerConnection, int maxFrameLength)

public RSocketServer

Returns:

the same instance for method chaining
acceptor(SocketAcceptor
the acceptor to handle incoming connections and requests with
acceptor)

Set the acceptor to handle incoming connections and handle requests.

private Mono<Void>
acceptor(ServerSetup serverSetup, DuplexConnection sourceConnection, int maxFrameLength)

private Mono<Void>
acceptResume(ServerSetup serverSetup, ByteBuf resumeFrame, DuplexConnection clientServerConnection)

private Mono<Void>
acceptSetup(ServerSetup serverSetup, ByteBuf setupFrame, DuplexConnection clientServerConnection, int maxFrameLength)

public ServerTransport.ConnectionAcceptor
asConnectionAcceptor()

An alternative to bind(ServerTransport) that is useful for installing RSocket on a server that is started independently.

public ServerTransport.ConnectionAcceptor
asConnectionAcceptor(int maxFrameLength)

An alternative to bind(ServerTransport) that is useful for installing RSocket on a server that is started independently.

public <
the type of Closeable for the given transport
T extends Closeable> Mono<T>

Returns:

a Mono with a Closeable that can be used to obtain information about the server, stop it, or be notified of when it is stopped.
bind(ServerTransport<T>
the transport of choice to connect with
transport)

Start the server on the given transport.

public <T extends Closeable> T
bindNow(ServerTransport<T> transport)

Start the server on the given transport.

public static RSocketServer
create()

Static factory method to create an RSocketServer.

public static RSocketServer

Returns:

the same instance for method chaining
create(SocketAcceptor
the acceptor to handle connections with
acceptor)

Static factory method to create an RSocketServer instance with the given SocketAcceptor.

pack-priv ServerSetup
createSetup(Duration timeout)

public RSocketServer

Returns:

the same instance for method chaining
fragment(int
the threshold size for fragmentation, must be no less than 64
mtu)

When this is set, frames larger than the given maximum transmission unit (mtu) size value are fragmented.

public RSocketServer

Returns:

the same instance for method chaining
interceptors(Consumer<InterceptorRegistry>
a configurer to customize interception with.
configurer)

Configure interception at one of the following levels:

  • Transport level
  • At the level of accepting new connections
  • Performing requests
  • Responding to requests
public RSocketServer

Returns:

the same instance for method chaining
lease(Consumer<LeaseSpec>
consumer which accepts LeaseSpec and use it for configuring
leaseConfigurer)

Enables the Lease feature of the RSocket protocol where the number of requests that can be performed from either side are rationed via LEASE frames from the responder side.

public RSocketServer

Returns:

the same instance for method chaining
maxInboundPayloadSize(int
the threshold size for reassembly, must no be less than 64 bytes. Please note, maxInboundPayloadSize must always be greater or equal to io.rsocket.transport.Transport#maxFrameLength(), otherwise inbound frame can exceed the maxInboundPayloadSize
maxInboundPayloadSize)

When this is set, frames reassembler control maximum payload size which can be reassembled.

public RSocketServer

Returns:

the same instance for method chaining
maxTimeToFirstFrame(Duration
duration
timeout)

Specify the max time to wait for the first frame (e.g. SETUP) on an accepted connection.

public RSocketServer

Returns:

the same instance for method chaining
payloadDecoder(PayloadDecoder
the decoder to use
decoder)

Configure the PayloadDecoder used to create Payload's from incoming raw frame buffers.

private RSocketErrorException
rejectedSetupError(Throwable err)

public RSocketServer

Returns:

the same instance for method chaining
resume(Resume
configuration for the Resume capability
resume)

Enables the Resume capability of the RSocket protocol where if the client gets disconnected, the connection is re-acquired and any interrupted streams are transparently resumed.

private ServerSetup
serverSetup(Duration timeout)

Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAlltoStringwaitwaitwait

Field Detail

acceptorback to summary
private SocketAcceptor acceptor
interceptorsback to summary
private InitializingInterceptorRegistry interceptors
leaseConfigurerback to summary
private Consumer<LeaseSpec> leaseConfigurer
maxInboundPayloadSizeback to summary
private int maxInboundPayloadSize
mtuback to summary
private int mtu
payloadDecoderback to summary
private PayloadDecoder payloadDecoder
resumeback to summary
private Resume resume
SERVER_TAGback to summary
private static final String SERVER_TAG
timeoutback to summary
private Duration timeout

Constructor Detail

RSocketServerback to summary
private RSocketServer()

Method Detail

acceptback to summary
private Mono<Void> accept(ServerSetup serverSetup, ByteBuf startFrame, DuplexConnection clientServerConnection, int maxFrameLength)
acceptorback to summary
public RSocketServer acceptor(SocketAcceptor acceptor)

Set the acceptor to handle incoming connections and handle requests.

An example with access to the SETUP frame and sending RSocket for performing requests back to the client if needed: 

RSocketServer.create((setup, sendingRSocket) -> Mono.just(new RSocket() {...}))
        .bind(TcpServerTransport.create("localhost", 7000))
        .subscribe();

A shortcut to provide the handling RSocket only: 

RSocketServer.create(SocketAcceptor.with(new RSocket() {...}))
        .bind(TcpServerTransport.create("localhost", 7000))
        .subscribe();

A shortcut to handle request-response interactions only: 

RSocketServer.create(SocketAcceptor.forRequestResponse(payload -> ...))
        .bind(TcpServerTransport.create("localhost", 7000))
        .subscribe();

By default, new RSocket(){} is used for handling which rejects requests from the client with UnsupportedOperationException.

Parameters
acceptor:SocketAcceptor

the acceptor to handle incoming connections and requests with

Returns:RSocketServer

the same instance for method chaining

acceptorback to summary
private Mono<Void> acceptor(ServerSetup serverSetup, DuplexConnection sourceConnection, int maxFrameLength)
acceptResumeback to summary
private Mono<Void> acceptResume(ServerSetup serverSetup, ByteBuf resumeFrame, DuplexConnection clientServerConnection)
acceptSetupback to summary
private Mono<Void> acceptSetup(ServerSetup serverSetup, ByteBuf setupFrame, DuplexConnection clientServerConnection, int maxFrameLength)
asConnectionAcceptorback to summary
public ServerTransport.ConnectionAcceptor asConnectionAcceptor()

An alternative to bind(ServerTransport) that is useful for installing RSocket on a server that is started independently.

See Also
io.rsocket.examples.transport.ws.WebSocketHeadersSample
asConnectionAcceptorback to summary
public ServerTransport.ConnectionAcceptor asConnectionAcceptor(int maxFrameLength)

An alternative to bind(ServerTransport) that is useful for installing RSocket on a server that is started independently.

See Also
io.rsocket.examples.transport.ws.WebSocketHeadersSample
bindback to summary
public <T extends Closeable> Mono<T> bind(ServerTransport<T> transport)

Start the server on the given transport.

The following transports are available from additional RSocket Java modules:

Parameters
<T>
the type of Closeable for the given transport
transport:ServerTransport<T>

the transport of choice to connect with

Returns:Mono<T>

a Mono with a Closeable that can be used to obtain information about the server, stop it, or be notified of when it is stopped.

bindNowback to summary
public <T extends Closeable> T bindNow(ServerTransport<T> transport)

Start the server on the given transport. Effectively is a shortcut for .bind(ServerTransport).block()

createback to summary
public static RSocketServer create()

Static factory method to create an RSocketServer.

createback to summary
public static RSocketServer create(SocketAcceptor acceptor)

Static factory method to create an RSocketServer instance with the given SocketAcceptor. Effectively a shortcut for: 

RSocketServer.create().acceptor(...);
Parameters
acceptor:SocketAcceptor

the acceptor to handle connections with

Returns:RSocketServer

the same instance for method chaining

See Also
acceptor(SocketAcceptor)
createSetupback to summary
pack-priv ServerSetup createSetup(Duration timeout)
fragmentback to summary
public RSocketServer fragment(int mtu)

When this is set, frames larger than the given maximum transmission unit (mtu) size value are fragmented.

By default this is not set in which case payloads are sent whole up to the maximum frame size of 16,777,215 bytes.

Parameters
mtu:int

the threshold size for fragmentation, must be no less than 64

Returns:RSocketServer

the same instance for method chaining

See Also
Fragmentation and Reassembly
interceptorsback to summary
public RSocketServer interceptors(Consumer<InterceptorRegistry> configurer)

Configure interception at one of the following levels:

  • Transport level
  • At the level of accepting new connections
  • Performing requests
  • Responding to requests
Parameters
configurer:Consumer<InterceptorRegistry>

a configurer to customize interception with.

Returns:RSocketServer

the same instance for method chaining

See Also
io.rsocket.plugins.LimitRateInterceptor
leaseback to summary
public RSocketServer lease(Consumer<LeaseSpec> leaseConfigurer)

Enables the Lease feature of the RSocket protocol where the number of requests that can be performed from either side are rationed via LEASE frames from the responder side. For this to work clients must also support and request to enable this when connecting.

Example usage: 

RSocketServer.create(SocketAcceptor.with(new RSocket() {...}))
        .lease(spec ->
           spec.sender(() -> Flux.interval(ofSeconds(1))
                                 .map(__ -> Lease.create(ofSeconds(1), 1)))
        )
        .bind(TcpServerTransport.create("localhost", 7000))
        .subscribe();

By default this is not enabled.

Parameters
leaseConfigurer:Consumer<LeaseSpec>

consumer which accepts LeaseSpec and use it for configuring

Returns:RSocketServer

the same instance for method chaining

See Also
Lease Semantics
maxInboundPayloadSizeback to summary
public RSocketServer maxInboundPayloadSize(int maxInboundPayloadSize)

When this is set, frames reassembler control maximum payload size which can be reassembled.

By default this is not set in which case maximum reassembled payloads size is not controlled.

Parameters
maxInboundPayloadSize:int

the threshold size for reassembly, must no be less than 64 bytes. Please note, maxInboundPayloadSize must always be greater or equal to io.rsocket.transport.Transport#maxFrameLength(), otherwise inbound frame can exceed the maxInboundPayloadSize

Returns:RSocketServer

the same instance for method chaining

See Also
Fragmentation and Reassembly
maxTimeToFirstFrameback to summary
public RSocketServer maxTimeToFirstFrame(Duration timeout)

Specify the max time to wait for the first frame (e.g. SETUP) on an accepted connection.

By default this is set to 1 minute.

Parameters
timeout:Duration

duration

Returns:RSocketServer

the same instance for method chaining

payloadDecoderback to summary
public RSocketServer payloadDecoder(PayloadDecoder decoder)

Configure the PayloadDecoder used to create Payload's from incoming raw frame buffers. The following decoders are available:

By default this is set to PayloadDecoder#DEFAULT in which case data and metadata are copied and do not need to be tracked and released.

Parameters
decoder:PayloadDecoder

the decoder to use

Returns:RSocketServer

the same instance for method chaining

rejectedSetupErrorback to summary
private RSocketErrorException rejectedSetupError(Throwable err)
resumeback to summary
public RSocketServer resume(Resume resume)

Enables the Resume capability of the RSocket protocol where if the client gets disconnected, the connection is re-acquired and any interrupted streams are transparently resumed. For this to work clients must also support and request to enable this when connecting.

Use the Resume argument to customize the Resume session duration, storage, retry logic, and others.

By default this is not enabled.

Parameters
resume:Resume

configuration for the Resume capability

Returns:RSocketServer

the same instance for method chaining

See Also
Resuming Operation
serverSetupback to summary
private ServerSetup serverSetup(Duration timeout)