Sockudo HTTP Rust Client

A fast, safe, and idiomatic Rust client for interacting with the Sockudo HTTP API, allowing you to publish events, authorize channels, authenticate users, and handle webhooks from your Rust applications.

Features

• Trigger events on public, private, and presence channels
• Trigger events to specific users (User Authentication)
• Trigger batch events for efficiency
• Support for end-to-end encrypted channels
• Authorize client subscriptions to private, presence, and encrypted channels
• Authenticate users for user-specific Sockudo features
• Terminate user connections
• Validate and process incoming Sockudo webhooks
• Configurable host, port, scheme (HTTP/HTTPS), and timeout
• Asynchronous API using async/await
• Typed responses and errors
• Fast JSON with SIMD-accelerated sonic-rs library

Installation

Add the following to your Cargo.toml:

[dependencies]
sockudo-http = "2.2.0"
sonic-rs = "0.5"
tokio = { version = "1", features = ["full"] }

Then run:

cargo build

Usage

1. Initialization

Configure and create a client:

use sockudo_http::{Config, Sockudo, SockudoError};

#[tokio::main]
async fn main() -> Result<(), SockudoError> {
    let config = Config::builder()
        .app_id("YOUR_APP_ID")
        .key("YOUR_APP_KEY")
        .secret("YOUR_APP_SECRET")
        .host("127.0.0.1")
        .port(6001)
        .use_tls(false)
        .timeout(std::time::Duration::from_secs(5)) // Optional
        .build()?;

    let sockudo = Sockudo::new(config)?;

    // Your application logic here...
    Ok(())
}

For encrypted channels, add the encryption master key:

let config = Config::builder()
    .app_id("YOUR_APP_ID")
    .key("YOUR_APP_KEY")
    .secret("YOUR_APP_SECRET")
    .host("127.0.0.1")
    .port(6001)
    .use_tls(false)
    .encryption_master_key_base64("YOUR_BASE64_ENCRYPTION_MASTER_KEY")?
    .build()?;

You can also initialize from a URL:

use sockudo_http::{Sockudo, SockudoError};

let sockudo = Sockudo::from_url(
    "http://YOUR_APP_KEY:YOUR_APP_SECRET@127.0.0.1:6001/apps/YOUR_APP_ID",
    None,
)?;

2. Triggering Events

use sockudo_http::{Sockudo, Channel, SockudoError};
use sonic_rs::json;

async fn trigger_event(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let channels = vec![Channel::from_string("my-channel")?];
    let event_name = "new-message";
    let data = json!({ "text": "Hello from Rust!" });

    match sockudo.trigger(&channels, event_name, data, None).await {
        Ok(response) => {
            println!("Event triggered! Status: {}", response.status());
        }
        Err(e) => eprintln!("Error triggering event: {:?}", e),
    }
    Ok(())
}

Encrypted channels: If channels contains a single encrypted channel (e.g., "private-encrypted-mychannel") and you've set the encryption_master_key in the Config, the library will encrypt data automatically.

Excluding a recipient:

use sockudo_http::{Sockudo, Channel, SockudoError, events::TriggerParams};
use sonic_rs::json;

async fn trigger_event_exclude(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let channels = vec![Channel::from_string("my-channel")?];
    let event_name = "new-message";
    let data = json!({ "text": "Hello from Rust!" });
    
    let params = TriggerParams::builder()
        .socket_id("socket_id_to_exclude")
        .build();

    sockudo.trigger(&channels, event_name, data, Some(params)).await?;
    Ok(())
}

Idempotency key:

Attach an idempotency key so the server deduplicates the event on retries, ensuring at-most-once delivery:

let params = TriggerParams::builder()
    .idempotency_key("unique-key-for-this-event")
    .build();

sockudo.trigger(&channels, event_name, data, Some(params)).await?;

3. Triggering Batch Events

use sockudo_http::{Sockudo, SockudoError, events::BatchEvent};
use sonic_rs::json;

async fn trigger_batch(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let batch = vec![
        BatchEvent::new("event1", "channel-a", json!({ "value": 1 })),
        BatchEvent::new("event2", "channel-b", json!({ "value": 2 })),
    ];

    match sockudo.trigger_batch(batch).await {
        Ok(response) => println!("Batch triggered! Status: {}", response.status()),
        Err(e) => eprintln!("Error triggering batch: {:?}", e),
    }
    Ok(())
}

4. Tag Filtering

Tag filtering allows you to add metadata tags to events, enabling clients to filter which events they receive based on tag values. This can significantly reduce bandwidth usage (60-90%) in high-volume scenarios.

Triggering a single event with tags:

use sockudo_http::{Sockudo, Channel, SockudoError, events::TriggerParams};
use sonic_rs::json;
use std::collections::HashMap;

