rsocket-core 1.1.4 · io.rsocket.core.RSocketClient

Contract for performing RSocket requests.

RSocketClient differs from RSocket in a number of ways:

RSocket represents a "live" connection that is transient and needs to be obtained typically from a Mono<RSocket> source via flatMap or block. By contrast, RSocketClient is a higher level layer that contains such a source of connections and transparently obtains and re-obtains a shared connection as needed when requests are made concurrently. That means an RSocketClient can simply be created once, even before a connection is established, and shared as a singleton across multiple places as you would with any other client.
For request input RSocket accepts an instance of Payload and does not allow more than one subscription per request because there is no way to safely re-use that input. By contrast RSocketClient accepts Publisher<Payload> and allow re-subscribing which repeats the request.
RSocket can be used for sending and it can also be implemented for receiving. By contrast RSocketClient is used only for sending, typically from the client side which allows obtaining and re-obtaining connections from a source as needed. However it can also be used from the server side by wrapping the "live" RSocket for a given connection.

The example below shows how to create an RSocketClient:

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);

The below configures retry logic to use when a shared RSocket connection is obtained:

Mono<RSocket> source =
        RSocketConnector.create()
                .metadataMimeType("message/x.rsocket.composite-metadata.v0")
                .dataMimeType("application/cbor")
                .reconnect(Retry.fixedDelay(3, Duration.ofSeconds(1)))
                .connect(TcpClientTransport.create("localhost", 7000));

RSocketClient client = RSocketClient.from(source);

Method Summary

Modifier and Type	Method and Description
public default boolean	Returns: `true` if an attempt to connect was triggered or if already connected, or `false` if the client is terminated. connect() Connect to the remote rsocket endpoint, if not yet connected.
public Mono<Void>	fireAndForget(Mono<Payload> payloadMono) Perform a Fire-and-Forget interaction via `RSocket#fireAndForget(Payload)`.
public static RSocketClient	Returns: the created client instance from(Mono<RSocket> the source for connections, typically prepared via `RSocketConnector`. source) Create an `RSocketClient` that obtains shared connections as needed, when requests are made, from the given `Mono<RSocket>` source.
public static RSocketClient	Returns: the created client instance from(RSocket the `RSocket` to perform requests with rsocket) Adapt the given `RSocket` to use as `RSocketClient`.
public Mono<Void>	metadataPush(Mono<Payload> payloadMono) Perform a Metadata Push via `RSocket#metadataPush(Payload)`.
public default Mono<Void>	onClose() Implements io.rsocket.Closeable.onClose. Returns a `Mono` that terminates when the instance is terminated by any reason.
public Flux<Payload>	requestChannel(Publisher<Payload> payloads) Perform a Request-Channel interaction via `RSocket#requestChannel(Publisher)`.
public Mono<Payload>	requestResponse(Mono<Payload> payloadMono) Perform a Request-Response interaction via `RSocket#requestResponse(Payload)`.
public Flux<Payload>	requestStream(Mono<Payload> payloadMono) Perform a Request-Stream interaction via `RSocket#requestStream(Payload)`.
public Mono<RSocket>	source() Return the underlying source used to obtain a shared `RSocket` connection.

Method Detail

connect back to summary

connect	back to summary
public default boolean connect() Connect to the remote rsocket endpoint, if not yet connected. This method is a shortcut for `RSocketClient#source().subscribe()`. Returns:boolean `true` if an attempt to connect was triggered or if already connected, or `false` if the client is terminated.

public default boolean connect()

Connect to the remote rsocket endpoint, if not yet connected. This method is a shortcut for RSocketClient#source().subscribe().

Returns:boolean: true if an attempt to connect was triggered or if already connected, or false if the client is terminated.

fireAndForget	back to summary
public Mono<Void> fireAndForget(Mono<Payload> payloadMono) Perform a Fire-and-Forget interaction via `RSocket#fireAndForget(Payload)`. Allows multiple subscriptions and performs a request per subscriber.

