OddSockets Rust SDK

Overview & Features

The OddSockets Rust SDK provides a high-performance, async-first interface for real-time messaging with full Rust type safety and zero-cost abstractions.

Installation

[dependencies]
oddsockets = "0.1.0-beta.1"
tokio = { version = "1.0", features = ["full"] }

cargo add oddsockets
cargo add tokio --features full

[dependencies]
oddsockets = { git = "https://github.com/jyswee/oddsockets-rust-sdk" }
tokio = { version = "1.0", features = ["full"] }

Quick Start

Basic Usage

use oddsockets::{OddSocketsClient, OddSocketsConfig};
use tokio;

#[tokio::main]
async fn main() -> Result<(), Box> {
    // Create a client
    let config = OddSocketsConfig::new("ak_live_1234567890abcdef");
    let client = OddSocketsClient::new(config).await?;

    // Connect to OddSockets
    client.connect().await?;

    // Get a channel
    let channel = client.channel("my-channel");

    // Subscribe to messages
    let mut message_stream = channel.subscribe(Default::default()).await?;
    
    // Publish a message
    channel.publish("Hello, Rust!", Default::default()).await?;

    // Listen for messages
    while let Some(message) = message_stream.recv().await {
        println!("Received: {:?}", message);
    }

    Ok(())
}

Advanced Configuration

use oddsockets::OddSocketsConfig;
use std::time::Duration;

let config = OddSocketsConfig::builder("ak_live_1234567890abcdef")
    .high_performance() // Optimized for high-performance scenarios
    .heartbeat_interval(Duration::from_secs(60))
    .reconnect_attempts(3)
    .build()?;

Bulk Publishing

use oddsockets::{BulkMessage, message_types};

let messages = vec![
    BulkMessage::new("channel1", message_types::chat_message("Hello", "user1", None), None),
    BulkMessage::new("channel2", message_types::chat_message("World", "user2", None), None),
];

let results = client.publish_bulk(messages).await?;
for result in results {
    if result.is_successful() {
        println!("Message published successfully");
    }
}

Configuration

Client Options

let config = OddSocketsConfig::builder("ak_live_1234567890abcdef")
    .user_id("user-id".to_string())           // Optional: User identifier
    .auto_connect(true)                       // Optional: Auto-connect on creation
    .reconnect_attempts(5)                    // Optional: Max reconnection attempts
    .heartbeat_interval(Duration::from_secs(30)) // Optional: Heartbeat interval
    .build()?;

Channel Options

let subscribe_options = SubscribeOptions::builder()
    .enable_presence(true)                    // Enable presence tracking
    .retain_history(true)                     // Retain message history
    .filter("user.premium == true".to_string()) // Message filter expression
    .build();

let publish_options = PublishOptions::builder()
    .ttl(3600)                               // Time to live (seconds)
    .metadata(serde_json::json!({"priority": "high"})) // Additional metadata
    .store_in_history(true)                  // Store in message history
    .build();

Examples

Explore comprehensive examples demonstrating the OddSockets Rust SDK in action:

Enhanced Features

Beyond core pub/sub, OddSockets ships a Slack-like enhanced surface — reactions, typing indicators, threads, read receipts, presence/status, notifications, DMs, channel management, message editing and search. It lives on the EnhancedFeatures struct, which wraps a shared handle to your connected client. You send an action with an enhanced.* method (snake_case, positional arguments, all async) and receive the paired broadcast with client.on("<event>", |data| { ... }) — the worker forwards every enhanced broadcast onto the client's raw event surface as a serde_json::Value.

OddSocketsClient is cheap to clone and every clone shares the same underlying socket, so wrapping a clone in Arc<RwLock<...>> gives EnhancedFeatures a handle to the very same connection your listeners are attached to.

Typing & Reactions

use std::sync::Arc;
use tokio::sync::RwLock;
use oddsockets::{EnhancedFeatures, OddSocketsClient, OddSocketsConfig};

let config = OddSocketsConfig::new("ak_your_api_key_here");
let client = OddSocketsClient::new(config).await?;
client.connect().await?;

let channel = client.channel("room-42");
let _stream = channel.subscribe(Default::default()).await?;

// Receive-path: enhanced broadcasts arrive on the client's raw event surface
client.on("user_typing",    |data| println!("someone is typing: {data}"));
client.on("reaction_added", |data| println!("reaction added: {data}"));