async fn trigger_with_tags(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let channels = vec![Channel::from_string("sports-updates")?];
    let event_name = "match-event";
    let data = json!({
        "match_id": "123",
        "team": "Home",
        "player": "John Doe",
        "minute": 45
    });

    // Create tags for filtering
    let mut tags = HashMap::new();
    tags.insert("event_type".to_string(), "goal".to_string());
    tags.insert("priority".to_string(), "high".to_string());

    let params = TriggerParams::builder()
        .tags(tags)
        .build();

    sockudo.trigger(&channels, event_name, data, Some(params)).await?;
    Ok(())
}

Triggering batch events with tags:

use sockudo_http::{Sockudo, SockudoError, events::BatchEvent};
use sonic_rs::json;
use std::collections::HashMap;

async fn trigger_batch_with_tags(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let mut goal_tags = HashMap::new();
    goal_tags.insert("event_type".to_string(), "goal".to_string());
    goal_tags.insert("priority".to_string(), "high".to_string());

    let mut shot_tags = HashMap::new();
    shot_tags.insert("event_type".to_string(), "shot".to_string());
    shot_tags.insert("xG".to_string(), "0.85".to_string());

    let batch = vec![
        BatchEvent::new("match-event", "match:123", json!({ "type": "goal", "player": "Smith" }))
            .with_tags(goal_tags),
        BatchEvent::new("match-event", "match:123", json!({ "type": "shot", "player": "Jones" }))
            .with_tags(shot_tags),
    ];

    sockudo.trigger_batch(batch).await?;
    Ok(())
}

Note: Tag filtering must be enabled on the Sockudo server (TAG_FILTERING_ENABLED=true) for clients to filter events. Tags are key-value pairs where both keys and values are strings. Clients can subscribe with filter expressions to receive only events matching their criteria.

5. Authorizing Channels

Typically done in your HTTP handler when a client attempts to subscribe:

use sockudo_http::{Sockudo, Channel, SockudoError};
use sonic_rs::json;

fn authorize_channel(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let socket_id = "123.456";
    let channel_name = "private-mychannel";
    let channel = Channel::from_string(channel_name)?;

    // For presence channels, include user data:
    let presence_data = Some(json!({
        "user_id": "unique_user_id",
        "user_info": { "name": "Alice" }
    }));

    match sockudo.authorize_channel(socket_id, &channel, presence_data.as_ref()) {
        Ok(auth_signature) => {
            println!("Auth success: {:?}", auth_signature);
            // Return `auth_signature` as JSON to client
        }
        Err(e) => eprintln!("Auth error: {:?}", e),
    }
    Ok(())
}

6. Authenticating Users

For server-to-user events:

use sockudo_http::{Sockudo, SockudoError};
use sonic_rs::json;

fn authenticate_user(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let socket_id = "789.012";
    let user_data = json!({
        "id": "user-bob",      // required
        "name": "Bob The Builder",
        "email": "bob@example.com"
    });

    match sockudo.authenticate_user(socket_id, &user_data) {
        Ok(user_auth) => {
            println!("User auth success: {:?}", user_auth);
            // Return `user_auth` as JSON to client
        }
        Err(e) => eprintln!("User auth error: {:?}", e),
    }
    Ok(())
}

7. Sending an Event to a User

use sockudo_http::{Sockudo, SockudoError};
use sonic_rs::json;

async fn send_to_user(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let user_id = "user-bob";
    let event_name = "personal-notification";
    let data = json!({ "alert": "Your report is ready!" });

    match sockudo.send_to_user(user_id, event_name, data).await {
        Ok(response) => println!("Sent to user! Status: {}", response.status()),
        Err(e) => eprintln!("Error sending to user: {:?}", e),
    }
    Ok(())
}

8. Terminating User Connections

use sockudo_http::{Sockudo, SockudoError};

async fn terminate_user(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let user_id = "user-charlie";

    match sockudo.terminate_user_connections(user_id).await {
        Ok(response) => println!("Terminate successful! Status: {}", response.status()),
        Err(e) => eprintln!("Error terminating user: {:?}", e),
    }
    Ok(())
}

9. Force Reconnect User

Close all active connections for a user with code 4200, prompting clients to reconnect:

use sockudo_http::{Sockudo, SockudoError};

async fn force_reconnect(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let user_id = "user-123";
    match sockudo.force_reconnect_user(user_id).await {
        Ok(_) => println!("Force reconnect initiated for {}", user_id),
        Err(e) => eprintln!("Error: {}", e),
    }
    Ok(())
}

10. Handling Webhooks

use sockudo_http::{Sockudo, SockudoError, webhook::WebhookEvent};
use std::collections::BTreeMap;

