JetStream

JetStream is an RPC framework designed to be a high performance, low latency, secure, and reliable RPC framework.

Transport Backends

JetStream supports multiple transport backends:

quinn - QUIC transport with TLS/mTLS support
iroh - P2P transport with built-in NAT traversal
webtransport - WebTransport transport for browser and server environments

Features

Bidirectional streaming
0-RTT
mTLS
Binary encoding
Cross-platform (Linux, macOS, Windows, WebAssembly)
Cross-language — TypeScript and Swift clients with wire-compatible codegen

For detailed API documentation, see the rustdoc documentation.

Examples

echo - Basic QUIC-based echo service example
iroh_echo - Echo service using iroh transport
wasm_example - WebAssembly example
wasm_example_bindings - WebAssembly bindings example

JetStream Iroh

Iroh is a peer-to-peer networking library that provides NAT traversal, hole punching, and relay fallback. JetStream integrates with Iroh to enable peer-to-peer RPC communication.

Features

NAT Traversal: Automatic hole punching for direct peer-to-peer connections
Relay Fallback: Falls back to relay servers when direct connection isn’t possible
Discovery: Uses JetStream’s discovery service at discovery.jetstream.rs
ALPN-based Protocol Negotiation: Each service defines its own protocol version

Example

Here’s a complete example of an echo service using Iroh transport:

#![cfg(feature = "iroh")]
use crate::square_protocol::{SquareChannel, SquareService};
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use jetstream::prelude::*;
use jetstream_macros::service;

#[service(tracing)]
pub trait Square {
    async fn square(&self, ctx: Context, i: u32) -> Result<String>;
}

#[derive(Debug, Clone)]
struct SquareServer;

impl Square for SquareServer {
    async fn square(&self, _ctx: Context, i: u32) -> Result<String> {
        Ok((i * i).to_string())
    }
}

#[tokio::main]
async fn main() {
    // Initialize tracing subscriber
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::DEBUG)
        .with_thread_ids(true)
        .with_span_events(
            tracing_subscriber::fmt::format::FmtSpan::ENTER
                | tracing_subscriber::fmt::format::FmtSpan::EXIT,
        )
        .init();
    // Build the server router with the echo service
    let router = jetstream_iroh::server_builder(SquareService {
        inner: SquareServer {},
    })
    .await
    .unwrap();

    // get our own address. At this point we have a running router
    // that's ready to accept connections.
    let addr = router.endpoint().addr();

    // Build client transport and connect
    let transport = jetstream_iroh::client_builder::<SquareChannel>(addr)
        .await
        .unwrap();

    let ec = SquareChannel::new(10, Box::new(transport));
    let mut futures = FuturesUnordered::new();
    for i in 0..1000 {
        futures.push(ec.square(Context::default(), i));
    }

    while let Some(result) = futures.next().await {
        let response = result.unwrap();
        println!("Response: {}", response);
    }
    router.shutdown().await.unwrap();
}

Server Setup

To create an Iroh server, use the server_builder function:

#![allow(unused)]
fn main() {
use jetstream_iroh::server_builder;

let router = server_builder(EchoService { inner: EchoServer {} })
    .await
    .unwrap();

// Get the node address to share with clients
let addr = router.endpoint().node_addr();
}

Client Setup

To connect to an Iroh server, use the client_builder function:

#![allow(unused)]
fn main() {
use jetstream_iroh::client_builder;

let mut transport = client_builder::<EchoChannel>(addr)
    .await
    .unwrap();
}

Discovery

JetStream uses a custom discovery service for Iroh nodes. The discovery URL is:

https://discovery.jetstream.rs

This allows nodes to find each other using their public keys without needing to exchange IP addresses directly.

Feature Flag

To use Iroh transport, enable the iroh feature in your Cargo.toml:

[dependencies]
jetstream = { version = "10", features = ["iroh"] }

For more details, see the jetstream_iroh API documentation.

JetStream QUIC

QUIC is a modern transport protocol that provides multiplexed connections over UDP with built-in TLS 1.3 encryption. JetStream QUIC provides the transport layer using quinn.

Features

ALPN-based Routing: Route connections to different handlers based on ALPN protocol negotiation
TLS 1.3: Built-in secure transport with rustls
0-RTT: Support for zero round-trip connection resumption
mTLS Support: Mutual TLS authentication for client certificate verification
Peer Identity: Extract client certificate information (CN, fingerprint, SANs) in request handlers

Architecture

The crate is organized around these core components:

Server: The main QUIC server that accepts incoming connections
Router: Routes connections to protocol handlers based on ALPN
ProtocolHandler: Trait for implementing custom protocol handlers
Client: QUIC client for connecting to servers

For HTTP/3 support, see jetstream_http.

Example

Here’s a complete echo service example:

use std::{net::SocketAddr, path::Path, sync::Arc};

use echo_protocol::EchoChannel;
use jetstream::prelude::*;
use jetstream_macros::service;
use jetstream_quic::{
    Client, QuicRouter, QuicRouterHandler, QuicTransport, Server,
};

use rustls::pki_types::{CertificateDer, PrivateKeyDer};

#[service]
pub trait Echo {
    async fn ping(&mut self) -> Result<()>;
}

#[derive(Clone)]
struct EchoImpl {}

impl Echo for EchoImpl {
    async fn ping(&mut self) -> Result<()> {
        eprintln!("Ping received");
        eprintln!("Pong sent");
        Ok(())
    }
}

