Async Job Runner - PAGI Demo

A real-time job queue dashboard demonstrating PAGI's async capabilities with HTTP, SSE, and WebSocket protocols working together.

This example names no event loop. The background worker tick, the WebSocket keepalive ping and the per-job delays all pace themselves with Future::IO->sleep, which dispatches to whichever implementation the server bound at startup. Nothing here constructs an IO::Async::Loop or holds a timer object, so the same application runs unchanged under any conforming PAGI server.

The pattern is worth copying: where you would reach for a loop timer, an async sub that sleeps and re-checks its own flag does the same job without binding the application to one implementation.

$worker_tick = (async sub {
    while ($is_running) {
        _check_queue();
        await Future::IO->sleep(0.1);
    }
})->();

Stopping is a flag plus a cancel, so shutdown does not wait out the interval. See PAGI-Tools/examples/process-streaming for the same principle applied to reading a subprocess.

Running

From the PAGI root directory, using pagi-server from the PAGI-Server distribution. This example keeps its modules under its own lib/, so put that on the module search path:

PERL5LIB=examples/11-job-runner/lib pagi-server \
    --app examples/11-job-runner/app.pl --port 5001

Then open http://localhost:5001 in your browser.

Features

• Real-time job queue - Create countdown jobs and watch them execute
• Live progress streaming - SSE updates show second-by-second progress
• WebSocket dashboard - Queue-wide updates pushed to all connected clients
• Concurrent execution - Worker processes up to 3 jobs simultaneously

Architecture

┌─────────────────────────────────────────────────────────────┐
│                      Browser (app.js)                       │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │  WebSocket   │  │     SSE      │  │   HTTP (REST)    │  │
│  │ Queue Events │  │ Job Progress │  │   Static Files   │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬─────────┘  │
└─────────┼─────────────────┼───────────────────┼────────────┘
          │                 │                   │
          ▼                 ▼                   ▼
┌─────────────────────────────────────────────────────────────┐
│                     PAGI Server                             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ WebSocket.pm │  │   SSE.pm     │  │     HTTP.pm      │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬─────────┘  │
│         │                 │                   │            │
│         └────────────┬────┴───────────────────┘            │
│                      ▼                                     │
│              ┌───────────────┐                             │
│              │   Queue.pm    │ ◄── Job State Management    │
│              └───────┬───────┘                             │
│                      │                                     │
│              ┌───────▼───────┐                             │
│              │  Worker.pm    │ ◄── Async Job Execution     │
│              └───────┬───────┘                             │
│                      │                                     │
│              ┌───────▼───────┐                             │
│              │   Jobs.pm     │ ◄── Job Type Definitions    │
│              └───────────────┘                             │
└─────────────────────────────────────────────────────────────┘

API Endpoints

| Method | Endpoint | Description | |--------|----------|-------------| | GET | /api/stats | Queue and worker statistics | | GET | /api/job-types | Available job types | | GET | /api/jobs | List all jobs | | POST | /api/jobs | Create a new job | | GET | /api/jobs/:id | Get job details | | DELETE | /api/jobs/:id | Cancel a job | | GET | /api/jobs/:id/progress | SSE progress stream |

WebSocket Protocol

Connect to /ws/queue for real-time updates.

Server -> Client:

• queue_state - Full state on connect
• job_created - New job added
• job_started - Job began executing
• job_progress - Progress update
• job_completed - Job finished successfully
• job_failed - Job failed with error
• job_cancelled - Job was cancelled

Client -> Server:

• create_job - Create new job { job_type, params }
• cancel_job - Cancel job { job_id }
• clear_completed - Remove finished jobs
• get_state - Request full state

Job Types

• countdown - Counts down from N seconds with progress updates
• prime - Finds all prime numbers up to N (with progress)
• fibonacci - Calculates Fibonacci sequence up to N terms
• echo - Echoes a message back after a configurable delay

Testing with curl

# Create a 5-second countdown job
curl -X POST http://localhost:5001/api/jobs \
    -H "Content-Type: application/json" \
    -d '{"job_type":"countdown","params":{"seconds":5}}'

# Watch job progress (SSE)
curl -N http://localhost:5001/api/jobs/1/progress \
    -H "Accept: text/event-stream"

# Get queue stats
curl http://localhost:5001/api/stats