Top Description Fields Constructors Methods
io.rsocket.core

public Class RSocketConnector

extends Object
Class Inheritance
Static Imports
io.rsocket.core.FragmentationUtils.assertMtu, .PayloadValidationUtils.assertValidateSetup, .ReassemblyUtils.assertInboundPayloadSize
The main class to use to establish a connection to an RSocket server.

For using TCP using default settings: 

import io.rsocket.transport.netty.client.TcpClientTransport;

Mono<RSocket> source =
        RSocketConnector.connectWith(TcpClientTransport.create("localhost", 7000));
RSocketClient client = RSocketClient.from(source);

To customize connection settings before connecting: 

Mono<RSocket> source =
        RSocketConnector.create()
                .metadataMimeType("message/x.rsocket.composite-metadata.v0")
                .dataMimeType("application/cbor")
                .connect(TcpClientTransport.create("localhost", 7000));
RSocketClient client = RSocketClient.from(source);

Field Summary

Modifier and TypeField and Description
private SocketAcceptor
acceptor

private static final String
CLIENT_TAG

private String
dataMimeType

private InitializingInterceptorRegistry
interceptors

private static final BiConsumer<RSocket, Invalidatable>
INVALIDATE_FUNCTION

private Duration
keepAliveInterval

private Duration
keepAliveMaxLifeTime

private Consumer<LeaseSpec>
leaseConfigurer

private int
maxInboundPayloadSize

private String
metadataMimeType

private int
mtu

private PayloadDecoder
payloadDecoder

private Resume
resume

private Retry
retrySpec

private Mono<Payload>
setupPayloadMono

Constructor Summary

AccessConstructor and Description
private
RSocketConnector()

Method Summary

Modifier and TypeMethod and Description
public RSocketConnector

Returns:

the same instance for method chaining
acceptor(SocketAcceptor
the acceptor to use for responding to server requests
acceptor)

Configure a client-side SocketAcceptor for responding to requests from the server.

public Mono<RSocket>

Returns:

a Mono with the connected RSocket
connect(ClientTransport
the transport of choice to connect with
transport)

Connect with the given transport and obtain a live RSocket to use for making requests.

public Mono<RSocket>

Returns:

a Mono with the connected RSocket
connect(Supplier<ClientTransport>
supplier for the transport to connect with
transportSupplier)

Variant of connect(ClientTransport) with a Supplier for the ClientTransport.

public static Mono<RSocket>

Returns:

a Mono with the connected RSocket
connectWith(ClientTransport
the transport of choice to connect with
transport)

Static factory method to connect with default settings, effectively a shortcut for: 

RSocketConnector.create().connect(transport);
public static RSocketConnector
create()

Static factory method to create an RSocketConnector instance and customize default settings before connecting.

public RSocketConnector

Returns:

the same instance for method chaining
dataMimeType(String
the MIME type to be used for payload data
dataMimeType)

Set the MIME type to use for formatting payload data on the established connection.

public RSocketConnector

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 broken down into fragments to fit that size.

public RSocketConnector

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 RSocketConnector

Returns:

the same instance for method chaining
keepAlive(Duration
how frequently to emit KEEPALIVE frames
interval, Duration
how long to allow between KEEPALIVE frames from the remote end before assuming that connectivity is lost; the value should be generous and allow for multiple missed KEEPALIVE frames.
maxLifeTime)

Set the "Time Between KEEPALIVE Frames" which is how frequently KEEPALIVE frames should be emitted, and the "Max Lifetime" which is how long to allow between KEEPALIVE frames from the remote end before concluding that connectivity is lost.

public RSocketConnector

Returns:

the same instance for method chaining
lease()

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 RSocketConnector

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 RSocketConnector

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 RSocketConnector

Returns:

the same instance for method chaining
metadataMimeType(String
the MIME type to be used for payload metadata
metadataMimeType)

Set the MIME type to use for formatting payload metadata on the established connection.