from back to summary

from	back to summary
public static RSocketClient from(Mono<RSocket> source) Create an `RSocketClient` that obtains shared connections as needed, when requests are made, from the given `Mono<RSocket>` source. Parameters source:Mono<RSocket> the source for connections, typically prepared via `RSocketConnector`. Returns:RSocketClient the created client instance

public static RSocketClient from(Mono<RSocket> source)

Create an RSocketClient that obtains shared connections as needed, when requests are made, from the given Mono<RSocket> source.

Parameters

source:Mono<RSocket>: the source for connections, typically prepared via RSocketConnector.

Returns:RSocketClient

the created client instance

from back to summary

from	back to summary
public static RSocketClient from(RSocket rsocket) Adapt the given `RSocket` to use as `RSocketClient`. This is useful to wrap the sending `RSocket` in a server. Note unlike an `RSocketClient` created via `RSocketClient#from(Mono)`, the instance returned from this factory method can only perform requests for as long as the given `RSocket` remains "live". Parameters rsocket:RSocket the `RSocket` to perform requests with Returns:RSocketClient the created client instance

public static RSocketClient from(RSocket rsocket)

Adapt the given RSocket to use as RSocketClient. This is useful to wrap the sending RSocket in a server.

Note

unlike an RSocketClient created via RSocketClient#from(Mono), the instance returned from this factory method can only perform requests for as long as the given RSocket remains "live".

Parameters

rsocket:RSocket: the RSocket to perform requests with

Returns:RSocketClient

the created client instance

metadataPush	back to summary
public Mono<Void> metadataPush(Mono<Payload> payloadMono) Perform a Metadata Push via `RSocket#metadataPush(Payload)`. Allows multiple subscriptions and performs a request per subscriber.

onClose back to summary

onClose	back to summary
public default Mono<Void> onClose() Implements io.rsocket.Closeable.onClose. Doc from io.rsocket.Closeable.onClose. Returns a `Mono` that terminates when the instance is terminated by any reason. Note, in case of error termination, the cause of error will be propagated as an error signal through `org.reactivestreams.Subscriber#onError(Throwable)`. Otherwise, `Subscriber#onComplete()` will be called. Returns:Mono<Void> a `Mono` to track completion with success or error of the underlying resource. When the underlying resource is an `RSocket`, the `Mono` exposes stream 0 (i.e. connection level) errors.

public default Mono<Void> onClose()

Implements io.rsocket.Closeable.onClose.

Doc from io.rsocket.Closeable.onClose.

Returns a Mono that terminates when the instance is terminated by any reason. Note, in case of error termination, the cause of error will be propagated as an error signal through org.reactivestreams.Subscriber#onError(Throwable). Otherwise, Subscriber#onComplete() will be called.

Returns:Mono<Void>: a Mono to track completion with success or error of the underlying resource. When the underlying resource is an `RSocket`, the Mono exposes stream 0 (i.e. connection level) errors.

requestChannel	back to summary
public Flux<Payload> requestChannel(Publisher<Payload> payloads) Perform a Request-Channel interaction via `RSocket#requestChannel(Publisher)`. Allows multiple subscriptions and performs a request per subscriber.

requestResponse	back to summary
public Mono<Payload> requestResponse(Mono<Payload> payloadMono) Perform a Request-Response interaction via `RSocket#requestResponse(Payload)`. Allows multiple subscriptions and performs a request per subscriber.

requestStream	back to summary
public Flux<Payload> requestStream(Mono<Payload> payloadMono) Perform a Request-Stream interaction via `RSocket#requestStream(Payload)`. Allows multiple subscriptions and performs a request per subscriber.

source	back to summary
public Mono<RSocket> source() Return the underlying source used to obtain a shared `RSocket` connection.

public Interface RSocketClient

Method Summary

Method Detail