fn handle_webhook(sockudo: &Sockudo) -> Result<(), SockudoError> {
    let mut headers = BTreeMap::new();
    headers.insert("X-Pusher-Key".to_string(), "YOUR_APP_KEY".to_string());
    headers.insert("X-Pusher-Signature".to_string(), "RECEIVED_SIGNATURE".to_string());
    headers.insert("Content-Type".to_string(), "application/json".to_string());

    let body = r#"{
        "time_ms": 1600000000000,
        "events":[{"name":"channel_occupied","channel":"my-channel"}]
    }"#;

    let webhook = sockudo.webhook(&headers, body);

    if webhook.is_valid(None) {
        println!("Webhook is valid!");
        match webhook.get_events() {
            Ok(events) => {
                for event in events {
                    match event {
                        WebhookEvent::ChannelOccupied { channel } => {
                            println!("Channel occupied: {}", channel);
                        }
                        WebhookEvent::ChannelVacated { channel } => {
                            println!("Channel vacated: {}", channel);
                        }
                        WebhookEvent::MemberAdded { channel, user_id } => {
                            println!("Member {} added to {}", user_id, channel);
                        }
                        WebhookEvent::MemberRemoved { channel, user_id } => {
                            println!("Member {} removed from {}", user_id, channel);
                        }
                        _ => {}
                    }
                }
            }
            Err(e) => eprintln!("Error getting events: {:?}", e),
        }
    } else {
        eprintln!("Invalid webhook!");
    }
    Ok(())
}

11. Example: Integration with Axum

use axum::{
    extract::{Json, State},
    http::{HeaderMap, StatusCode},
    response::IntoResponse,
    routing::post,
    Router,
};
use sockudo_http::{Config, Sockudo, Channel};
use serde::Deserialize;
use sonic_rs::{json, Value};
use std::{collections::BTreeMap, sync::Arc};

#[derive(Clone)]
struct AppState {
    sockudo: Arc<Sockudo>,
}

#[tokio::main]
async fn main() {
    let config = Config::builder()
        .app_id("YOUR_APP_ID")
        .key("YOUR_APP_KEY")
        .secret("YOUR_APP_SECRET")
        .host("127.0.0.1")
        .port(6001)
        .use_tls(false)
        .build()
        .expect("Failed to build config");

    let sockudo = Arc::new(Sockudo::new(config).expect("Failed to create client"));
    let app_state = AppState { sockudo };

    let app = Router::new()
        .route("/sockudo/auth", post(sockudo_auth_handler))
        .route("/sockudo/webhook", post(sockudo_webhook_handler))
        .with_state(app_state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    println!("Listening on {}", listener.local_addr().unwrap());
    axum::serve(listener, app).await.unwrap();
}

#[derive(Deserialize)]
struct AuthRequest {
    socket_id: String,
    channel_name: String,
    #[serde(alias = "channel_data")]
    presence_data: Option<Value>,
}

async fn sockudo_auth_handler(
    State(state): State<AppState>,
    Json(payload): Json<AuthRequest>,
) -> impl IntoResponse {
    let channel = match Channel::from_string(&payload.channel_name) {
        Ok(ch) => ch,
        Err(_) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "Invalid channel_name"})),
            ).into_response();
        }
    };

    match state.sockudo.authorize_channel(
        &payload.socket_id,
        &channel,
        payload.presence_data.as_ref(),
    ) {
        Ok(auth_response) => (StatusCode::OK, Json(auth_response)).into_response(),
        Err(e) => {
            eprintln!("Auth error: {:?}", e);
            (StatusCode::FORBIDDEN, Json(json!({ "error": "Forbidden" }))).into_response()
        }
    }
}

async fn sockudo_webhook_handler(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: String,
) -> impl IntoResponse {
    let mut hdrs_btreemap = BTreeMap::new();
    for (k, v) in headers.iter() {
        if let Ok(s) = v.to_str() {
            hdrs_btreemap.insert(k.as_str().to_string(), s.to_string());
        }
    }

    let webhook = state.sockudo.webhook(&hdrs_btreemap, &body);

    if webhook.is_valid(None) {
        (StatusCode::OK, Json(json!({ "status": "ok" }))).into_response()
    } else {
        (StatusCode::UNAUTHORIZED, Json(json!({ "error": "Unauthorized" }))).into_response()
    }
}

Configuration Options

The Config struct is used to configure the Sockudo client. Create it using Config::builder():

Call .build() on the ConfigBuilder to get a Result<Config, SockudoError>.

Channel History

use sockudo_http::{HistoryParams, Sockudo};

async fn fetch_history(sockudo: &Sockudo) -> Result<(), sockudo_http::SockudoError> {
    let page = sockudo
        .channel_history_with_name(
            "my-channel",
            Some(&HistoryParams {
                limit: Some(50),
                direction: Some("newest_first".to_string()),
                ..Default::default()
            }),
        )
        .await?;

    if let Some(cursor) = page.next_cursor {
        let _next_page = sockudo
            .channel_history_with_name(
                "my-channel",
                Some(&HistoryParams {
                    cursor: Some(cursor),
                    ..Default::default()
                }),
            )
            .await?;
    }

    Ok(())
}

Error Handling

All fallible methods return Result<T, SockudoError>. The SockudoError enum variants:

Contributing

Contributions are welcome! Please open issues for bugs or feature requests, or submit pull requests for improvements.

When contributing code, please ensure:

• Code is formatted with cargo fmt
• Clippy lints are addressed (cargo clippy --all-targets --all-features)
• New functionality is covered by tests
• Documentation is updated accordingly

License

This project is licensed under the GNU Affero General Public License v3.0. See the LICENSE.md file for details.