public RSocketConnector

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.

public RSocketConnector

Returns:

the same instance for method chaining
reconnect(Retry
a retry spec that declares the rules for reconnecting
retry)

When this is enabled, the connect methods of this class return a special Mono<RSocket> that maintains a single, shared RSocket for all subscribers: 

Mono<RSocket> rsocketMono =
  RSocketConnector.create()
          .reconnect(Retry.fixedDelay(3, Duration.ofSeconds(1)))
          .connect(transport);

 RSocket r1 = rsocketMono.block();
 RSocket r2 = rsocketMono.block();

 assert r1 == r2;
public RSocketConnector

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 resumed automatically.

public RSocketConnector

Returns:

the same instance for method chaining
setupPayload(Mono<Payload>
the payload with data and/or metadata for the SETUP frame.
setupPayloadMono)

Provide a Mono from which to obtain the Payload for the initial SETUP frame.

public RSocketConnector

Returns:

the same instance for method chaining
setupPayload(Payload
the payload with data and/or metadata for the SETUP frame.
payload)

Variant of setupPayload(Mono) that accepts a Payload instance.

Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAlltoStringwaitwaitwait

Field Detail

acceptorback to summary
private SocketAcceptor acceptor
Annotations
@Nullable
CLIENT_TAGback to summary
private static final String CLIENT_TAG
dataMimeTypeback to summary
private String dataMimeType
interceptorsback to summary
private InitializingInterceptorRegistry interceptors
INVALIDATE_FUNCTIONback to summary
private static final BiConsumer<RSocket, Invalidatable> INVALIDATE_FUNCTION
keepAliveIntervalback to summary
private Duration keepAliveInterval
keepAliveMaxLifeTimeback to summary
private Duration keepAliveMaxLifeTime
leaseConfigurerback to summary
private Consumer<LeaseSpec> leaseConfigurer
Annotations
@Nullable
maxInboundPayloadSizeback to summary
private int maxInboundPayloadSize
metadataMimeTypeback to summary
private String metadataMimeType
mtuback to summary
private int mtu
payloadDecoderback to summary
private PayloadDecoder payloadDecoder
resumeback to summary
private Resume resume
retrySpecback to summary
private Retry retrySpec
setupPayloadMonoback to summary
private Mono<Payload> setupPayloadMono

Constructor Detail

RSocketConnectorback to summary
private RSocketConnector()

Method Detail

acceptorback to summary
public RSocketConnector acceptor(SocketAcceptor acceptor)

Configure a client-side SocketAcceptor for responding to requests from the server.

A full-form example with access to the SETUP frame and the "sending" RSocket (the same as the one returned from connect(ClientTransport)): 

Mono<RSocket> rsocketMono =
    RSocketConnector.create()
            .acceptor((setup, sendingRSocket) -> Mono.just(new RSocket() {...}))
            .connect(transport);

A shortcut example with just the handling RSocket: 

Mono<RSocket> rsocketMono =
    RSocketConnector.create()
            .acceptor(SocketAcceptor.with(new RSocket() {...})))
            .connect(transport);

A shortcut example handling only request-response: 

Mono<RSocket> rsocketMono =
    RSocketConnector.create()
            .acceptor(SocketAcceptor.forRequestResponse(payload -> ...))
            .connect(transport);

By default, new RSocket(){} is used which rejects all requests from the server with UnsupportedOperationException.

Parameters
acceptor:SocketAcceptor

the acceptor to use for responding to server requests

Returns:RSocketConnector

the same instance for method chaining

connectback to summary
public Mono<RSocket> connect(ClientTransport transport)

Connect with the given transport and obtain a live RSocket to use for making requests. Each subscriber to the returned Mono receives a new connection, if neither reconnect nor resume(Resume) are enabled.

The following transports are available through additional RSocket Java modules:

Parameters
transport:ClientTransport

the transport of choice to connect with

Returns:Mono<RSocket>

a Mono with the connected RSocket