pub static CA_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/ca.pem");
pub static CLIENT_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/client.pem");
pub static CLIENT_KEY_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/client.key");
pub static SERVER_CERT_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.pem");
pub static SERVER_KEY_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.key");

fn load_certs(path: &str) -> Vec<CertificateDer<'static>> {
    let data = std::fs::read(Path::new(path)).expect("Failed to read cert");
    rustls_pemfile::certs(&mut &*data)
        .filter_map(|r| r.ok())
        .collect()
}

fn load_key(path: &str) -> PrivateKeyDer<'static> {
    let data = std::fs::read(Path::new(path)).expect("Failed to read key");
    rustls_pemfile::private_key(&mut &*data)
        .expect("Failed to parse key")
        .expect("No key found")
}

async fn server(
    addr: SocketAddr,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let server_cert = load_certs(SERVER_CERT_PEM).pop().unwrap();
    let server_key = load_key(SERVER_KEY_PEM);
    let ca_cert = load_certs(CA_CERT_PEM).pop().unwrap();

    let mut root_store = rustls::RootCertStore::empty();
    root_store.add(ca_cert).expect("Failed to add CA cert");
    let client_verifier =
        rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
            .allow_unauthenticated()
            .build()
            .expect("Failed to build client verifier");

    // Register the EchoService via the per-stream RPC router
    let echo_service = echo_protocol::EchoService { inner: EchoImpl {} };

    let rpc_router = Arc::new(
        jetstream_rpc::Router::new()
            .with_handler(echo_protocol::PROTOCOL_NAME, echo_service),
    );
    let quic_handler = QuicRouterHandler::new(rpc_router);

    let mut router = QuicRouter::new();
    router.register(Arc::new(quic_handler));

    let server = Server::new_with_mtls(
        vec![server_cert],
        server_key,
        client_verifier,
        addr,
        router,
    );

    eprintln!("Server listening on {}", addr);
    server.run().await;

    Ok(())
}

async fn client(
    addr: SocketAddr,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
    // Wait for server to start
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

    let ca_cert = load_certs(CA_CERT_PEM).pop().unwrap();
    let client_cert = load_certs(CLIENT_CERT_PEM).pop().unwrap();
    let client_key = load_key(CLIENT_KEY_PEM);

    let alpn = vec![b"jetstream".to_vec()];
    let bind_addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
    let client = Client::new_with_mtls(
        ca_cert,
        client_cert,
        client_key,
        alpn,
        bind_addr,
    )?;

    let connection = client.connect(addr, "localhost").await?;

    // Open a bidirectional stream and wrap it in QuicTransport
    let (send, recv) = connection.open_bi().await?;
    let transport: QuicTransport<EchoChannel> = (send, recv).into();
    let mut chan = EchoChannel::new(10, Box::new(transport));
    chan.negotiate_version(u32::MAX).await?;

    eprintln!("Ping sent");
    chan.ping().await?;
    eprintln!("Pong received");

    Ok(())
}

#[tokio::main]
async fn main() {
    // Install the ring crypto provider for rustls
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    let addr: SocketAddr = "127.0.0.1:4433".parse().unwrap();
    tokio::select! {
      _ = server(addr) => {},
      _ = client(addr) => {},
    }
}

Defining a Service

Use the #[service] macro to define an RPC service:

#![allow(unused)]
fn main() {
use jetstream::prelude::*;
use jetstream_macros::service;

#[service]
pub trait Echo {
    async fn ping(&mut self) -> Result<()>;
    async fn echo(&mut self, message: String) -> Result<String>;
}
}

The macro generates:

EchoChannel - Client-side channel for calling methods
EchoService - Server-side wrapper for your implementation

Implementing the Service

#![allow(unused)]
fn main() {
#[derive(Clone)]
struct EchoImpl;

impl Echo for EchoImpl {
    async fn ping(&mut self) -> Result<()> {
        Ok(())
    }

    async fn echo(&mut self, message: String) -> Result<String> {
        Ok(message)
    }
}
}

Server Setup

#![allow(unused)]
fn main() {
use std::sync::Arc;
use jetstream_quic::{Server, Router};

// Create the service
let echo_service = echo_protocol::EchoService { inner: EchoImpl {} };

// Register with router
let mut router = Router::new();
router.register(Arc::new(echo_service));

// Create and run server
let server = Server::new_with_addr(cert, key, addr, router);
server.run().await;
}

Client Setup

#![allow(unused)]
fn main() {
use jetstream_quic::{Client, QuicTransport};
use jetstream_rpc::Protocol;

// Create client
let alpn = vec![EchoChannel::VERSION.as_bytes().to_vec()];
let client = Client::new_with_mtls(ca_cert, client_cert, client_key, alpn)?;

// Connect
let connection = client.connect(addr, "localhost").await?;

// Open stream and create channel
let (send, recv) = connection.open_bi().await?;
let transport: QuicTransport<EchoChannel> = (send, recv).into();
let mut chan = EchoChannel::new(10, Box::new(transport));

// Call methods
chan.ping().await?;
let response = chan.echo("Hello".to_string()).await?;
}

TLS Certificates

Generate development certificates:

cd certs
./generate_certs.sh

This generates:

ca.pem / ca.key - Certificate Authority
server.pem / server.key - Server certificate
client.pem / client.key - Client certificate
client.p12 - PKCS12 bundle for browser import

For production, use certificates from a trusted CA or Let’s Encrypt.

Mutual TLS (mTLS)

JetStream QUIC supports mutual TLS authentication:

