How to set up an RPC server on a tokio::DuplexStream?

[AArnott]

I have a tokio::io::DuplexStream and need to establish an RPC server on it. This is not a WebSocket or HTTP connection.

As a DuplexStream, there is no clearly defined message boundary, so the protocol for that is proprietary and has a couple of options we must support. The simplest is that a big endian 32-bit integer will prepend ever JSON-RPC message that indicates the bytes to decode into the JSON-RPC message.
Assuming I implement the part that reads from the duplex and creates a slice of the bytes that contain exactly one message, how can I then deserialize the message and get it to invoke the RPC server? So far, I figured out how to create an RpcModule<RpcTestTarget> for the server.

I'll also need to know how to serialize the response so I can send it back over the duplex.

Thanks!

[niklasad1]

Hey,

If I understand correctly you basically want to know how to decode/encode the bytes to/from jsonrpc method call and response. I can think of the following two options to do that:

1. If it's possible create RpcService which is very similar to a tower service but that requires you decode the JSON-RPC yourself such as:

    // I'm not sure whether everything is exported to instantiate it from another crate 
    let rpc_service = ...;
    let stream = ..;
    
    let bytes = stream.read(..).await.unwrap();
    let request: jsonrpsee:types::Request = serde_json::from_slice(bytes).unwrap();
    let rp = rpc_service.call(request).await.unwrap();
    stream.write(rp.into_result().as_bytes()).await.unwrap();

2. Use a RpcModule directly and there is a few helper functions for you such Methods::raw_json_request but you need to decode the responses manually then. Or you decode a jsonrpsee::types::Request from your bytes do something similar to https://github.com/paritytech/jsonrpsee/blob/master/server/src/middleware/rpc/layer/rpc_service.rs#L86-#L148

I think 1) doesn't work because the constructor is not exported but we can fix that if you think that it will work.

[AArnott]

Thanks. That sounds like a start. We may be reading all 4 message types (request, notification, result, error) from the stream. As we cannot predict what it will be before we deserialize it, and deserialization requires we know what type it will be, how do we handle that?

I noticed from another transport/example that is in this repo that you have a sort of try-this-else-try-that form of deserialization. That could work for us. But I've noticed some of your message types take generic type arguments and I'm not sure what to provide there.

Our scenario is a p2p one, where our endpoint may send and receive requests. If we deserialize a response, what API do we send it to so that a prior Future for a pending request can be resolved?

[AArnott]

I figured this out. Here is a client that works off of DuplexStream:

use async_trait::async_trait;
use jsonrpsee::{
    async_client::{Client, ClientBuilder},
    core::client::{ReceivedMessage, TransportReceiverT, TransportSenderT},
    proc_macros::rpc,
};
use tokio::io::{
    duplex, split, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, DuplexStream, ReadHalf,
    WriteHalf,
};

#[tokio::main]
async fn main() {
    let (client_stream, server_stream) = duplex(1024);

    tokio::spawn(mock_server(server_stream));

    let client = build_client(client_stream);
    let sum = client.add(3, 5).await.unwrap();
    println!("{}", sum);
}

async fn mock_server(mut pipe: DuplexStream) -> Result<(), std::io::Error> {
    let mut buffer = vec![0u8; 1024];
    let read = pipe.read(&mut buffer).await?;
    println!(
        "Received JSON request:\n{}",
        String::from_utf8_lossy(&buffer[..read])
    );

    // Send our mock response
    let response = r#"
    {
        "jsonrpc": "2.0",
        "id": 0,
        "result": 8
    }
    "#;
    pipe.write_all(response.as_bytes()).await?;
    pipe.flush().await?;

    Ok(())
}

fn duplex_transport<T: AsyncRead + AsyncWrite>(
    duplex: T,
) -> (
    AsyncReadTransport<ReadHalf<T>>,
    AsyncWriteTransport<WriteHalf<T>>,
) {
    let (reader, writer) = split(duplex);
    let reader = AsyncReadTransport { reader };
    let writer = AsyncWriteTransport { writer };
    (reader, writer)
}

fn build_client<T: AsyncRead + AsyncWrite + Send + 'static>(duplex: T) -> Client {
    let (r, w) = duplex_transport(duplex);
    ClientBuilder::default().build_with_tokio(w, r)
}

#[rpc(client)]
trait Calculator {
    #[method(name = "add")]
    fn add(&self, a: u64, b: u64) -> Result<u64, ErrorObject>;
}

struct AsyncWriteTransport<T: AsyncWrite> {
    writer: T,
}

#[async_trait]
impl<T: AsyncWrite + Send + 'static + std::marker::Unpin> TransportSenderT
    for AsyncWriteTransport<T>
{
    type Error = std::io::Error;

    async fn send(&mut self, msg: String) -> Result<(), Self::Error> {
        self.writer.write_all(msg.as_bytes()).await?;
        Ok(())
    }
}

struct AsyncReadTransport<T: AsyncRead> {
    reader: T,
}

#[async_trait]
impl<T: AsyncRead + Send + 'static + std::marker::Unpin> TransportReceiverT
    for AsyncReadTransport<T>
{
    type Error = std::io::Error;

    async fn receive(&mut self) -> Result<ReceivedMessage, Self::Error> {
        let mut buffer = vec![0u8; 1024];
        let read = self.reader.read(&mut buffer[..]).await?;
        Ok(ReceivedMessage::Bytes(buffer[..read].to_vec()))
    }
}

And here is a server that does the same:

use jsonrpsee::proc_macros::rpc;

#[tokio::main]
async fn main() {
    let request = r#"
    {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "add",
        "params": { "a": 3, "b": 5 }
    }
    "#;


    let server = CalculatorService {}.into_rpc();
    let response = server.raw_json_request(request, 1).await.unwrap();
    println!("{}", response.0);
}

#[rpc(server)]
trait Calculator {
    #[method(name = "add")]
    fn add(&self, a: u64, b: u64) -> u64;
}

struct CalculatorService;

impl CalculatorServer for CalculatorService {
    fn add(&self, a: u64, b: u64) -> u64 {
        a + b
    }
}

[AArnott]

But what this does not solve is how would P2P work? The ClientBuilder seems to be oriented only around being a client. But what if the transport fed into it receives requests?