connectback to summary
public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier)

Variant of connect(ClientTransport) with a Supplier for the ClientTransport.

// TODO: when to use?

Parameters
transportSupplier:Supplier<ClientTransport>

supplier for the transport to connect with

Returns:Mono<RSocket>

a Mono with the connected RSocket

connectWithback to summary
public static Mono<RSocket> connectWith(ClientTransport transport)

Static factory method to connect with default settings, effectively a shortcut for: 

RSocketConnector.create().connect(transport);
Parameters
transport:ClientTransport

the transport of choice to connect with

Returns:Mono<RSocket>

a Mono with the connected RSocket

createback to summary
public static RSocketConnector create()

Static factory method to create an RSocketConnector instance and customize default settings before connecting. To connect only, use connectWith(ClientTransport).

dataMimeTypeback to summary
public RSocketConnector dataMimeType(String dataMimeType)

Set the MIME type to use for formatting payload data on the established connection. This is set in the initial SETUP frame sent to the server.

By default this is set to "application/binary".

Parameters
dataMimeType:String

the MIME type to be used for payload data

Returns:RSocketConnector

the same instance for method chaining

See Also
SETUP Frame
fragmentback to summary
public RSocketConnector fragment(int mtu)

When this is set, frames larger than the given maximum transmission unit (mtu) size value are broken down into fragments to fit that size.

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:RSocketConnector

the same instance for method chaining

See Also
Fragmentation and Reassembly
interceptorsback to summary
public RSocketConnector 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:RSocketConnector

the same instance for method chaining

See Also
io.rsocket.plugins.LimitRateInterceptor
keepAliveback to summary
public RSocketConnector keepAlive(Duration interval, Duration maxLifeTime)

Set the "Time Between KEEPALIVE Frames" which is how frequently KEEPALIVE frames should be emitted, and the "Max Lifetime" which is how long to allow between KEEPALIVE frames from the remote end before concluding that connectivity is lost. Both settings are specified in the initial SETUP frame sent to the server. The spec mentions the following:

  • For server-to-server connections, a reasonable time interval between client KEEPALIVE frames is 500ms.
  • For mobile-to-server connections, the time interval between client KEEPALIVE frames is often > 30,000ms.

By default these are set to 20 seconds and 90 seconds respectively.

Parameters
interval:Duration

how frequently to emit KEEPALIVE frames

maxLifeTime:Duration

how long to allow between KEEPALIVE frames from the remote end before assuming that connectivity is lost; the value should be generous and allow for multiple missed KEEPALIVE frames.

Returns:RSocketConnector

the same instance for method chaining

See Also
SETUP Frame
leaseback to summary
public RSocketConnector lease()

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.

Example usage: 

Mono<RSocket> rocketMono =
        RSocketConnector.create()
                        .lease()
                        .connect(transport);

By default this is not enabled.

Returns:RSocketConnector

the same instance for method chaining

See Also
Lease Semantics
leaseback to summary
public RSocketConnector 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.

Example usage: 

Mono<RSocket> rocketMono =
        RSocketConnector.create()
                        .lease(spec -> spec.maxPendingRequests(128))
                        .connect(transport);

By default this is not enabled.

Parameters
leaseConfigurer:Consumer<LeaseSpec>

consumer which accepts LeaseSpec and use it for configuring

Returns:RSocketConnector

the same instance for method chaining

See Also
Lease Semantics
maxInboundPayloadSizeback to summary
public RSocketConnector 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:RSocketConnector

the same instance for method chaining

See Also
Fragmentation and Reassembly
metadataMimeTypeback to summary
public RSocketConnector metadataMimeType(String metadataMimeType)

Set the MIME type to use for formatting payload metadata on the established connection. This is set in the initial SETUP frame sent to the server.

For metadata encoding, consider using one of the following encoders:

For more on the above metadata formats, see the corresponding protocol extensions

By default this is set to "application/binary".

Parameters
metadataMimeType:String