#![allow(unused)]
fn main() {
// Build a client certificate verifier from a CA cert
let mut root_store = rustls::RootCertStore::empty();
root_store.add(ca_cert).expect("Failed to add CA cert");
let client_verifier =
    rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
        .allow_unauthenticated()
        .build()
        .expect("Failed to build client verifier");

let server = Server::new_with_mtls(
    server_cert,
    server_key,
    client_verifier,  // any Arc<dyn ClientCertVerifier>
    addr,
    router,
);
}

Accessing Peer Certificate Info

#![allow(unused)]
fn main() {
use jetstream_rpc::context::{Context, Peer};

// In your service implementation or handler
if let Some(Peer::Tls(tls_peer)) = ctx.peer() {
    if let Some(leaf) = tls_peer.leaf() {
        println!("Client CN: {:?}", leaf.common_name);
        println!("Fingerprint: {}", leaf.fingerprint);
        println!("DNS SANs: {:?}", leaf.dns_names);
    }
}
}

Dependencies

Add to your Cargo.toml:

[dependencies]
jetstream = "13"
jetstream_quic = "13"
jetstream_macros = "13"
tokio = { version = "1", features = ["full"] }

For more details, see the jetstream_quic API documentation.

JetStream HTTP

JetStream HTTP provides HTTP/2 and HTTP/3 server support with a unified Axum router interface. It enables serving both protocols on the same port using TCP for HTTP/2 and UDP (QUIC) for HTTP/3.

Features

HTTP/2 + HTTP/3: Serve both protocols simultaneously on the same port
Axum Integration: Use familiar Axum routers and handlers
Alt-Svc Header: Automatically advertise HTTP/3 availability to HTTP/2 clients
Context Extractor: Access connection metadata (peer certificates, remote address) in handlers
mTLS Support: Mutual TLS authentication for both protocols

Architecture

The crate provides these components:

H3Service: HTTP/3 protocol handler for use with jetstream_quic
AltSvcLayer: Tower layer that adds Alt-Svc header to advertise HTTP/3
JetStreamContext: Axum extractor for accessing connection context

Example

Here’s a complete example serving HTTP/2 and HTTP/3:

use std::fs;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;

use axum::Router;
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as HttpBuilder;
use hyper_util::service::TowerToHyperService;
use jetstream::prelude::*;
use jetstream_http::{AltSvcLayer, H3Service};
use jetstream_macros::service;
use tokio::net::TcpListener;
use tokio_rustls::TlsAcceptor;
use tower_http::services::ServeDir;
use tracing::{error, info};

use rustls::pki_types::{CertificateDer, PrivateKeyDer};

// r[impl jetstream.webtransport.http-example]
#[service]
pub trait EchoHttp {
    async fn ping(&mut self, message: String) -> Result<String>;
    async fn add(&mut self, a: i32, b: i32) -> Result<i32>;
}

#[derive(Clone)]
struct EchoHttpImpl;

impl EchoHttp for EchoHttpImpl {
    async fn ping(&mut self, message: String) -> Result<String> {
        Ok(message)
    }
    async fn add(&mut self, a: i32, b: i32) -> Result<i32> {
        Ok(a + b)
    }
}

pub static CA_PEM: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/certs/ca.pem");
pub static SERVER_PEM: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.pem");
pub static SERVER_KEY: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/certs/server.key");

fn load_certs(path: &str) -> Vec<CertificateDer<'static>> {
    let data = fs::read(Path::new(path)).expect("Failed to read cert");
    rustls_pemfile::certs(&mut &*data)
        .filter_map(|r| r.ok())
        .collect()
}

fn load_key(path: &str) -> PrivateKeyDer<'static> {
    let data = fs::read(Path::new(path)).expect("Failed to read key");
    rustls_pemfile::private_key(&mut &*data)
        .expect("Failed to parse key")
        .expect("No key found")
}

pub static APP_DIST: &str =
    concat!(env!("CARGO_MANIFEST_DIR"), "/examples/app/dist");

/// Run HTTP/2 server with TLS
async fn run_http2_server(
    addr: SocketAddr,
    router: Router,
    tls_acceptor: TlsAcceptor,
) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let listener = TcpListener::bind(addr).await?;

    loop {
        let (stream, _remote_addr) = listener.accept().await?;
        let tls_acceptor = tls_acceptor.clone();
        let router = router.clone();

        tokio::spawn(async move {
            let tls_stream = match tls_acceptor.accept(stream).await {
                Ok(s) => s,
                Err(e) => {
                    error!("TLS accept error: {}", e);
                    return;
                }
            };

            let io = TokioIo::new(tls_stream);
            let svc = TowerToHyperService::new(router);

            if let Err(e) = HttpBuilder::new(TokioExecutor::new())
                .serve_connection(io, svc)
                .await
            {
                error!("HTTP/2 connection error: {}", e);
            }
        });
    }
}

