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!
Here are some example codes which show how RSocket works in Rust.
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"extern crate 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]
async fn main() -> 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(async move {
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
}extern crate log;
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<()> {
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(())
}Example for access Redis(crates):
NOTICE: add dependency in Cargo.toml => redis = { version = "0.19.0", features = [ "aio" ] }
use std::str::FromStr;
use redis::Client as RedisClient;
use rsocket_rust::async_trait;
use rsocket_rust::prelude::*;
use rsocket_rust::Result;
#[derive(Clone)]
pub struct RedisDao {
inner: RedisClient,
}
// Create RedisDao from str.
// Example: RedisDao::from_str("redis://127.0.0.1").expect("Connect redis failed!");
impl FromStr for RedisDao {
type Err = redis::RedisError;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
let client = redis::Client::open(s)?;
Ok(RedisDao { inner: client })
}
}
#[async_trait]
impl RSocket for RedisDao {
async fn request_response(&self, req: Payload) -> Result<Option<Payload>> {
let client = self.inner.clone();
let mut 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()),
}
}
async fn metadata_push(&self, _req: Payload) -> Result<()> {
todo!()
}
async fn fire_and_forget(&self, _req: Payload) -> Result<()> {
todo!()
}
fn request_stream(&self, _req: Payload) -> Flux<Result<Payload>> {
todo!()
}
fn request_channel(&self, _reqs: Flux<Result<Payload>>) -> Flux<Result<Payload>> {
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
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent