= Examples
:sectanchors:
:toc: left
:toclevels: 2
== Chat Server (~30 lines)
[source,rust]
----
use std::sync::Arc;
use rifts::RiftServer;
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== Authenticated Pub/Sub
[source,rust]
----
use std::sync::Arc;
use rifts::{
RiftServer, ServerConfig, AuthMode,
session::{TokenAuth, AuthContext, ClientId},
};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
let auth = Arc::new(TokenAuth::new());
auth.register("admin-token", AuthContext {
client_id: ClientId::new("admin"),
claims: serde_json::json!({"role": "admin"}),
mode: AuthMode::Bearer,
hints: Default::default(),
});
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.auth(auth)
.config(ServerConfig::default())
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== Sled Persistence
[source,rust]
----
// Cargo.toml
// rifts = { version = "0.1", features = ["sled"] }
use std::sync::Arc;
use std::time::Duration;
use rifts::{
RiftServer, TopicProfile,
broker::InMemoryBroker,
storage::{
SledEngine, SledOffsetStore, SledLogStore,
SledDedupeStore, SledSnapshotStore,
},
};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
let db = sled::Config::new()
.path("/var/lib/rifts/broker")
.open()?;
let broker = InMemoryBroker::with_stores(
TopicProfile::default(),
Duration::from_secs(60),
65_536,
SledOffsetStore::new(SledEngine::new(db.open_tree(b"offsets")?)),
SledLogStore::new(SledEngine::new(db.open_tree(b"log")?)),
SledDedupeStore::new(SledEngine::new(db.open_tree(b"dedupe")?)),
SledSnapshotStore::new(SledEngine::new(db.open_tree(b"snapshots")?)),
);
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.broker(Arc::new(broker))
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== Actor‑Based Broker (per‑topic parallelism)
[source,rust]
----
use std::sync::Arc;
use std::time::Duration;
use rifts::{
RiftServer, TopicProfile,
actor::TopicRegistry,
broker::ActorBroker,
storage::{
MemoryOffsetStore, MemoryLogStore,
MemoryDedupeStore, MemorySnapshotStore,
},
};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
let registry = Arc::new(TopicRegistry::new(
Arc::new(MemoryOffsetStore::new()),
Arc::new(MemoryLogStore::new()),
Arc::new(MemoryDedupeStore::new()),
Arc::new(MemorySnapshotStore::new()),
TopicProfile::default(),
Duration::from_secs(60),
));
let broker = ActorBroker::new(registry);
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.broker(Arc::new(broker))
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== RemoteBroker (Gateway ↔ External Broker Node)
[source,rust]
----
use std::sync::Arc;
use rifts::{RiftServer, broker::RemoteBroker};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
let broker = RemoteBroker::connect("192.168.1.10:9200".parse()?).await?;
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.broker(Arc::new(broker))
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
The broker node must speak the `WireMsg` CBOR protocol defined in
`src/broker/wire.rs`. Any language / stack can implement it.
== Direct Broker Usage (no network)
[source,rust]
----
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use rifts::{
broker::{InMemoryBroker, Broker, SubscribeIntent},
frame::{Frame, FrameType, Codec},
TopicProfile, RetentionPolicy,
};
#[tokio::main]
async fn main() -> rifts::Result<()> {
let profile = TopicProfile {
retention: RetentionPolicy::Count(100),
..TopicProfile::default()
};
let broker = InMemoryBroker::new(profile, Duration::from_secs(60), 65_536);
let frame = Frame {
frame_type: FrameType::Data,
codec: Codec::Json,
topic: Some("test".into()),
message_id: Some("msg-1".into()),
payload: Some(Bytes::from_static(b"hello")),
..Frame::default()
};
let outcome = broker.publish(&frame).await?;
println!("published at offset {}", outcome.offset);
Ok(())
}
----
== Axum WebSocket Adapter
[source,rust]
----
// Cargo.toml
// rifts = { version = "0.1", default-features = false, features = ["axum"] }
use axum::{Router, routing::get, extract::ws::WebSocketUpgrade, response::IntoResponse};
use rifts::transport::axum::AxumWsConnection;
async fn ws_upgrade(ws: WebSocketUpgrade) -> impl IntoResponse {
ws.on_upgrade(|socket| async move {
let conn = AxumWsConnection::new(socket);
// Pass conn to your Rift connection handler.
})
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/ws", get(ws_upgrade));
let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
axum::serve(listener, app).await.unwrap();
}
----
== Custom AuthProvider
[source,rust]
----
use async_trait::async_trait;
use rifts::{
session::{AuthProvider, AuthContext, AuthHints, ClientId},
AuthMode, Result, RiftError,
};
struct MyAuth {
jwks_url: String,
}
#[async_trait]
impl AuthProvider for MyAuth {
async fn authenticate(&self, mode: AuthMode, token: Option<&str>)
-> Result<AuthContext>
{
let token = token.ok_or_else(||
RiftError::Auth(rifts::error::AuthReject::Required)
)?;
let claims = validate_jwt(token, &self.jwks_url).await?;
Ok(AuthContext {
client_id: ClientId::new(claims.sub),
claims: serde_json::to_value(&claims).unwrap(),
mode,
hints: AuthHints::default(),
})
}
async fn revoke(&self, _client_id: &ClientId) -> Result<()> {
Ok(())
}
}
----
== Custom TopicProfile
[source,rust]
----
use std::sync::Arc;
use std::time::Duration;
use rifts::{
RiftServer, TopicProfile, RetentionPolicy, OrderingPolicy,
};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
// A topic profile for a chat room: keep last 500 messages,
// ordered by topic, replay enabled, max 200 subscribers.
let profile = TopicProfile {
retention: RetentionPolicy::Count(500),
ordering: OrderingPolicy::Topic,
max_subscribers: 200,
max_publishers: 50,
replay_enabled: true,
snapshot_enabled: true,
..TopicProfile::default()
};
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.default_topic_profile(profile)
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== Metrics Export (Prometheus)
[source,rust]
----
use std::sync::Arc;
use std::time::Duration;
use rifts::{RiftServer, Metrics};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
let metrics = Arc::new(Metrics::new());
// Spawn a background task that snapshots metrics every 15 s.
let m = metrics.clone();
tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(15)).await;
println!(
"conns={} msgs_in={} msgs_out={} dropped={} queue_depth={}",
m.active_connections.load(std::sync::atomic::Ordering::Relaxed),
m.messages_in_total.load(std::sync::atomic::Ordering::Relaxed),
m.messages_out_total.load(std::sync::atomic::Ordering::Relaxed),
m.messages_dropped_total.load(std::sync::atomic::Ordering::Relaxed),
m.send_queue_depth.load(std::sync::atomic::Ordering::Relaxed),
);
}
});
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.metrics(metrics)
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== Flow Control: Per‑Connection Rate Limiting
[source,rust]
----
use std::sync::Arc;
use rifts::{
RiftServer, ServerConfig,
flow::RateLimiter,
};
use tokio::sync::Notify;
#[tokio::main]
async fn main() -> rifts::Result<()> {
// Limit each connection to 100 messages/sec with burst of 200.
let limiter = Arc::new(RateLimiter::new(100, 200));
let config = ServerConfig {
max_send_queue_bytes: 512 * 1024, // 512 KiB send queue
..ServerConfig::default()
};
let shutdown = Arc::new(Notify::new());
RiftServer::builder()
.websocket_transport()
.config(config)
.rate_limiter(limiter)
.build()?
.run("0.0.0.0:9000".parse().unwrap(), shutdown)
.await?;
Ok(())
}
----
== Trace Context Propagation
[source,rust]
----
use rifts::trace::TraceContext;
fn handle_incoming_frame(incoming: TraceContext) {
// Create a child span for server-side processing.
let server_span = incoming.child();
println!(
"trace_id={:?} span_id={:?} parent={:?}",
server_span.trace_id,
server_span.span_id,
server_span.parent_span_id,
);
// Pass server_span to downstream calls (DB, RPC, etc.).
do_work(server_span);
}
fn do_work(ctx: TraceContext) {
let db_span = ctx.child();
// ... use db_span for database query tracing
}
----