qrpc

qrpc is a small QUIC + mTLS messaging library where each QrpcInstance works as both server and client.

Features:

• bidirectional long-lived connections,
• peer registration by unique ID,
• direct send and broadcast,
• typestate builder (WithoutState / WithState),
• user-defined message trait (QrpcMessage) with no protobuf requirement.

Minimal Usage (Self-Contained, No Proto)

This is the smallest complete setup using:

• an empty AppState struct,
• a custom text message type,
• a single QrpcInstance.

use std::sync::Arc;

use qrpc::{Ctx, QrpcInstance, QrpcMessage, QrpcResult, State};

#[derive(Clone, Default)]
struct AppState {}

struct TextMessage(String);

impl QrpcMessage for TextMessage {
    fn cmd_id(&self) -> u32 { 1 }

    fn encode_vec(&self) -> Vec<u8> {
        self.0.as_bytes().to_vec()
    }

    fn decode_vec(cmd_id: u32, data: &[u8]) -> QrpcResult<Self> {
        if cmd_id != 1 {
            return Err(qrpc::QrpcError::MessageDecode("unexpected cmd_id".to_string()));
        }
        let s = String::from_utf8(data.to_vec())
            .map_err(|e| qrpc::QrpcError::MessageDecode(format!("utf8 decode failed: {e}")))?;
        Ok(Self(s))
    }
}

#[tokio::main]
async fn main() -> QrpcResult<()> {
    let instance = QrpcInstance::<AppState, TextMessage, _>::builder(
        |_state: State<AppState>, _ctx: Ctx<TextMessage>, source_peer_id: String, msg: TextMessage| async move {
            println!("from={}, text={}", source_peer_id, msg.0);
            Ok(())
        },
    )
    .with_state(AppState::default())
    .with_id("node-a")
    .with_ca_cert("tests/certs/ca.crt")
    .with_identity("tests/certs/server.crt", "tests/certs/server.key")
    .with_port(20001)
    .build()?;

    let instance = Arc::new(instance);
    let shutdown_instance = Arc::clone(&instance);
    tokio::spawn(async move {
        let _ = tokio::signal::ctrl_c().await;
        shutdown_instance.shutdown().await;
    });

    // Blocks until `shutdown()` is called.
    instance.serve().await;

    Ok(())
}

Lifecycle APIs

• start() starts background accept/connect loops and returns immediately.
• serve() calls start() and then blocks until shutdown() is called.
• serve_with(|ctx| async { ... }) runs one worker callback while serving.
  • If worker returns Ok(()), instance keeps serving until shutdown().
  • If worker returns Err(_), instance logs ERROR and keeps serving until shutdown().
• serve_with_rx(rx) serves while consuming OutboundCmd from an mpsc::Receiver.
  • SendTo waits for target peer connection before sending.
  • Broadcast sends to currently connected peers only (no peers => no-op).

Active publish pattern:

use std::{sync::Arc, time::Duration};

let instance = Arc::new(instance);
let worker_instance = Arc::clone(&instance);

instance
    .serve_with(move |_ctx| async move {
        let mut ticker = tokio::time::interval(Duration::from_secs(1));
        loop {
            ticker.tick().await;
            let _ = worker_instance.broadcast(&TextMessage("tick".into())).await;
        }
        #[allow(unreachable_code)]
        Ok(())
    })
    .await?;

TLS Identity Notes

• with_identity(cert, key) configures the local server certificate (used when peers dial this node).
• By default, outgoing dials reuse the same identity.
• If your certs split EKU (serverAuth vs clientAuth), also set:
  • with_client_identity(client_cert, client_key) for outgoing dials.

Example:

.with_identity("tests/certs/server.crt", "tests/certs/server.key")
.with_client_identity("tests/certs/client.crt", "tests/certs/client.key")

Connection Liveness

• Default keepalive interval: 10s.
• Default max idle timeout: 600s (10 minutes).
• Tune with builder methods:
  • with_keep_alive_interval(Some(Duration::from_secs(...)))
  • with_max_idle_timeout(Some(Duration::from_secs(...)))
  • pass None to disable keepalive or set infinite idle timeout.

Example:

use std::time::Duration;

.with_keep_alive_interval(Some(Duration::from_secs(5)))
.with_max_idle_timeout(Some(Duration::from_secs(1800)))

State Rules

• Use .with_state(S) to build an instance with any custom state type S.
• Callback signature uses State<T>. T is extracted from S via FromRef<S>.
• Callback also receives Ctx<M> plus source_peer_id: String.
• Ctx<M> supports send_to/broadcast/peer_ids/wait_for_peer/shutdown.
• By default, FromRef<T> for T is available when T: Clone.
• If you need shared ownership, pass Arc<T> as S (same pattern as axum state).
• If you do not call with_state, build() is available only for S = () and the builder injects () automatically.

Running Built-in Examples

cargo run --example two_nodes_ping_pong
cargo run --example three_nodes_mesh_direct
cargo run --example three_nodes_broadcast
cargo run --example two_node_20001
cargo run --example two_node_20002
cargo run --example two_node_serve_20001
cargo run --example two_node_ctx_reply_20001
cargo run --example two_node_ctx_reply_20002

LICENSE

See LICENSE