update api, fix tests and examples
This commit is contained in:
Timur Gordon
182
examples/README.md
Normal file
182
examples/README.md
Normal file
@@ -0,0 +1,182 @@
|
||||
# Hero Supervisor Examples
|
||||
|
||||
This directory contains examples demonstrating the new job API functionality and workflows.
|
||||
|
||||
## Examples Overview
|
||||
|
||||
### 1. `job_api_examples.rs` - Comprehensive API Demo
|
||||
Complete demonstration of all new job API methods:
|
||||
- **Fire-and-forget execution** using `job.run`
|
||||
- **Asynchronous processing** with `jobs.create`, `job.start`, `job.status`, `job.result`
|
||||
- **Batch job processing** for multiple jobs
|
||||
- **Job listing** with `jobs.list`
|
||||
|
||||
**Run with:**
|
||||
```bash
|
||||
cargo run --example job_api_examples
|
||||
```
|
||||
|
||||
### 2. `simple_job_workflow.rs` - Basic Workflow
|
||||
Simple example showing the basic job lifecycle:
|
||||
1. Create job with `jobs.create`
|
||||
2. Start job with `job.start`
|
||||
3. Monitor with `job.status`
|
||||
4. Get result with `job.result`
|
||||
|
||||
**Run with:**
|
||||
```bash
|
||||
cargo run --example simple_job_workflow
|
||||
```
|
||||
|
||||
### 3. `integration_test.rs` - Integration Tests
|
||||
Comprehensive integration tests validating:
|
||||
- Complete job lifecycle
|
||||
- Immediate job execution
|
||||
- Job listing functionality
|
||||
- Authentication error handling
|
||||
- Nonexistent job operations
|
||||
|
||||
**Run with:**
|
||||
```bash
|
||||
cargo test --test integration_test
|
||||
```
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Before running the examples, ensure:
|
||||
|
||||
1. **Redis is running:**
|
||||
```bash
|
||||
docker run -d -p 6379:6379 redis:alpine
|
||||
```
|
||||
|
||||
2. **Supervisor is running:**
|
||||
```bash
|
||||
./target/debug/supervisor --config examples/supervisor/config.toml
|
||||
```
|
||||
|
||||
3. **Runners are configured** in your config.toml:
|
||||
```toml
|
||||
[[actors]]
|
||||
id = "osis_runner_1"
|
||||
name = "osis_runner_1"
|
||||
binary_path = "/path/to/osis_runner"
|
||||
db_path = "/tmp/osis_db"
|
||||
redis_url = "redis://localhost:6379"
|
||||
process_manager = "simple"
|
||||
```
|
||||
|
||||
## API Convention Summary
|
||||
|
||||
The examples demonstrate the new job API convention:
|
||||
|
||||
### General Operations (`jobs.`)
|
||||
- `jobs.create` - Create a job without queuing it
|
||||
- `jobs.list` - List all job IDs in the system
|
||||
|
||||
### Specific Operations (`job.`)
|
||||
- `job.run` - Run a job immediately and return result
|
||||
- `job.start` - Start a previously created job
|
||||
- `job.status` - Get current job status (non-blocking)
|
||||
- `job.result` - Get job result (blocking until complete)
|
||||
|
||||
## Workflow Patterns
|
||||
|
||||
### Pattern 1: Fire-and-Forget
|
||||
```rust
|
||||
let result = client.job_run(secret, job).await?;
|
||||
match result {
|
||||
JobResult::Success { success } => println!("Output: {}", success),
|
||||
JobResult::Error { error } => println!("Error: {}", error),
|
||||
}
|
||||
```
|
||||
|
||||
### Pattern 2: Asynchronous Processing
|
||||
```rust
|
||||
// Create and start
|
||||
let job_id = client.jobs_create(secret, job).await?;
|
||||
client.job_start(secret, &job_id).await?;
|
||||
|
||||
// Monitor (non-blocking)
|
||||
loop {
|
||||
let status = client.job_status(&job_id).await?;
|
||||
if status.status == "completed" { break; }
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
|
||||
// Get result
|
||||
let result = client.job_result(&job_id).await?;
|
||||
```
|
||||
|
||||
### Pattern 3: Batch Processing
|
||||
```rust
|
||||
// Create all jobs
|
||||
let mut job_ids = Vec::new();
|
||||
for job_spec in job_specs {
|
||||
let job_id = client.jobs_create(secret, job_spec).await?;
|
||||
job_ids.push(job_id);
|
||||
}
|
||||
|
||||
// Start all jobs
|
||||
for job_id in &job_ids {
|
||||
client.job_start(secret, job_id).await?;
|
||||
}
|
||||
|
||||
// Collect results
|
||||
for job_id in &job_ids {
|
||||
let result = client.job_result(job_id).await?;
|
||||
// Process result...
|
||||
}
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
The examples demonstrate proper error handling for:
|
||||
- **Authentication errors** - Invalid secrets
|
||||
- **Job not found errors** - Nonexistent job IDs
|
||||
- **Connection errors** - Supervisor not available
|
||||
- **Execution errors** - Job failures
|
||||
|
||||
## Authentication
|
||||
|
||||
Examples use different secret types:
|
||||
- **Admin secrets**: Full system access
|
||||
- **User secrets**: Job operations only (used in examples)
|
||||
- **Register secrets**: Runner registration only
|
||||
|
||||
Configure secrets in your supervisor config:
|
||||
```toml
|
||||
admin_secrets = ["admin-secret-123"]
|
||||
user_secrets = ["user-secret-456"]
|
||||
register_secrets = ["register-secret-789"]
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
1. **Connection refused**
|
||||
- Ensure supervisor is running on localhost:3030
|
||||
- Check supervisor logs for errors
|
||||
|
||||
2. **Authentication failed**
|
||||
- Verify secret is configured in supervisor
|
||||
- Check secret type matches operation requirements
|
||||
|
||||
3. **Job execution failed**
|
||||
- Ensure runners are properly configured and running
|
||||
- Check runner logs for execution errors
|
||||
- Verify job payload is valid for the target runner
|
||||
|
||||
4. **Redis connection failed**
|
||||
- Ensure Redis is running on localhost:6379
|
||||
- Check Redis connectivity from supervisor
|
||||
|
||||
### Debug Mode
|
||||
|
||||
Run examples with debug logging:
|
||||
```bash
|
||||
RUST_LOG=debug cargo run --example job_api_examples
|
||||
```
|
||||
|
||||
This will show detailed API calls and responses for troubleshooting.
|
||||
@@ -17,7 +17,7 @@
|
||||
|
||||
use hero_supervisor_openrpc_client::{
|
||||
SupervisorClient, RunnerConfig, RunnerType, ProcessManagerType,
|
||||
JobBuilder, JobType, ClientError
|
||||
JobBuilder, JobType
|
||||
};
|
||||
use std::time::Duration;
|
||||
use escargot::CargoBuild;
|
||||
@@ -136,8 +136,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
.context_id("demo")
|
||||
.payload(payload)
|
||||
.job_type(JobType::OSIS)
|
||||
.runner_name("basic_example_actor")
|
||||
.timeout(Duration::from_secs(30))
|
||||
.runner("basic_example_actor")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
println!("📤 Queuing job '{}': {}", description, job.id);
|
||||
@@ -164,8 +164,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
.context_id("sync_demo")
|
||||
.payload(payload)
|
||||
.job_type(JobType::OSIS)
|
||||
.runner_name("basic_example_actor")
|
||||
.timeout(Duration::from_secs(30))
|
||||
.runner("basic_example_actor")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
println!("🚀 Executing '{}' with result verification...", description);
|
||||
|
||||
190
examples/integration_test.rs
Normal file
190
examples/integration_test.rs
Normal file
@@ -0,0 +1,190 @@
|
||||
//! Integration test for the new job API
|
||||
//!
|
||||
//! This test demonstrates the complete job lifecycle and validates
|
||||
//! that all new API methods work correctly together.
|
||||
|
||||
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_complete_job_lifecycle() -> Result<(), Box<dyn std::error::Error>> {
|
||||
// Skip test if supervisor is not running
|
||||
let client = match SupervisorClient::new("http://localhost:3030") {
|
||||
Ok(c) => c,
|
||||
Err(_) => {
|
||||
println!("Skipping integration test - supervisor not available");
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
// Test connection
|
||||
if client.discover().await.is_err() {
|
||||
println!("Skipping integration test - supervisor not responding");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let secret = "user-secret-456";
|
||||
|
||||
// Test 1: Create job
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("integration_test")
|
||||
.context_id("test_lifecycle")
|
||||
.payload("echo 'Integration test job'")
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
let job_id = client.jobs_create(secret, job).await?;
|
||||
assert!(!job_id.is_empty());
|
||||
|
||||
// Test 2: Start job
|
||||
client.job_start(secret, &job_id).await?;
|
||||
|
||||
// Test 3: Monitor status
|
||||
let mut attempts = 0;
|
||||
let max_attempts = 15; // 15 seconds max
|
||||
let mut final_status = String::new();
|
||||
|
||||
while attempts < max_attempts {
|
||||
let status = client.job_status(&job_id).await?;
|
||||
final_status = status.status.clone();
|
||||
|
||||
if final_status == "completed" || final_status == "failed" || final_status == "timeout" {
|
||||
break;
|
||||
}
|
||||
|
||||
attempts += 1;
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
|
||||
// Test 4: Get result
|
||||
let result = client.job_result(&job_id).await?;
|
||||
match result {
|
||||
JobResult::Success { success: _ } => {
|
||||
assert_eq!(final_status, "completed");
|
||||
},
|
||||
JobResult::Error { error: _ } => {
|
||||
assert!(final_status == "failed" || final_status == "timeout");
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_job_run_immediate() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let client = match SupervisorClient::new("http://localhost:3030") {
|
||||
Ok(c) => c,
|
||||
Err(_) => return Ok(()), // Skip if not available
|
||||
};
|
||||
|
||||
if client.discover().await.is_err() {
|
||||
return Ok(()); // Skip if not responding
|
||||
}
|
||||
|
||||
let secret = "user-secret-456";
|
||||
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("integration_test")
|
||||
.context_id("test_immediate")
|
||||
.payload("echo 'Immediate job test'")
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
// Test immediate execution
|
||||
let result = client.job_run(secret, job).await?;
|
||||
|
||||
// Should get either success or error, but not panic
|
||||
match result {
|
||||
JobResult::Success { success } => {
|
||||
assert!(!success.is_empty());
|
||||
},
|
||||
JobResult::Error { error } => {
|
||||
assert!(!error.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_jobs_list() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let client = match SupervisorClient::new("http://localhost:3030") {
|
||||
Ok(c) => c,
|
||||
Err(_) => return Ok(()), // Skip if not available
|
||||
};
|
||||
|
||||
if client.discover().await.is_err() {
|
||||
return Ok(()); // Skip if not responding
|
||||
}
|
||||
|
||||
// Test listing jobs
|
||||
let job_ids = client.jobs_list().await?;
|
||||
|
||||
// Should return a vector (might be empty)
|
||||
assert!(job_ids.len() >= 0);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_authentication_errors() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let client = match SupervisorClient::new("http://localhost:3030") {
|
||||
Ok(c) => c,
|
||||
Err(_) => return Ok(()), // Skip if not available
|
||||
};
|
||||
|
||||
if client.discover().await.is_err() {
|
||||
return Ok(()); // Skip if not responding
|
||||
}
|
||||
|
||||
let invalid_secret = "invalid-secret";
|
||||
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("integration_test")
|
||||
.context_id("test_auth")
|
||||
.payload("echo 'Auth test'")
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
// Test that invalid secret fails
|
||||
let result = client.jobs_create(invalid_secret, job.clone()).await;
|
||||
assert!(result.is_err());
|
||||
|
||||
let result = client.job_run(invalid_secret, job.clone()).await;
|
||||
assert!(result.is_err());
|
||||
|
||||
let result = client.job_start(invalid_secret, "fake-job-id").await;
|
||||
assert!(result.is_err());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_nonexistent_job_operations() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let client = match SupervisorClient::new("http://localhost:3030") {
|
||||
Ok(c) => c,
|
||||
Err(_) => return Ok(()), // Skip if not available
|
||||
};
|
||||
|
||||
if client.discover().await.is_err() {
|
||||
return Ok(()); // Skip if not responding
|
||||
}
|
||||
|
||||
let fake_job_id = "nonexistent-job-id";
|
||||
|
||||
// Test operations on nonexistent job
|
||||
let result = client.job_status(fake_job_id).await;
|
||||
assert!(result.is_err());
|
||||
|
||||
let result = client.job_result(fake_job_id).await;
|
||||
assert!(result.is_err());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
269
examples/job_api_examples.rs
Normal file
269
examples/job_api_examples.rs
Normal file
@@ -0,0 +1,269 @@
|
||||
//! Examples demonstrating the new job API workflows
|
||||
//!
|
||||
//! This example shows how to use the new job API methods:
|
||||
//! - jobs.create: Create a job without queuing
|
||||
//! - jobs.list: List all jobs
|
||||
//! - job.run: Run a job and get result immediately
|
||||
//! - job.start: Start a created job
|
||||
//! - job.status: Get job status (non-blocking)
|
||||
//! - job.result: Get job result (blocking)
|
||||
|
||||
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
// Initialize logging
|
||||
env_logger::init();
|
||||
|
||||
println!("🚀 Hero Supervisor Job API Examples");
|
||||
println!("===================================\n");
|
||||
|
||||
// Create client
|
||||
let client = SupervisorClient::new("http://localhost:3030")?;
|
||||
let secret = "user-secret-456"; // Use a user secret for job operations
|
||||
|
||||
// Test connection
|
||||
println!("📡 Testing connection...");
|
||||
match client.discover().await {
|
||||
Ok(_) => println!("✅ Connected to supervisor\n"),
|
||||
Err(e) => {
|
||||
println!("❌ Failed to connect: {}", e);
|
||||
println!("Make sure the supervisor is running with: ./supervisor --config examples/supervisor/config.toml\n");
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
// Example 1: Fire-and-forget job execution
|
||||
println!("🔥 Example 1: Fire-and-forget job execution");
|
||||
println!("--------------------------------------------");
|
||||
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("example_client")
|
||||
.context_id("fire_and_forget")
|
||||
.payload("echo 'Hello from fire-and-forget job!'")
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
println!("Running job immediately...");
|
||||
match client.job_run(secret, job).await {
|
||||
Ok(JobResult::Success { success }) => {
|
||||
println!("✅ Job completed successfully:");
|
||||
println!(" Output: {}", success);
|
||||
},
|
||||
Ok(JobResult::Error { error }) => {
|
||||
println!("❌ Job failed:");
|
||||
println!(" Error: {}", error);
|
||||
},
|
||||
Err(e) => {
|
||||
println!("❌ API call failed: {}", e);
|
||||
}
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 2: Asynchronous job processing
|
||||
println!("⏰ Example 2: Asynchronous job processing");
|
||||
println!("------------------------------------------");
|
||||
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("example_client")
|
||||
.context_id("async_processing")
|
||||
.payload("sleep 2 && echo 'Hello from async job!'")
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(60)
|
||||
.build()?;
|
||||
|
||||
// Step 1: Create the job
|
||||
println!("1. Creating job...");
|
||||
let job_id = match client.jobs_create(secret, job).await {
|
||||
Ok(id) => {
|
||||
println!("✅ Job created with ID: {}", id);
|
||||
id
|
||||
},
|
||||
Err(e) => {
|
||||
println!("❌ Failed to create job: {}", e);
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
// Step 2: Start the job
|
||||
println!("2. Starting job...");
|
||||
match client.job_start(secret, &job_id).await {
|
||||
Ok(_) => println!("✅ Job started"),
|
||||
Err(e) => {
|
||||
println!("❌ Failed to start job: {}", e);
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
// Step 3: Poll for completion (non-blocking)
|
||||
println!("3. Monitoring job progress...");
|
||||
let mut attempts = 0;
|
||||
let max_attempts = 30; // 30 seconds max
|
||||
|
||||
loop {
|
||||
attempts += 1;
|
||||
|
||||
match client.job_status(&job_id).await {
|
||||
Ok(status) => {
|
||||
println!(" Status: {} (attempt {})", status.status, attempts);
|
||||
|
||||
if status.status == "completed" || status.status == "failed" || status.status == "timeout" {
|
||||
break;
|
||||
}
|
||||
|
||||
if attempts >= max_attempts {
|
||||
println!(" ⏰ Timeout waiting for job completion");
|
||||
break;
|
||||
}
|
||||
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
},
|
||||
Err(e) => {
|
||||
println!(" ❌ Failed to get job status: {}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Step 4: Get the result
|
||||
println!("4. Getting job result...");
|
||||
match client.job_result(&job_id).await {
|
||||
Ok(JobResult::Success { success }) => {
|
||||
println!("✅ Job completed successfully:");
|
||||
println!(" Output: {}", success);
|
||||
},
|
||||
Ok(JobResult::Error { error }) => {
|
||||
println!("❌ Job failed:");
|
||||
println!(" Error: {}", error);
|
||||
},
|
||||
Err(e) => {
|
||||
println!("❌ Failed to get job result: {}", e);
|
||||
}
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 3: Batch job processing
|
||||
println!("📦 Example 3: Batch job processing");
|
||||
println!("-----------------------------------");
|
||||
|
||||
let job_specs = vec![
|
||||
("echo 'Batch job 1'", "batch_1"),
|
||||
("echo 'Batch job 2'", "batch_2"),
|
||||
("echo 'Batch job 3'", "batch_3"),
|
||||
];
|
||||
|
||||
let mut job_ids = Vec::new();
|
||||
|
||||
// Create all jobs
|
||||
println!("Creating batch jobs...");
|
||||
for (i, (payload, context)) in job_specs.iter().enumerate() {
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("example_client")
|
||||
.context_id(context)
|
||||
.payload(payload)
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(30)
|
||||
.build()?;
|
||||
|
||||
match client.jobs_create(secret, job).await {
|
||||
Ok(job_id) => {
|
||||
println!("✅ Created job {}: {}", i + 1, job_id);
|
||||
job_ids.push(job_id);
|
||||
},
|
||||
Err(e) => {
|
||||
println!("❌ Failed to create job {}: {}", i + 1, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Start all jobs
|
||||
println!("Starting all batch jobs...");
|
||||
for (i, job_id) in job_ids.iter().enumerate() {
|
||||
match client.job_start(secret, job_id).await {
|
||||
Ok(_) => println!("✅ Started job {}", i + 1),
|
||||
Err(e) => println!("❌ Failed to start job {}: {}", i + 1, e),
|
||||
}
|
||||
}
|
||||
|
||||
// Collect results
|
||||
println!("Collecting results...");
|
||||
for (i, job_id) in job_ids.iter().enumerate() {
|
||||
match client.job_result(job_id).await {
|
||||
Ok(JobResult::Success { success }) => {
|
||||
println!("✅ Job {} result: {}", i + 1, success);
|
||||
},
|
||||
Ok(JobResult::Error { error }) => {
|
||||
println!("❌ Job {} failed: {}", i + 1, error);
|
||||
},
|
||||
Err(e) => {
|
||||
println!("❌ Failed to get result for job {}: {}", i + 1, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
println!();
|
||||
|
||||
// Example 4: List all jobs
|
||||
println!("📋 Example 4: Listing all jobs");
|
||||
println!("-------------------------------");
|
||||
|
||||
match client.jobs_list().await {
|
||||
Ok(job_ids) => {
|
||||
println!("✅ Found {} jobs in the system:", job_ids.len());
|
||||
for (i, job_id) in job_ids.iter().take(10).enumerate() {
|
||||
println!(" {}. {}", i + 1, job_id);
|
||||
}
|
||||
if job_ids.len() > 10 {
|
||||
println!(" ... and {} more", job_ids.len() - 10);
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
println!("❌ Failed to list jobs: {}", e);
|
||||
}
|
||||
}
|
||||
println!();
|
||||
|
||||
println!("🎉 All examples completed!");
|
||||
println!("\nAPI Convention Summary:");
|
||||
println!("- jobs.create: Create job without queuing");
|
||||
println!("- jobs.list: List all job IDs");
|
||||
println!("- job.run: Run job and return result immediately");
|
||||
println!("- job.start: Start a created job");
|
||||
println!("- job.status: Get job status (non-blocking)");
|
||||
println!("- job.result: Get job result (blocking)");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_job_builder() {
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("test")
|
||||
.context_id("test")
|
||||
.payload("echo 'test'")
|
||||
.executor("osis")
|
||||
.runner("test_runner")
|
||||
.build();
|
||||
|
||||
assert!(job.is_ok());
|
||||
let job = job.unwrap();
|
||||
assert_eq!(job.caller_id, "test");
|
||||
assert_eq!(job.context_id, "test");
|
||||
assert_eq!(job.payload, "echo 'test'");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_client_creation() {
|
||||
let client = SupervisorClient::new("http://localhost:3030");
|
||||
assert!(client.is_ok());
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,7 @@ use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
use redis::AsyncCommands;
|
||||
use hero_supervisor::{
|
||||
job::{Job, JobStatus, JobType, keys},
|
||||
Job, JobStatus, JobError, client::{Client, ClientBuilder}
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -43,6 +43,14 @@ impl MockRunnerConfig {
|
||||
return Err("Missing value for --actor-id".into());
|
||||
}
|
||||
}
|
||||
"--db-path" => {
|
||||
if i + 1 < args.len() {
|
||||
db_path = Some(args[i + 1].clone());
|
||||
i += 2;
|
||||
} else {
|
||||
return Err("Missing value for --db-path".into());
|
||||
}
|
||||
}
|
||||
"--redis-url" => {
|
||||
if i + 1 < args.len() {
|
||||
redis_url = Some(args[i + 1].clone());
|
||||
@@ -65,16 +73,19 @@ impl MockRunnerConfig {
|
||||
|
||||
pub struct MockRunner {
|
||||
config: MockRunnerConfig,
|
||||
redis_client: redis::Client,
|
||||
client: Client,
|
||||
}
|
||||
|
||||
impl MockRunner {
|
||||
pub fn new(config: MockRunnerConfig) -> Result<Self, Box<dyn std::error::Error>> {
|
||||
let redis_client = redis::Client::open(config.redis_url.clone())?;
|
||||
pub async fn new(config: MockRunnerConfig) -> Result<Self, Box<dyn std::error::Error>> {
|
||||
let client = ClientBuilder::new()
|
||||
.redis_url(&config.redis_url)
|
||||
.build()
|
||||
.await?;
|
||||
|
||||
Ok(MockRunner {
|
||||
config,
|
||||
redis_client,
|
||||
client,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -83,53 +94,52 @@ impl MockRunner {
|
||||
println!("📂 DB Path: {}", self.config.db_path);
|
||||
println!("🔗 Redis URL: {}", self.config.redis_url);
|
||||
|
||||
let mut conn = self.redis_client.get_multiplexed_async_connection().await?;
|
||||
|
||||
// Use the proper Hero job queue key for this actor instance
|
||||
// Format: hero:q:work:type:{job_type}:group:{group}:inst:{instance}
|
||||
let work_queue_key = keys::work_instance(&JobType::OSIS, "default", &self.config.actor_id);
|
||||
let work_queue_key = format!("hero:q:work:type:osis:group:default:inst:{}", self.config.actor_id);
|
||||
|
||||
println!("👂 Listening for jobs on queue: {}", work_queue_key);
|
||||
|
||||
loop {
|
||||
// Try to pop a job ID from the work queue using the Hero protocol
|
||||
let result: redis::RedisResult<Option<String>> = conn.lpop(&work_queue_key, None).await;
|
||||
let job_id = self.client.get_job_id(&work_queue_key).await?;
|
||||
|
||||
match result {
|
||||
Ok(Some(job_id)) => {
|
||||
match job_id {
|
||||
Some(job_id) => {
|
||||
println!("📨 Received job ID: {}", job_id);
|
||||
if let Err(e) = self.process_job(&mut conn, &job_id).await {
|
||||
if let Err(e) = self.process_job(&job_id).await {
|
||||
eprintln!("❌ Error processing job {}: {}", job_id, e);
|
||||
// Mark job as error
|
||||
if let Err(e2) = Job::set_error(&mut conn, &job_id, &format!("Processing error: {}", e)).await {
|
||||
if let Err(e2) = self.client.set_job_status(&job_id, JobStatus::Error).await {
|
||||
eprintln!("❌ Failed to set job error status: {}", e2);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
None => {
|
||||
// No jobs available, wait a bit
|
||||
sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("❌ Redis error: {}", e);
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn process_job(&self, conn: &mut redis::aio::MultiplexedConnection, job_id: &str) -> Result<(), Box<dyn std::error::Error>> {
|
||||
async fn process_job(&self, job_id: &str) -> Result<(), JobError> {
|
||||
// Load the job from Redis using the Hero job system
|
||||
let job = Job::load_from_redis(conn, job_id).await?;
|
||||
let job = self.client.get_job(job_id).await?;
|
||||
|
||||
println!("📝 Processing job: {}", job.id);
|
||||
println!("📝 Caller: {}", job.caller_id);
|
||||
println!("📝 Context: {}", job.context_id);
|
||||
println!("📝 Payload: {}", job.payload);
|
||||
println!("📝 Job Type: {:?}", job.job_type);
|
||||
self.process_job_internal(&self.client, job_id, &job).await
|
||||
}
|
||||
|
||||
async fn process_job_internal(
|
||||
&self,
|
||||
client: &Client,
|
||||
job_id: &str,
|
||||
job: &Job,
|
||||
) -> Result<(), JobError> {
|
||||
println!("🔄 Processing job {} with payload: {}", job_id, job.payload);
|
||||
|
||||
// Mark job as started
|
||||
Job::update_status(conn, job_id, JobStatus::Started).await?;
|
||||
client.set_job_status(job_id, JobStatus::Started).await?;
|
||||
println!("🚀 Job {} marked as Started", job_id);
|
||||
|
||||
// Simulate processing time
|
||||
@@ -140,10 +150,8 @@ impl MockRunner {
|
||||
println!("📤 Output: {}", output);
|
||||
|
||||
// Set the job result
|
||||
Job::set_result(conn, job_id, &output).await?;
|
||||
client.set_result(job_id, &output).await?;
|
||||
|
||||
// Mark job as finished
|
||||
Job::update_status(conn, job_id, JobStatus::Finished).await?;
|
||||
println!("✅ Job {} completed successfully", job_id);
|
||||
|
||||
Ok(())
|
||||
@@ -156,7 +164,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let config = MockRunnerConfig::from_args()?;
|
||||
|
||||
// Create and run the mock runner
|
||||
let runner = MockRunner::new(config)?;
|
||||
let runner = MockRunner::new(config).await?;
|
||||
runner.run().await?;
|
||||
|
||||
Ok(())
|
||||
|
||||
64
examples/simple_job_workflow.rs
Normal file
64
examples/simple_job_workflow.rs
Normal file
@@ -0,0 +1,64 @@
|
||||
//! Simple job workflow example
|
||||
//!
|
||||
//! This example demonstrates the basic job lifecycle using the new API:
|
||||
//! 1. Create a job
|
||||
//! 2. Start the job
|
||||
//! 3. Monitor its progress
|
||||
//! 4. Get the result
|
||||
|
||||
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("Simple Job Workflow Example");
|
||||
println!("============================\n");
|
||||
|
||||
// Create client
|
||||
let client = SupervisorClient::new("http://localhost:3030")?;
|
||||
let secret = "user-secret-456";
|
||||
|
||||
// Create a simple job
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("simple_example")
|
||||
.context_id("demo")
|
||||
.payload("echo 'Hello from Hero Supervisor!' && sleep 3 && echo 'Job completed!'")
|
||||
.executor("osis")
|
||||
.runner("osis_runner_1")
|
||||
.timeout(60)
|
||||
.env_var("EXAMPLE_VAR", "example_value")
|
||||
.build()?;
|
||||
|
||||
println!("📝 Creating job...");
|
||||
let job_id = client.jobs_create(secret, job).await?;
|
||||
println!("✅ Job created: {}\n", job_id);
|
||||
|
||||
println!("🚀 Starting job...");
|
||||
client.job_start(secret, &job_id).await?;
|
||||
println!("✅ Job started\n");
|
||||
|
||||
println!("👀 Monitoring job progress...");
|
||||
loop {
|
||||
let status = client.job_status(&job_id).await?;
|
||||
println!(" Status: {}", status.status);
|
||||
|
||||
if status.status == "completed" || status.status == "failed" {
|
||||
break;
|
||||
}
|
||||
|
||||
sleep(Duration::from_secs(2)).await;
|
||||
}
|
||||
|
||||
println!("\n📋 Getting job result...");
|
||||
match client.job_result(&job_id).await? {
|
||||
JobResult::Success { success } => {
|
||||
println!("✅ Success: {}", success);
|
||||
},
|
||||
JobResult::Error { error } => {
|
||||
println!("❌ Error: {}", error);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -69,7 +69,7 @@ Once running, the supervisor will:
|
||||
1. Load the configuration from `config.toml`
|
||||
2. Initialize and start all configured actors
|
||||
3. Listen for jobs on the Redis queue (`hero:supervisor:jobs`)
|
||||
4. Dispatch jobs to appropriate actors based on the `runner_name` field
|
||||
4. Dispatch jobs to appropriate actors based on the `runner` field
|
||||
5. Monitor actor health and status
|
||||
|
||||
## Testing
|
||||
@@ -78,7 +78,7 @@ You can test the supervisor by dispatching jobs to the Redis queue:
|
||||
|
||||
```bash
|
||||
# Using redis-cli to add a test job
|
||||
redis-cli LPUSH "hero:supervisor:jobs" '{"id":"test-123","runner_name":"sal_actor_1","script":"print(\"Hello from SAL actor!\")"}'
|
||||
redis-cli LPUSH "hero:supervisor:jobs" '{"id":"test-123","runner":"sal_actor_1","script":"print(\"Hello from SAL actor!\")"}'
|
||||
```
|
||||
|
||||
## Stopping
|
||||
|
||||
@@ -1,59 +0,0 @@
|
||||
//! Test to verify OpenRPC method registration
|
||||
|
||||
use hero_supervisor_openrpc_client::SupervisorClient;
|
||||
use tokio::time::{sleep, Duration};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("🔍 Testing OpenRPC method registration");
|
||||
|
||||
// Start a local supervisor with OpenRPC (assume it's running)
|
||||
println!("📡 Connecting to OpenRPC server...");
|
||||
let client = SupervisorClient::new("http://127.0.0.1:3030").await?;
|
||||
|
||||
// Test basic methods first
|
||||
println!("🧪 Testing basic methods...");
|
||||
|
||||
// Test list_runners (should work)
|
||||
match client.list_runners().await {
|
||||
Ok(runners) => println!("✅ list_runners works: {:?}", runners),
|
||||
Err(e) => println!("❌ list_runners failed: {}", e),
|
||||
}
|
||||
|
||||
// Test get_all_runner_status (might have serialization issues)
|
||||
match client.get_all_runner_status().await {
|
||||
Ok(statuses) => println!("✅ get_all_runner_status works: {} runners", statuses.len()),
|
||||
Err(e) => println!("❌ get_all_runner_status failed: {}", e),
|
||||
}
|
||||
|
||||
// Test the new queue_and_wait method
|
||||
println!("🎯 Testing queue_and_wait method...");
|
||||
|
||||
// Create a simple test job
|
||||
use hero_supervisor::job::{JobBuilder, JobType};
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("test_client")
|
||||
.context_id("method_test")
|
||||
.payload("print('Testing queue_and_wait method registration');")
|
||||
.job_type(JobType::OSIS)
|
||||
.runner_name("osis_actor") // Use existing runner
|
||||
.timeout(Duration::from_secs(10))
|
||||
.build()?;
|
||||
|
||||
match client.queue_and_wait("osis_actor", job, 10).await {
|
||||
Ok(Some(result)) => println!("✅ queue_and_wait works! Result: {}", result),
|
||||
Ok(None) => println!("⏰ queue_and_wait timed out"),
|
||||
Err(e) => {
|
||||
println!("❌ queue_and_wait failed: {}", e);
|
||||
|
||||
// Check if it's a MethodNotFound error
|
||||
if e.to_string().contains("Method not found") {
|
||||
println!("🔍 Method not found - this suggests trait registration issue");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
println!("🏁 OpenRPC method test completed");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,70 +0,0 @@
|
||||
//! Simple test for the queue_and_wait functionality
|
||||
|
||||
use hero_supervisor::{
|
||||
supervisor::{Supervisor, ProcessManagerType},
|
||||
runner::RunnerConfig,
|
||||
job::{JobBuilder, JobType},
|
||||
};
|
||||
use std::time::Duration;
|
||||
use std::path::PathBuf;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
println!("🧪 Testing queue_and_wait functionality directly");
|
||||
|
||||
// Create supervisor
|
||||
let mut supervisor = Supervisor::new();
|
||||
|
||||
// Create a runner config
|
||||
let config = RunnerConfig::new(
|
||||
"test_actor".to_string(),
|
||||
hero_supervisor::runner::RunnerType::OSISRunner,
|
||||
PathBuf::from("./target/debug/examples/mock_runner"),
|
||||
"/tmp/test_db".to_string(),
|
||||
"redis://localhost:6379".to_string(),
|
||||
);
|
||||
|
||||
// Add runner
|
||||
println!("➕ Adding test runner...");
|
||||
supervisor.add_runner(config, ProcessManagerType::Simple).await?;
|
||||
|
||||
// Start runner
|
||||
println!("▶️ Starting test runner...");
|
||||
supervisor.start_runner("test_actor").await?;
|
||||
|
||||
// Create a test job
|
||||
let job = JobBuilder::new()
|
||||
.caller_id("test_client")
|
||||
.context_id("direct_test")
|
||||
.payload("print('Direct queue_and_wait test!');")
|
||||
.job_type(JobType::OSIS)
|
||||
.runner_name("test_actor")
|
||||
.timeout(Duration::from_secs(10))
|
||||
.build()?;
|
||||
|
||||
println!("🚀 Testing queue_and_wait directly...");
|
||||
println!("📋 Job ID: {}", job.id);
|
||||
|
||||
// Test queue_and_wait directly
|
||||
match supervisor.queue_and_wait("test_actor", job, 10).await {
|
||||
Ok(Some(result)) => {
|
||||
println!("✅ queue_and_wait succeeded!");
|
||||
println!("📤 Result: {}", result);
|
||||
}
|
||||
Ok(None) => {
|
||||
println!("⏰ queue_and_wait timed out");
|
||||
}
|
||||
Err(e) => {
|
||||
println!("❌ queue_and_wait failed: {}", e);
|
||||
}
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
println!("🧹 Cleaning up...");
|
||||
supervisor.stop_runner("test_actor", false).await?;
|
||||
supervisor.remove_runner("test_actor").await?;
|
||||
|
||||
println!("✅ Direct test completed!");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
//! Test program for register_runner functionality with secret authentication
|
||||
|
||||
use hero_supervisor::{SupervisorApp};
|
||||
use log::info;
|
||||
use tokio;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
env_logger::init();
|
||||
|
||||
info!("Starting supervisor with test secrets...");
|
||||
|
||||
// Create supervisor app with test secrets
|
||||
let mut app = SupervisorApp::builder()
|
||||
.redis_url("redis://localhost:6379")
|
||||
.db_path("/tmp/hero_test_db")
|
||||
.queue_key("hero:test_queue")
|
||||
.admin_secret("admin123")
|
||||
.register_secret("register456")
|
||||
.user_secret("user789")
|
||||
.build()
|
||||
.await?;
|
||||
|
||||
info!("Supervisor configured with secrets:");
|
||||
info!(" Admin secrets: {:?}", app.supervisor.admin_secrets());
|
||||
info!(" Register secrets: {:?}", app.supervisor.register_secrets());
|
||||
info!(" User secrets: {:?}", app.supervisor.user_secrets());
|
||||
|
||||
// Start OpenRPC server
|
||||
let supervisor_arc = std::sync::Arc::new(tokio::sync::Mutex::new(app.supervisor.clone()));
|
||||
|
||||
info!("Starting OpenRPC server...");
|
||||
hero_supervisor::openrpc::start_openrpc_servers(supervisor_arc).await?;
|
||||
|
||||
info!("Supervisor is running with OpenRPC server on http://127.0.0.1:3030");
|
||||
info!("Test secrets configured:");
|
||||
info!(" Admin secret: admin123");
|
||||
info!(" Register secret: register456");
|
||||
info!(" User secret: user789");
|
||||
info!("Press Ctrl+C to stop...");
|
||||
|
||||
// Keep running
|
||||
loop {
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user