hero/core/dispatcher

Hero Dispatcher

A Redis-based job dispatcher for managing Rhai/HeroScript execution across distributed workers.

Overview

The Hero Dispatcher provides a robust job queue system where:

• Jobs represent script execution requests (Rhai or HeroScript)
• Creating a job stores job parameters in Redis as an hset entry
• Submitting a job pushes the job ID to a worker's queue
• Running a job creates, submits, and awaits results on a dedicated reply queue

Key Features

• Asynchronous Operations: Built with tokio for non-blocking I/O
• Request-Reply Pattern: Submit jobs and await results without polling
• Configurable Jobs: Set timeouts, retries, concurrency, and logging options
• Worker Targeting: Direct job routing to specific worker queues
• Job Lifecycle: Create, submit, monitor status, and retrieve results

Core Components

DispatcherBuilder

Builder for creating Dispatcher instances with caller ID, worker ID, context ID, and Redis URL.

Dispatcher

Main interface for job management:

• new_job() - Create a new JobBuilder
• create_job() - Store job in Redis
• run_job_and_await_result() - Execute job and wait for completion
• get_job_status() - Check job execution status
• get_job_output() - Retrieve job results

JobBuilder

Fluent builder for configuring jobs:

• script() - Set the script content
• worker_id() - Target specific worker
• timeout() - Set execution timeout
• build() - Create the job
• submit() - Fire-and-forget submission
• await_response() - Submit and wait for result

Job

Represents a script execution request with:

• Unique ID and timestamps
• Script content and target worker
• Execution settings (timeout, retries, concurrency)
• Logging configuration

Redis Schema

Jobs are stored using the hero: namespace:

• hero:job:{job_id} - Job parameters as Redis hash
• hero:work_queue:{worker_id} - Worker-specific job queues
• hero:reply:{job_id} - Dedicated reply queues for results

Prerequisites

• Redis server accessible by dispatcher and workers

Usage Example

Basic Job Creation and Submission

use hero_dispatcher::{DispatcherBuilder, DispatcherError};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create dispatcher
    let dispatcher = DispatcherBuilder::new()
        .caller_id("my-app")
        .worker_id("worker-1")
        .context_id("my-context")
        .redis_url("redis://127.0.0.1:6379")
        .build()?;

    // Create a job
    let job = dispatcher
        .new_job()
        .script(r#"print("Hello from worker!"); "success""#)
        .timeout(Duration::from_secs(30))
        .build()?;

    // Store job in Redis
    dispatcher.create_job(&job)?;
    println!("Job {} created and stored in Redis", job.id);

    // Run job and await result (requires worker)
    match dispatcher.run_job_and_await_result(&job, "worker-1".to_string()) {
        Ok(result) => println!("Job completed: {}", result),
        Err(DispatcherError::Timeout(_)) => println!("Job timed out"),
        Err(e) => println!("Job failed: {}", e),
    }

    Ok(())
}

Job Status Monitoring

// Check job status
match dispatcher.get_job_status(&job.id) {
    Ok(status) => println!("Job status: {:?}", status),
    Err(e) => println!("Error getting status: {}", e),
}

// Get job output
match dispatcher.get_job_output(&job.id) {
    Ok(output) => println!("Job output: {:?}", output),
    Err(e) => println!("Error getting output: {}", e),
}

Examples

Run the comprehensive demo to see dispatcher functionality and Redis entries:

cargo run --example dispatcher_demo

Other examples:

• timeout_example.rs - Demonstrates timeout handling

Ensure Redis is running at redis://127.0.0.1:6379.