the MIME type to be used for payload metadata

Returns:RSocketConnector

the same instance for method chaining

See Also
SETUP Frame
payloadDecoderback to summary
public RSocketConnector 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:RSocketConnector

the same instance for method chaining

reconnectback to summary
public RSocketConnector reconnect(Retry retry)

When this is enabled, the connect methods of this class return a special Mono<RSocket> that maintains a single, shared RSocket for all subscribers: 

Mono<RSocket> rsocketMono =
  RSocketConnector.create()
          .reconnect(Retry.fixedDelay(3, Duration.ofSeconds(1)))
          .connect(transport);

 RSocket r1 = rsocketMono.block();
 RSocket r2 = rsocketMono.block();

 assert r1 == r2;

The RSocket remains cached until the connection is lost and after that, new attempts to subscribe or re-subscribe trigger a reconnect and result in a new shared RSocket: 

Mono<RSocket> rsocketMono =
  RSocketConnector.create()
          .reconnect(Retry.fixedDelay(3, Duration.ofSeconds(1)))
          .connect(transport);

 RSocket r1 = rsocketMono.block();
 RSocket r2 = rsocketMono.block();

 r1.dispose();

 RSocket r3 = rsocketMono.block();
 RSocket r4 = rsocketMono.block();

 assert r1 == r2;
 assert r3 == r4;
 assert r1 != r3;

Downstream subscribers for individual requests still need their own retry logic to determine if or when failed requests should be retried which in turn triggers the shared reconnect: 

Mono<RSocket> rocketMono =
  RSocketConnector.create()
          .reconnect(Retry.fixedDelay(3, Duration.ofSeconds(1)))
          .connect(transport);

 rsocketMono.flatMap(rsocket -> rsocket.requestResponse(...))
          .retryWhen(Retry.fixedDelay(1, Duration.ofSeconds(5)))
          .subscribe()

Note

this feature is mutually exclusive with resume(Resume). If both are enabled, "resume" takes precedence. Consider using "reconnect" when the server does not have "resume" enabled or supported, or when you don't need to incur the overhead of saving in-flight frames to be potentially replayed after a reconnect.

By default this is not enabled in which case a new connection is obtained per subscriber.

Parameters
retry:Retry

a retry spec that declares the rules for reconnecting

Returns:RSocketConnector

the same instance for method chaining

resumeback to summary
public RSocketConnector 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 resumed automatically. For this to work the server must also support and have the Resume capability enabled.

See Resume for settings to customize the Resume capability.

Note

this feature is mutually exclusive with reconnect(Retry). If both are enabled, "resume" takes precedence. Consider using "reconnect" when the server does not have "resume" enabled or supported, or when you don't need to incur the overhead of saving in-flight frames to be potentially replayed after a reconnect.

By default this is not enabled.

Parameters
resume:Resume

configuration for the Resume capability

Returns:RSocketConnector

the same instance for method chaining

See Also
Resuming Operation
setupPayloadback to summary
public RSocketConnector setupPayload(Mono<Payload> setupPayloadMono)

Provide a Mono from which to obtain the Payload for the initial SETUP frame. Data and metadata should be formatted according to the MIME types specified via dataMimeType(String) and metadataMimeType(String).

Parameters
setupPayloadMono:Mono<Payload>

the payload with data and/or metadata for the SETUP frame.

Returns:RSocketConnector

the same instance for method chaining

Since
1.0.2
See Also
SETUP Frame
setupPayloadback to summary
public RSocketConnector setupPayload(Payload payload)

Variant of setupPayload(Mono) that accepts a Payload instance.

Note

if the given payload is io.rsocket.util.ByteBufPayload, it is copied to a DefaultPayload and released immediately. This ensures it can re-used to obtain a connection more than once.

Parameters
payload:Payload

the payload with data and/or metadata for the SETUP frame.

Returns:RSocketConnector

the same instance for method chaining

See Also
SETUP Frame