// Send-path: wrap a clone (same socket) for the enhanced surface
let enhanced = EnhancedFeatures::new(Arc::new(RwLock::new(client.clone())));

enhanced.start_typing("alice", "room-42").await?;
enhanced.add_reaction("msg-1", "room-42", ":thumbsup:", "alice", "Alice").await?;

Threads

client.on("thread_reply", |data| println!("new thread reply: {data}"));

// Request-style methods return the worker response as a serde_json::Value
let reply = enhanced
    .thread_reply("room-42", "msg-1", "Replying in the thread", "alice", "Alice")
    .await?;
println!("thread reply ack: {reply}");

Fire-and-forget actions return Result<(), OddSocketsError>; query and request-style methods (get_*, search_*, thread_reply, create_channel, …) await the worker acknowledgement and return Result<Value, OddSocketsError>.

• Typing: start_typing, stop_typing → user_typing, user_stopped_typing
• Reactions: add_reaction, remove_reaction, get_reactions → reaction_added, reaction_removed
• Threads: thread_reply, get_thread, subscribe_thread, follow_thread, unfollow_thread, mark_thread_read → thread_reply, thread_subscribed, thread_followed, thread_read_updated
• Read receipts: mark_read, mark_all_read, get_unread_counts → user_read, unread_count_updated, all_marked_read
• Messages: edit_message, delete_message, pin_message, unpin_message, get_pinned_messages → message_edited, message_deleted, message_pinned, message_unpinned
• Presence & status: set_status, set_custom_status, clear_custom_status, set_dnd, clear_dnd, get_user_presence → user_status_changed, custom_status_updated, dnd_status_changed
• Channels: create_channel, update_channel, archive_channel, invite_to_channel, join_channel, leave_channel, get_channel_members → channel_created, channel_updated, user_invited, user_joined_channel, user_left_channel
• DMs: create_dm, send_dm, get_dm_conversations → dm_created, dm_received
• Notifications: subscribe_notifications, get_notifications, mark_notification_read, clear_notifications → notification, notification_read, notifications_cleared
• Search: search_messages, search_in_channel, search_by_user, filter_messages (query results returned as Value)

For any worker event not wrapped above, subscribe with the raw client.on("<event>", |data| { ... }) API — all enhanced broadcasts are forwarded onto the client surface.

Challenges & Leaderboards

Challenges, leaderboards and achievements build on the same live socket as the enhanced surface. The send side lives on EnhancedFeatures: request/query methods (create_challenge, complete_challenge, get_standings, …) resolve with the worker's reply as a serde_json::Value, while progress and achievement calls (report_progress, unlock_achievement) are fire-and-forget. Inbound broadcasts arrive on the client event surface — subscribe with the client's normal client.on("<event>", |data| { ... }).

use serde_json::json;

// Register a ranked leaderboard, then report progress and finalise a run.
let challenge = enhanced.create_challenge(json!({
    "challengeId": "weekly-sprint",
    "metric": "score",
    "ranked": true,
    "channel": "room-42",
})).await?;

enhanced.report_progress(json!({ "challengeId": "weekly-sprint", "value": 4200 })).await?;

let standings = enhanced.get_standings(json!({ "challengeId": "weekly-sprint", "limit": 10 })).await?;
println!("top 10 + my rank: {standings}");

enhanced.complete_challenge(json!({ "challengeId": "weekly-sprint", "outcome": "completed" })).await?;

Each method takes a single serde_json::Value params object. Request/query methods await the worker acknowledgement and return Result<Value, OddSocketsError>; fire-and-forget methods return Result<(), OddSocketsError>.

• create_challenge — create a challenge / leaderboard. Awaits challenge_create_success.
• report_progress — fire-and-forget metric progress (no ack).
• complete_challenge — finalize a run with an outcome. Awaits challenge_complete_success.
• unlock_achievement — fire-and-forget; pass percentComplete (0–100) (no ack).
• get_standings — request top-N plus the caller's rank. Awaits challenge_standings_success.
• get_achievements — query achievement state. Awaits achievement_state.
• send_challenge_invite — directed invite to another user. Awaits challenge_invite_success.
• reply_challenge_invite — accept / decline an invite. Awaits challenge_reply_success.
• cancel_challenge_invite — cancel a sent invite. Awaits challenge_invite_cancel_success.
• get_challenge_invites — list pending invites. Awaits challenge_invites.

Completion outcomes

The outcome field passed to complete_challenge is one of:

• completed — win (rank 1)
• failed — loss
• tied — draw
• conceded — resign / concede
• expired — timed out

