rsocket-rust

rsocket-rust is an implementation of the RSocket protocol in Rust(1.39+). It's an alpha version and still under active development. Do not use it in a production environment!

Example

Here are some example codes which show how RSocket works in Rust.

Dependencies

Add dependencies in your Cargo.toml.

[dependencies]
tokio = "1.0.3"rsocket_rust = "0.7"# add transport dependencies:# rsocket_rust_transport_tcp = "0.7"# rsocket_rust_transport_websocket = "0.7"

Server

externcrate log;use futures::executor::block_on;use rsocket_rust::prelude::*;use rsocket_rust::utils::EchoRSocket;use rsocket_rust::Result;use rsocket_rust_transport_tcp::*;#[tokio::main]asyncfnmain() -> Result<()>{
env_logger::builder().format_timestamp_millis().init();RSocketFactory::receive().transport(TcpServerTransport::from("127.0.0.1:7979")).acceptor(Box::new(|setup, _sending_socket| {info!("incoming socket: setup={:?}", setup);Ok(Box::new(block_on(asyncmove{RSocketFactory::connect().transport(TcpClientTransport::from("127.0.0.1:7878")).acceptor(Box::new(|| Box::new(EchoRSocket))).setup(Payload::from("I'm Rust!")).start().await.unwrap()})))})).serve().await}

Client

externcrate log;use rsocket_rust::prelude::*;use rsocket_rust::utils::EchoRSocket;use rsocket_rust::Result;use rsocket_rust_transport_tcp::TcpClientTransport;#[tokio::main]asyncfnmain() -> Result<()>{
env_logger::builder().format_timestamp_millis().init();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();match client.request_response(req).await{Ok(res) => info!("{:?}", res),Err(e) => error!("{}", e),}Ok(())}

Implement RSocket trait

Example for access Redis(crates):

NOTICE: add dependency in Cargo.toml => redis = { version = "0.19.0", features = [ "aio" ] }

use std::str::FromStr;use redis::ClientasRedisClient;use rsocket_rust::async_trait;use rsocket_rust::prelude::*;use rsocket_rust::Result;#[derive(Clone)]pubstructRedisDao{inner:RedisClient,}// Create RedisDao from str.// Example: RedisDao::from_str("redis://127.0.0.1").expect("Connect redis failed!");implFromStrforRedisDao{typeErr = redis::RedisError;fnfrom_str(s:&str) -> std::result::Result<Self,Self::Err>{let client = redis::Client::open(s)?;Ok(RedisDao{inner: client })}}#[async_trait]implRSocketforRedisDao{asyncfnrequest_response(&self,req:Payload) -> Result<Option<Payload>>{let client = self.inner.clone();letmut conn = client.get_async_connection().await?;let value: redis::RedisResult<Option<String>> = redis::cmd("GET").arg(&[req.data_utf8()]).query_async(&mut conn).await;match value {Ok(Some(value)) => Ok(Some(Payload::builder().set_data_utf8(&value).build())),Ok(None) => Ok(None),Err(e) => Err(e.into()),}}asyncfnmetadata_push(&self,_req:Payload) -> Result<()>{todo!()}asyncfnfire_and_forget(&self,_req:Payload) -> Result<()>{todo!()}fnrequest_stream(&self,_req:Payload) -> Flux<Result<Payload>>{todo!()}fnrequest_channel(&self,_reqs:Flux<Result<Payload>>) -> Flux<Result<Payload>>{todo!()}}

TODO

• Operations
  • METADATA_PUSH
  • REQUEST_FNF
  • REQUEST_RESPONSE
  • REQUEST_STREAM
  • REQUEST_CHANNEL
• More Operations
  • Error
  • Cancel
  • Fragmentation
  • Resume
  • Keepalive
• QoS
  • RequestN
  • Lease
• Transport
  • TCP
  • Websocket
  • WASM
• Reactor
  • ...
• High Level APIs
  • Client
  • Server