/// Run HTTP/3 server with QUIC + WebTransport
async fn run_http3_server(
    addr: SocketAddr,
    router: Router,
    server_cert: CertificateDer<'static>,
    server_key: PrivateKeyDer<'static>,
    ca_cert: Option<CertificateDer<'static>>,
) {
    // Register the EchoHttp service as a WebTransport handler
    let echo = echohttp_protocol::EchoHttpService {
        inner: EchoHttpImpl,
    };
    let rpc_router = Arc::new(
        jetstream_rpc::Router::new()
            .with_handler(echohttp_protocol::PROTOCOL_NAME, echo),
    );

    let mut quic_router = jetstream_quic::QuicRouter::new();

    let server = if let Some(ca) = ca_cert {
        let mut root_store = rustls::RootCertStore::empty();
        root_store.add(ca).expect("Failed to add CA cert");
        let client_verifier =
            rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
                .allow_unauthenticated()
                .build()
                .expect("Failed to build client verifier");

        let h3_service = Arc::new(H3Service::new_with_cert_verifier(
            router,
            rpc_router,
            client_verifier.clone(),
        ));
        quic_router.register(h3_service);

        jetstream_quic::Server::new_with_mtls(
            vec![server_cert],
            server_key,
            client_verifier,
            addr,
            quic_router,
        )
    } else {
        let h3_service = Arc::new(H3Service::new(router, rpc_router));
        quic_router.register(h3_service);

        jetstream_quic::Server::new_with_addr(
            vec![server_cert],
            server_key,
            addr,
            quic_router,
        )
    };

    server.run().await;
}

#[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing
    tracing_subscriber::fmt::init();

    // Install the ring crypto provider for rustls
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    // Check for --mtls flag
    let mtls_enabled = std::env::args().any(|arg| arg == "--mtls");

    let server_certs = load_certs(SERVER_PEM);
    let server_key = load_key(SERVER_KEY);
    let ca_cert = if mtls_enabled {
        Some(load_certs(CA_PEM).pop().unwrap())
    } else {
        None
    };

    // Create shared Axum router serving React app build + Alt-Svc header
    let router = Router::new()
        .fallback_service(ServeDir::new(APP_DIST))
        .layer(AltSvcLayer::new(4433));

    // Setup TLS config for HTTP/2
    let mut tls_config = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(server_certs.clone(), server_key.clone_key())?;
    tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
    let tls_acceptor = TlsAcceptor::from(Arc::new(tls_config));

    let addr: SocketAddr = "127.0.0.1:4433".parse()?;

    info!("=== JetStream HTTP + WebTransport Server ===");
    info!("Listening on https://{}", addr);
    info!("  - HTTP/2 over TLS (TCP)");
    info!("  - HTTP/3 + WebTransport over QUIC (UDP)");
    info!(
        "  - WebTransport protocol: {}",
        echohttp_protocol::PROTOCOL_VERSION
    );
    if mtls_enabled {
        info!("mTLS enabled - client certificates required");
    } else {
        info!("No client auth (use --mtls to enable)");
    }

    // Run both servers concurrently on the same port (TCP for HTTP/2, UDP for HTTP/3)
    tokio::select! {
        result = run_http2_server(addr, router.clone(), tls_acceptor) => {
            if let Err(e) = result {
                error!("HTTP/2 server error: {}", e);
            }
        }
        _ = run_http3_server(
            addr,
            router,
            server_certs.into_iter().next().unwrap(),
            server_key,
            ca_cert,
        ) => {}
    }

    Ok(())
}

Quick Start

1. Create an Axum Router

#![allow(unused)]
fn main() {
use axum::{routing::get, Router};
use jetstream_http::{AltSvcLayer, JetStreamContext};

async fn handler(ctx: JetStreamContext) -> &'static str {
    println!("Request from: {:?}", ctx.remote());
    "Hello, World!"
}

let router = Router::new()
    .route("/", get(handler))
    .layer(AltSvcLayer::new(4433));
}

2. Set Up HTTP/2 Server (TCP + TLS)

#![allow(unused)]
fn main() {
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as HttpBuilder;
use hyper_util::service::TowerToHyperService;
use tokio::net::TcpListener;
use tokio_rustls::TlsAcceptor;

let listener = TcpListener::bind("127.0.0.1:4433").await?;
let tls_acceptor = TlsAcceptor::from(Arc::new(tls_config));

loop {
    let (stream, _) = listener.accept().await?;
    let tls_stream = tls_acceptor.accept(stream).await?;
    let io = TokioIo::new(tls_stream);
    let svc = TowerToHyperService::new(router.clone());
    
    HttpBuilder::new(TokioExecutor::new())
        .serve_connection(io, svc)
        .await?;
}
}

3. Set Up HTTP/3 Server (QUIC)

#![allow(unused)]
fn main() {
use jetstream_http::H3Service;
use jetstream_quic::{Router as QuicRouter, Server};

let h3_service = Arc::new(H3Service::new(router));

let mut quic_router = QuicRouter::new();
quic_router.register(h3_service);

let server = Server::new_with_addr(cert, key, addr, quic_router);
server.run().await;
}

Alt-Svc Layer

The AltSvcLayer adds the Alt-Svc header to HTTP/2 responses, telling browsers that HTTP/3 is available:

#![allow(unused)]
fn main() {
use jetstream_http::AltSvcLayer;

// Advertise HTTP/3 on port 4433 with 24-hour max-age
let layer = AltSvcLayer::new(4433);
// Adds: Alt-Svc: h3=":4433"; ma=86400

// Or with custom value
let layer = AltSvcLayer::with_value("h3=\":443\"; ma=3600, h3-29=\":443\"");
}

When browsers receive this header over HTTP/2, they will attempt to upgrade to HTTP/3 on subsequent requests.

Context Extractor

Use JetStreamContext to access connection metadata in your handlers:

#![allow(unused)]
fn main() {
use jetstream_http::JetStreamContext;
use jetstream_rpc::context::{Peer, RemoteAddr};

async fn handler(ctx: JetStreamContext) -> String {
    // Get remote address
    let addr = match ctx.remote() {
        Some(RemoteAddr::IpAddr(ip)) => ip.to_string(),
        _ => "unknown".to_string(),
    };

    // Get peer certificate info (for mTLS)
    if let Some(Peer::Tls(tls_peer)) = ctx.peer() {
        if let Some(leaf) = tls_peer.leaf() {
            return format!(
                "Hello {}! (CN: {:?})",
                addr,
                leaf.common_name
            );
        }
    }

    format!("Hello {}!", addr)
}
}