Progressive achievements

unlock_achievement always emits the wire event achievement_unlock; the worker is authoritative and derives the outbound broadcast from percentComplete: a value < 100 broadcasts achievement_progress (status in_progress), while >= 100 or an omitted percentComplete broadcasts achievement_unlock (status unlocked). You never emit achievement_progress yourself.

Inbound events

Subscribe to these via the client's client.on("<event>", |data| { ... }) surface:

• Room broadcasts: challenge_progress, leaderboard_rank_change, challenge_complete, achievement_unlock, achievement_progress
• Directed (per-user): challenge_invited, challenge_reply_received, challenge_invite_cancelled

Usage Analytics

Pull your tenant's headline usage tiles — monthly active users, daily active users, total messages published, and error rate — straight from the SDK, without hand-rolling a REST call. get_usage_stats() resolves the same four tiles the developer dashboard renders.

let stats = client.get_usage_stats().await?;

// Each tile is an Option: None means "not live yet", never a real zero.
let dash = |v: Option<i64>| v.map(|n| n.to_string()).unwrap_or_else(|| "\u{2014}".into());
println!("MAU:      {}", dash(stats.mau));
println!("DAU:      {}", dash(stats.dau));
println!("Messages: {}", dash(stats.total_messages));
match stats.error_rate {
    Some(rate) => println!("Errors:   {:.3}", rate),
    None => println!("Errors:   \u{2014}"), // render an em-dash for None
}

The four tiles

• mau — monthly active users for your owner scope
• dau — daily active users
• total_messages — total messages published
• error_rate — publish error rate, 0–1

Honesty rule — None, never a fake zero

Each tile is an Option — Some(n) or None. A None means that leg of the analytics pipeline is not live yet for your tenant — the SDK returns it verbatim and never coerces it to Some(0). Render an em-dash (—) for a None tile so you never show a fabricated zero.

Requires an API key

get_usage_stats() reads your owner-scoped analytics, so it needs an api_key. Keyless / token-only clients have no owner scope to query and will return an error stating getUsageStats requires an apiKey.

Performance & Compatibility

OddSockets Rust SDK delivers superior performance with broad compatibility:

Async Patterns

The OddSockets Rust SDK is built with async-first design patterns. Here are common async usage patterns:

Concurrent Message Handling

use tokio::task;
use futures::stream::StreamExt;

#[tokio::main]
async fn main() -> Result<(), Box> {
    let client = OddSocketsClient::new(config).await?;
    let channel = client.channel("events");
    
    let mut message_stream = channel.subscribe(Default::default()).await?;
    
    // Process messages concurrently
    while let Some(message) = message_stream.next().await {
        task::spawn(async move {
            // Handle each message in its own task
            process_message(message).await;
        });
    }
    
    Ok(())
}

async fn process_message(message: Message) {
    // Your message processing logic here
    println!("Processing: {:?}", message);
}

Error Handling with Results

use oddsockets::{OddSocketsError, OddSocketsResultExt};

async fn handle_messages() -> Result<(), OddSocketsError> {
    let client = OddSocketsClient::new(config).await?;
    let channel = client.channel("my-channel");
    
    // Publish with error handling
    match channel.publish("Hello", Default::default()).await {
        Ok(result) => println!("Published: {:?}", result),
        Err(OddSocketsError::MessageTooLarge { size_kb, max_size_kb, message }) => {
            eprintln!("Message too large: {}KB > {}KB", size_kb, max_size_kb);
            eprintln!("Suggestion: {}", message);
        }
        Err(e) => {
            eprintln!("Error: {}", e);
            // Check if error is recoverable
            if e.is_recoverable() {
                println!("Retrying...");
                // Implement retry logic
            }
        }
    }
    
    Ok(())
}

Graceful Shutdown

use tokio::signal;
use tokio::select;

#[tokio::main]
async fn main() -> Result<(), Box> {
    let client = OddSocketsClient::new(config).await?;
    let channel = client.channel("my-channel");
    let mut message_stream = channel.subscribe(Default::default()).await?;
    
    loop {
        select! {
            // Handle incoming messages
            message = message_stream.recv() => {
                if let Some(msg) = message {
                    println!("Received: {:?}", msg);
                }
            }
            
            // Handle shutdown signal
            _ = signal::ctrl_c() => {
                println!("Shutting down gracefully...");
                client.disconnect().await?;
                break;
            }
        }
    }
    
    Ok(())
}