Crate rsocket_rust

Source

Expand description

Official RSocket Rust implementation using Tokio.

RSocket is an application protocol providing Reactive Streams semantics.

It is a binary protocol for use on byte stream transports such as TCP, WebSockets, and Aeron.

It enables the following symmetric interaction models via async message passing over a single connection:

request/response (stream of 1)
request/stream (finite stream of many)
fire-and-forget (no response)
channel (bi-directional streams)

§A Tour of RSocket

The easiest way to get started is to use RSocket. Do this by enabling TCP transport support.

rsocket_rust = "0.5.1"
rsocket_rust_transport_tcp = "0.5.1"

# If you want to use websocket transport instead.
# rsocket_rust_transport_websocket = "0.5.1"

§Examples

A simple TCP echo server:

use rsocket_rust::prelude::*;
use rsocket_rust::utils::EchoRSocket;
use rsocket_rust::Result;
use rsocket_rust_transport_tcp::TcpServerTransport;

#[tokio::main]
async fn main() -> Result<()> {
    RSocketFactory::receive()
        .transport(TcpServerTransport::from("127.0.0.1:7878"))
        .acceptor(Box::new(|setup, socket| {
            println!("socket establish: setup={:?}", setup);
            tokio::spawn(async move {
                let req = Payload::builder().set_data_utf8("Hello World!").build();
                let res = socket.request_response(req).await.unwrap();
                println!("SERVER request CLIENT success: response={:?}", res);
            });
            // Return a responder.
            // You can write you own responder by implementing `RSocket` trait.
            Ok(Box::new(EchoRSocket))
        }))
        .on_start(Box::new(|| println!("echo server start success!")))
        .serve()
        .await
}

Connect to echo server above:

use rsocket_rust::prelude::*;
use rsocket_rust::utils::EchoRSocket;
use rsocket_rust::Result;
use rsocket_rust_transport_tcp::TcpClientTransport;

#[tokio::main]
async fn main() -> Result<()> {
    let client = RSocketFactory::connect()
        .transport(TcpClientTransport::from("127.0.0.1:7878"))
        .acceptor(Box::new(|| {
            // Return a responder.
            Box::new(EchoRSocket)
        }))
        .start()
        .await
        .expect("Connect failed!");

    let req = Payload::builder().set_data_utf8("Ping!").build();
    let res = client.request_response(req).await.expect("Requet failed!");
    println!("request success: response={:?}", res);

    Ok(())
}