TLS Configuration

HTTP/2 TLS Config

#![allow(unused)]
fn main() {
let mut tls_config = rustls::ServerConfig::builder()
    .with_no_client_auth()
    .with_single_cert(certs, key)?;

// Important: set ALPN for HTTP/2
tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
}

HTTP/3 (QUIC) TLS

QUIC/HTTP/3 TLS is handled automatically by jetstream_quic::Server.

Certificates

Generate development certificates:

cd certs
./generate_certs.sh

Files created:

server.pem / server.key - Server certificate
ca.pem - CA certificate (for client trust)
client.pem / client.key - Client certificate (for mTLS)
client.p12 - PKCS12 bundle for browser import
server.crt - DER format for Chrome

Testing with Chrome

Chrome supports HTTP/3. To test:

# Start the server
cargo run --example http --features quic

# Launch Chrome with QUIC enabled
./examples/launch_chrome.sh

The launch_chrome.sh script:

Extracts the server’s SPKI hash
Launches Chrome with --origin-to-force-quic-on flag
Provides instructions for importing certificates

Dependencies

Add to your Cargo.toml:

[dependencies]
jetstream_http = "0.1"
jetstream_quic = "13"
axum = "0.8"
hyper-util = { version = "0.1", features = ["full"] }
tokio-rustls = "0.26"
tower = "0.5"

For more details, see the jetstream_http API documentation.

TypeScript

JetStream provides TypeScript packages for wire format encoding and RPC communication, published to the GitHub Package Registry under @sevki.

Installation

# Configure npm to use GitHub Package Registry for @sevki scope
echo "@sevki:registry=https://npm.pkg.github.com" >> .npmrc

pnpm add @sevki/jetstream-wireformat @sevki/jetstream-rpc

WireFormat Codecs

The @sevki/jetstream-wireformat package provides binary codecs for all JetStream primitive and composite types. Every codec implements the WireFormat<T> interface:

interface WireFormat<T> {
  byteSize(value: T): number;
  encode(value: T, writer: BinaryWriter): void;
  decode(reader: BinaryReader): T;
}

Primitives

import { BinaryReader, BinaryWriter, u32Codec, stringCodec, boolCodec } from '@sevki/jetstream-wireformat';

// Encode a u32
const writer = new BinaryWriter();
u32Codec.encode(42, writer);
const bytes = writer.toUint8Array();

// Decode it back
const reader = new BinaryReader(bytes);
const value = u32Codec.decode(reader); // 42

Available primitive codecs: u8Codec, u16Codec, u32Codec, u64Codec, i16Codec, i32Codec, i64Codec, f32Codec, f64Codec, boolCodec, stringCodec.

Composite Types

Build codecs for collections and optional types:

import { vecCodec, optionCodec, mapCodec, stringCodec, u32Codec } from '@sevki/jetstream-wireformat';

// Vec<String>
const tagsCodec = vecCodec(stringCodec);

// Option<u32>
const maybeIdCodec = optionCodec(u32Codec);

// HashMap<String, u32>
const scoresCodec = mapCodec(stringCodec, u32Codec);

Structs and Enums

Use the structCodec and enumCodec helpers, or generate codecs from Rust types using jetstream_codegen:

import { BinaryReader, BinaryWriter, u32Codec } from '@sevki/jetstream-wireformat';
import type { WireFormat } from '@sevki/jetstream-wireformat';

// A manually defined codec for a Point struct
interface Point {
  x: number;
  y: number;
}

const pointCodec: WireFormat<Point> = {
  byteSize(value: Point): number {
    return u32Codec.byteSize(value.x) + u32Codec.byteSize(value.y);
  },
  encode(value: Point, writer: BinaryWriter): void {
    u32Codec.encode(value.x, writer);
    u32Codec.encode(value.y, writer);
  },
  decode(reader: BinaryReader): Point {
    const x = u32Codec.decode(reader);
    const y = u32Codec.decode(reader);
    return { x, y };
  },
};

Code Generation

Instead of writing codecs by hand, use jetstream_codegen to generate TypeScript types and codecs from Rust source files:

cargo run -p jetstream_codegen -- \
  --input src/types.rs \
  --ts-out generated/

Given a Rust file:

#![allow(unused)]
fn main() {
#[derive(JetStreamWireFormat)]
pub struct Point {
    pub x: u32,
    pub y: u32,
}

#[derive(JetStreamWireFormat)]
pub enum Shape {
    Circle(u32),
    Rectangle { width: u32, height: u32 },
}
}

The codegen produces TypeScript interfaces and WireFormat<T> codec objects that are wire-compatible with the Rust implementations.

RPC

The @sevki/jetstream-rpc package provides the RPC runtime for multiplexed request/response communication.

Generated Code

The codegen generates RPC client and handler types from #[service] trait definitions:

#![allow(unused)]
fn main() {
// Rust service definition
#[service]
pub trait EchoHttp {
    async fn ping(&mut self, message: String) -> Result<String>;
    async fn add(&mut self, a: i32, b: i32) -> Result<i32>;
}
}

cargo run -p jetstream_codegen -- \
  --input examples/http.rs \
  --ts-out generated/

This generates:

