krossbar-rpc

横杆平台用于通信的 RPC 库。

该库

• 接收 tokio::net::UnixStream 并返回 RPC 处理器；
• 允许通过 RPC 连接进行调用、订阅端点以及发送 tokio::net::UnixStream；
• 支持在重新连接后替换流、重新订阅活动订阅并保持所有客户端处理器有效；
• 支持通过 [Monitor] 监控消息交换；

使用 rpc::Rpc::poll 方法轮询流。这包括等待调用或订阅响应。

示例

RPC 调用

use futures::{select, FutureExt};
use tokio::net::UnixStream;

use krossbar_rpc::rpc::Rpc;

async fn call() {
    let stream = UnixStream::connect("/tmp/hub.sock").await.unwrap();
    let mut rpc = Rpc::new(stream);

    let call = rpc.call::<u32, u32>("echo", &42).await.unwrap();

    select! {
        response = call.fuse() => {
            println!("Call response: {response:?}")
        },
        _ = rpc.poll().fuse() => {}
    }
}

RPC 订阅

use futures::{select, FutureExt, StreamExt};
use tokio::net::UnixStream;

use krossbar_rpc::rpc::Rpc;

async fn subscribe() {
    let stream = UnixStream::connect("/tmp/hub.sock").await.unwrap();
    let mut rpc = Rpc::new(stream);

    let subscription = rpc.subscribe::<u32>("signal").await.unwrap();

    select! {
        response = subscription.take(2).collect::<Vec<krossbar_rpc::Result<u32>>>() => {
            println!("Subscription response: {response:?}")
        },
        _ = rpc.poll().fuse() => {}
    }
}

单向消息

use futures::{select, FutureExt};
use tokio::net::UnixStream;

use krossbar_rpc::rpc::Rpc;

async fn message() {
    let stream = UnixStream::connect("/tmp/hub.sock").await.unwrap();
    let mut rpc = Rpc::new(stream);

    let call = rpc.send_message("echo", &42).await.unwrap();

    let incoming_message = rpc.poll().await;
}

轮询传入消息

use futures::{select, FutureExt};
use tokio::net::UnixStream;

use krossbar_rpc::{rpc::Rpc, request::Body};

async fn poll() {
    let stream = UnixStream::connect("/tmp/hub.sock").await.unwrap();
    let mut rpc = Rpc::new(stream);

    loop {
        let request = rpc.poll().await;

        if request.is_none() {
            println!("Client disconnected");
            return;
        }

        let mut request = request.unwrap();

        println!("Incoming method call: {}", request.endpoint());
        match request.take_body().unwrap() {
            Body::Message(bson) => {
                println!("Incoming message: {bson:?}");
            },
            Body::Call(bson) => {
                println!("Incoming call: {bson:?}");
                request.respond(Ok(bson)).await;
            },
            Body::Subscription => {
                println!("Incoming subscription");
                request.respond(Ok(41)).await;
                request.respond(Ok(42)).await;
                request.respond(Ok(43)).await;
            },
            Body::Fd { client_name, .. } => {
                println!("Incoming connection request from {client_name}");
                request.respond(Ok(())).await;
            }
        }
    }
}

有关更多示例，请参阅 tests/。