Request/response types: TPing, RPing, TAdd, RAdd with codecs
Frame unions: Tmessage, Rmessage discriminated unions with FramerCodec implementations
Framer wrappers: TmessageFramer, RmessageFramer classes implementing the Framer interface
rmessageDecode: A decoder function for use with Mux and WebTransportTransport
Protocol constants: PROTOCOL_NAME (e.g., 'rs.jetstream.proto/echohttp') and PROTOCOL_VERSION (e.g., 'rs.jetstream.proto/echohttp/15.0.0+bfd7d20e')
EchoHttpClient: A typed client class with async methods and version negotiation
EchoHttpHandler: A handler interface for implementing server-side dispatch
dispatchEchoHttp: A dispatch function that routes Tmessage frames to handler methods

Version Negotiation

Before making RPC calls, clients must perform a Tversion/Rversion handshake to negotiate the protocol version and maximum message size. The generated client provides a static negotiate method:

import { EchoHttpClient, rmessageDecode, PROTOCOL_NAME } from './generated/echohttp_rpc.js';

// Open a WebTransport session and bidi stream
const session = new WebTransport(`https://api.example.com:4433/${PROTOCOL_NAME}`);
await session.ready;
const stream = await session.createBidirectionalStream();

// Negotiate version on the raw stream before creating the Mux
const negotiated = await EchoHttpClient.negotiate(stream.readable, stream.writable);
console.log(`Negotiated: ${negotiated.version}, msize: ${negotiated.msize}`);

You can also call negotiateVersion directly from @sevki/jetstream-rpc:

import { negotiateVersion } from '@sevki/jetstream-rpc';
import { PROTOCOL_VERSION } from './generated/echohttp_rpc.js';

const negotiated = await negotiateVersion(stream.readable, stream.writable, PROTOCOL_VERSION);

After negotiation, the stream is ready for Mux framing.

Client Usage

import { Mux } from '@sevki/jetstream-rpc';
import { EchoHttpClient, rmessageDecode, PROTOCOL_NAME } from './generated/echohttp_rpc.js';

// 1. Connect
const session = new WebTransport(`https://api.example.com:4433/${PROTOCOL_NAME}`);
await session.ready;
const stream = await session.createBidirectionalStream();

// 2. Negotiate version
await EchoHttpClient.negotiate(stream.readable, stream.writable);

// 3. Create transport and mux
const transport = new WebTransportTransport(stream, rmessageDecode);
const mux = new Mux(transport);
await mux.start();

// 4. Create client and make RPC calls
const client = new EchoHttpClient(mux);
const reply = await client.ping("hello");  // "hello"
const sum = await client.add(2, 3);        // 5

// 5. Cleanup
await mux.close();
session.close();

Handler (Server-Side)

Implement the generated handler interface to serve RPCs:

import { EchoHttpHandler, dispatchEchoHttp } from './generated/echohttp_rpc.js';

const handler: EchoHttpHandler = {
  async ping(ctx, message) {
    return message; // echo back
  },
  async add(ctx, a, b) {
    return a + b;
  },
};

RPC frames follow the format [size:u32 LE][type:u8][tag:u16 LE][payload] where size includes itself (minimum 7 bytes). The tag field enables multiplexing concurrent requests over a single connection.

Special message types:

TVERSION (100) / RVERSION (101) — version negotiation
MESSAGE_ID_START (102) — first service method ID
RJETSTREAMERROR (5) — error response frames

Mux

The Mux class handles tag allocation, request/response matching, and concurrent RPC dispatch:

import { Mux } from '@sevki/jetstream-rpc';

const mux = new Mux(transport);
await mux.start();

// Each rpc() call acquires a tag, sends a frame, waits for the matching response, and releases the tag
const response = await mux.rpc(request);

await mux.close();

Protocol Interface

Every generated service implements the Protocol interface:

interface Protocol<TReq extends Framer, TRes extends Framer> {
  readonly VERSION: string;
  readonly NAME: string;
}

NAME is the protocol name used for routing (e.g., URI path, ALPN)
VERSION is the full version string used during Tversion/Rversion negotiation

React

@sevki/jetstream-react provides React hooks for bidirectional RPC over WebTransport. Downstream (browser) can call upstream services, and upstream can push RPCs downstream.

Installation

pnpm add @sevki/jetstream-react @sevki/jetstream-rpc @sevki/jetstream-wireformat

react (>=18) is a peer dependency.

Quick Start

import { JetStreamProvider, useJetStream, useJetStreamStatus, useRPC } from '@sevki/jetstream-react';
import { EchoHttpClient, rmessageDecode, PROTOCOL_VERSION, PROTOCOL_NAME } from './generated/echohttp_rpc.js';

function App() {
  return (
    <JetStreamProvider url={`https://api.example.com:4433/${PROTOCOL_NAME}`}>
      <EchoDemo />
    </JetStreamProvider>
  );
}

function EchoDemo() {
  const status = useJetStreamStatus();
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);

  const { data, error, isLoading } = useRPC(
    () => (echo ? echo.ping('hello') : Promise.resolve('')),
    [echo],
  );

  return (
    <div>
      <p>Connection: {status}</p>
      {isLoading && <p>Loading...</p>}
      {error && <p>Error: {error.message}</p>}
      {data && <p>Echo: {data}</p>}
    </div>
  );
}

Provider

JetStreamProvider manages the WebTransport session lifecycle. It establishes the session on mount and closes it on unmount.

<JetStreamProvider url="https://api.example.com:4433/rs.jetstream.proto/echohttp">
  <App />
</JetStreamProvider>

Props:

url (required) — the upstream WebTransport URL, typically https://host:port/{PROTOCOL_NAME}
children — React children

The provider exposes connection state and protocol version through context:

session — the WebTransport instance (or null before connected)
state — 'connecting' | 'connected' | 'disconnected' | 'error'
protocolVersion — the negotiated protocol version string (or null before negotiation)

useJetStreamStatus

Read the current connection state:

function StatusIndicator() {
  const status = useJetStreamStatus();
  return <span className={status}>{status}</span>;
}

useJetStream

Creates a memoized RPC client for calling upstream services. Handles stream creation, version negotiation, and Mux setup automatically.

const client = useJetStream(ClientClass, responseDecode, protocolVersion);

Parameters:

ClientClass — the generated client constructor (e.g., EchoHttpClient)
responseDecode — the generated response decoder function (e.g., rmessageDecode)
protocolVersion — the protocol version string for Tversion negotiation (e.g., PROTOCOL_VERSION)

Returns: the client instance (or null while connecting).

The hook:

Opens a bidirectional stream on the WebTransport session
Performs Tversion/Rversion negotiation on the raw stream
Stores the negotiated version in the provider context
Creates a WebTransportTransport and Mux over the stream
Constructs and returns the client

The returned client is stable across re-renders as long as the session and constructor remain the same.

import { EchoHttpClient, rmessageDecode, PROTOCOL_VERSION } from './generated/echohttp_rpc.js';

function MyComponent() {
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);

  if (!echo) return <p>Connecting...</p>;

  return <button onClick={() => echo.ping('hi')}>Ping</button>;
}

useRPC

Reactive wrapper around a single RPC call. Re-executes when dependencies change.

const { data, error, isLoading, refetch } = useRPC(fn, deps);

Parameters:

fn — a function returning a Promise<T> (the RPC call)
deps — dependency array (like useEffect); the call re-runs when dependencies change

Returns:

data: T | undefined — the resolved value
error: Error | undefined — the rejection reason
isLoading: boolean — whether a request is in flight
refetch: () => void — manually re-execute the call

When dependencies change, the previous in-flight request is discarded (stale results are never applied).

function SearchResults() {
  const [query, setQuery] = useState('');
  const search = useJetStream(SearchClient, rmessageDecode, PROTOCOL_VERSION);

  const { data, error, isLoading } = useRPC(
    () => (search ? search.find(query) : Promise.resolve([])),
    [search, query],
  );

  return (
    <div>
      <input value={query} onChange={e => setQuery(e.target.value)} />
      {isLoading && <p>Loading...</p>}
      {error && <p>Error: {error.message}</p>}
      {data?.map(item => <div key={item.id}>{item.name}</div>)}
    </div>
  );
}

useHandler

Registers a handler for incoming upstream-initiated RPCs (push notifications, cache invalidation, etc.).

import { NotificationHandler } from './generated/notification_rpc.js';

function NotificationBadge() {
  const { events, error } = useHandler(NotificationHandler, {
    async notify(ctx, title, body) {
      return { ack: true };
    },
  });

  const unread = events.filter(e => e.method === 'notify').length;
  return <span>{unread > 0 && `${unread} new`}</span>;
}

The handler is registered on mount and unregistered on unmount. Incoming bidirectional streams from upstream are dispatched to the matching handler.

Imperative Calls and Mutations

For imperative operations (mutations, fire-and-forget), call the client directly. The generated client methods return plain Promises:

function AddButton() {
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);
  const [sum, setSum] = useState<number>();

  return (
    <button
      disabled={!echo}
      onClick={async () => {
        if (echo) setSum(await echo.add(2, 3));
      }}
    >
      Add 2 + 3 {sum !== undefined && `= ${sum}`}
    </button>
  );
}

For richer mutation state (loading, error, retries, cache invalidation), use TanStack Query:

import { useMutation } from '@tanstack/react-query';

function AddForm() {
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);

  const { mutate, data, isPending } = useMutation({
    mutationFn: ({ a, b }: { a: number; b: number }) => echo!.add(a, b),
  });

  return (
    <div>
      <button onClick={() => mutate({ a: 2, b: 3 })} disabled={isPending || !echo}>
        Add 2 + 3
      </button>
      {data !== undefined && <p>Sum: {data}</p>}
    </div>
  );
}

Full Example

import { useState } from 'react';
import {
  JetStreamProvider,
  useJetStream,
  useJetStreamStatus,
  useRPC,
} from '@sevki/jetstream-react';
import {
  EchoHttpClient,
  rmessageDecode,
  PROTOCOL_VERSION,
  PROTOCOL_NAME,
} from './generated/echohttp_rpc.js';

const SERVER_URL = `https://127.0.0.1:4433/${PROTOCOL_NAME}`;

function EchoDemo() {
  const status = useJetStreamStatus();
  const echo = useJetStream(EchoHttpClient, rmessageDecode, PROTOCOL_VERSION);
  const [message, setMessage] = useState('hello');
  const [sum, setSum] = useState<number | null>(null);

  const { data, error, isLoading } = useRPC(
    () => (echo ? echo.ping(message) : Promise.resolve('')),
    [echo, message],
  );

  return (
    <div>
      <h1>JetStream Echo Demo</h1>
      <p>Connection: {status}</p>
      <p>Protocol: {PROTOCOL_VERSION}</p>

      <h2>Ping</h2>
      <input
        value={message}
        onChange={e => setMessage(e.target.value)}
        placeholder="Type a message..."
      />
      {isLoading && <p>Loading...</p>}
      {error && <p>Error: {error.message}</p>}
      {data && <p>Echo: {data}</p>}

      <h2>Add</h2>
      <button
        disabled={!echo}
        onClick={async () => {
          if (echo) setSum(await echo.add(2, 3));
        }}
      >
        Add 2 + 3
      </button>
      {sum !== null && <p>Sum: {sum}</p>}
    </div>
  );
}

export default function App() {
  return (
    <JetStreamProvider url={SERVER_URL}>
      <EchoDemo />
    </JetStreamProvider>
  );
}

Package Structure

Package	Description
`@sevki/jetstream-wireformat`	Binary codecs for primitives and composite types
`@sevki/jetstream-rpc`	RPC runtime: `Mux`, `TagPool`, framing, version negotiation
`@sevki/jetstream-react`	React hooks: `JetStreamProvider`, `useJetStream`, `useRPC`, `useHandler`

Swift

JetStream provides Swift packages for wire format encoding and RPC communication via Swift Package Manager.

Installation

Add the JetStream package to your Package.swift:

dependencies: [
    .package(url: "https://github.com/sevki/jetstream.git", from: "15.0.0"),
],
targets: [
    .target(
        name: "MyApp",
        dependencies: [
            .product(name: "JetStreamWireFormat", package: "jetstream"),
            .product(name: "JetStreamRpc", package: "jetstream"),
        ]
    ),
]

WireFormat Protocol

The JetStreamWireFormat module defines the WireFormat protocol that all serializable types conform to:

public protocol WireFormat: Sized {
    func byteSize() -> UInt32
    func encode(writer: inout BinaryWriter) throws
    static func decode(reader: inout BinaryReader) throws -> Self
}

Primitives

All Swift standard types (UInt8, UInt16, UInt32, UInt64, Int16, Int32, Int64, Float, Double, Bool, String) conform to WireFormat:

import JetStreamWireFormat

// Encode a UInt32
var writer = BinaryWriter()
try UInt32(42).encode(writer: &writer)
let bytes = writer.bytes

// Decode it back
var reader = BinaryReader(data: bytes)
let value = try UInt32.decode(reader: &reader) // 42

Collections and Optionals

Arrays, Dictionaries, Sets, and Optionals conform to WireFormat when their elements do:

// Array<String>
let tags: [String] = ["hello", "world"]
try tags.encode(writer: &writer)

// Optional<UInt32>
let maybeId: UInt32? = 42
try maybeId.encode(writer: &writer)

// Dictionary<String, UInt32>
let scores: [String: UInt32] = ["alice": 100, "bob": 200]
try scores.encode(writer: &writer)

Structs and Enums

Define custom types conforming to WireFormat, or generate them from Rust types using jetstream_codegen:

import JetStreamWireFormat

public struct Point: WireFormat {
    public var x: UInt32
    public var y: UInt32

    public func byteSize() -> UInt32 {
        return x.byteSize() + y.byteSize()
    }

    public func encode(writer: inout BinaryWriter) throws {
        try x.encode(writer: &writer)
        try y.encode(writer: &writer)
    }

    public static func decode(reader: inout BinaryReader) throws -> Point {
        let x = try UInt32.decode(reader: &reader)
        let y = try UInt32.decode(reader: &reader)
        return Point(x: x, y: y)
    }
}

Code Generation

Instead of writing conformances by hand, use jetstream_codegen to generate Swift types from Rust source files:

cargo run -p jetstream_codegen -- \
  --input src/types.rs \
  --swift-out generated/

Given a Rust file:

#![allow(unused)]
fn main() {
#[derive(JetStreamWireFormat)]
pub struct Point {
    pub x: u32,
    pub y: u32,
}

#[derive(JetStreamWireFormat)]
pub enum Shape {
    Circle(u32),
    Rectangle { width: u32, height: u32 },
}
}

The codegen produces Swift structs and enums conforming to WireFormat that are wire-compatible with the Rust implementations.

RPC

The JetStreamRpc module provides the RPC runtime for multiplexed request/response communication.

Generated Client

The codegen generates RPC client classes and handler protocols from #[service] trait definitions:

#![allow(unused)]
fn main() {
// Rust service definition
#[service]
pub trait Echo {
    async fn ping(&mut self, message: String) -> Result<String>;
    async fn add(&mut self, a: u32, b: u32) -> Result<u32>;
}
}

cargo run -p jetstream_codegen -- \
  --input examples/echo.rs \
  --swift-out generated/

This generates a typed EchoClient class and an EchoHandler protocol:

// Generated client usage
let client = EchoClient(mux: mux)
let reply = try await client.ping(message: "hello")
let sum = try await client.add(a: 2, b: 3)

// Generated handler protocol — implement this for your server
public protocol EchoHandler {
    func ping(message: String) async throws -> String
    func add(a: UInt32, b: UInt32) async throws -> UInt32
}

Framer Protocol

The Framer protocol handles message type dispatch for RPC envelopes:

public protocol Framer {
    func messageType() -> UInt8
    func byteSize() -> UInt32
    func encode(writer: inout BinaryWriter) throws
    static func decode(reader: inout BinaryReader, type: UInt8) throws -> Self
}

Generated Tmessage and Rmessage enums conform to Framer and dispatch encoding/decoding based on the message type byte.

Frame Wire Format

Mux

The Mux class handles tag allocation, request/response matching, and concurrent RPC dispatch:

let mux = Mux(transport: transport)

// Each rpc() call acquires a tag, sends a frame, waits for the matching response, and releases the tag
let response = try await mux.rpc(request)