PHP code example of smwks / superprocess

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

smwks / superprocess example snippets

use SMWks\SuperProcess\Child;
use SMWks\SuperProcess\CreateReason;
use SMWks\SuperProcess\ExitReason;
use SMWks\SuperProcess\SuperProcess;

$sp = new SuperProcess;
$sp->command('php artisan inspire:loop')
   ->scaleLimits(min: 2, max: 5)
   ->onChildCreate(fn (Child $c, CreateReason $r) => printf("[master] spawned %d\n", $c->pid))
   ->onChildExit(fn (Child $c, ExitReason $r)    => printf("[master] %d exited\n",   $c->pid))
   ->onChildOutput(fn (Child $c, string $data)   => print $data)
   ->run(); // blocks until SIGTERM is received

use SMWks\SuperProcess\Child;
use SMWks\SuperProcess\SuperProcess;

// The closure is forked by the master — no separate file needed.
$sp = new SuperProcess;
$sp->closure(function (mixed $socket): void {
        for ($i = 1; $i <= 5; $i++) {
            fwrite($socket, json_encode(['step' => $i]) . "\n");
            sleep(1);
        }
    })
    ->scaleLimits(min: 3, max: 3)
    ->onChildMessage(fn (Child $c, mixed $msg) => printf("[%d] step %d\n", $c->pid, $msg['step']))
    ->run();

use SMWks\SuperProcess\SuperProcess;

// Every 5 seconds the master checks the queue depth and adjusts the pool.
$sp = new SuperProcess;
$sp->command('php artisan queue:work')
   ->scaleLimits(min: 1, max: 20)
   ->heartbeat(5, function (SuperProcess $sp): void {
       $depth = (int) DB::scalar('SELECT COUNT(*) FROM jobs WHERE queue = ?', ['default']);

       if ($depth > 100) { $sp->scaleUp();   } // no-op if already at max
       if ($depth < 10)  { $sp->scaleDown(); } // no-op if already at min
   })
   ->run();

$sp->closure(function (mixed $socket): void {
    for ($i = 1; $i <= 5; $i++) {
        fwrite($socket, json_encode(['progress' => $i * 20]) . "\n");
        sleep(1);
    }
});

$sp->command('php artisan queue:work')
   ->scaleLimits(min: 1, max: 20)
   ->heartbeat(5, function (SuperProcess $sp): void {
       $depth = (int) DB::scalar('SELECT COUNT(*) FROM jobs WHERE queue = ?', ['default']);

       if ($depth > 100) { $sp->scaleUp();   } // no-op if already at max
       if ($depth < 10)  { $sp->scaleDown(); } // no-op if already at min
   })
   ->run();

$sp->onShutdown(function (SuperProcess $sp): void {
    echo "Shutting down — waiting for workers to finish current jobs\n";
})
->run();

// Set the command to run in each child (mutually exclusive with closure())
->command(string $command): static

// Set a PHP closure to run in each child (mutually exclusive with command())
->closure(Closure $fn): static   // fn(resource $socket): void

// Set the min/max number of running children (default: 1, 1)
->scaleLimits(int $min, int $max): static

// Register a periodic master heartbeat
->heartbeat(int $intervalSeconds, Closure $fn): static  // fn(SuperProcess $self): void

// Called when a child is spawned
->onChildCreate(Closure $fn): static   // fn(Child $child, CreateReason $reason): void

// Called when a child exits
->onChildExit(Closure $fn): static     // fn(Child $child, ExitReason $reason): void

// Called when SIGUSR1 or SIGUSR2 is received by the master
->onChildSignal(Closure $fn): static   // fn(Child $child, int $signal): void

// Called for each JSON message received on the child's IPC channel
->onChildMessage(Closure $fn): static  // fn(Child $child, mixed $message): void

// Called with raw stdout/stderr data from a command child
->onChildOutput(Closure $fn): static   // fn(Child $child, string $data): void

// Called once on shutdown (SIGTERM or SIGINT), before children are signalled
->onShutdown(Closure $fn): static      // fn(SuperProcess $self): void

// Write to a running child's stdin
->sendChildInput(int $pid, string $data): void

// Send any POSIX signal to a PID (use ProcessSignal constants)
->signal(string|int $pid, ProcessSignal $signal): void

// Spawn one more child (if below max)
->scaleUp(): static

// Terminate one child (if above min)
->scaleDown(): static

// Start the blocking event loop
->run(): void

// Why a child was created
CreateReason::Initial       // first spawn on run()
CreateReason::Replacement   // auto-restarted after exit
CreateReason::ScaleUp       // spawned by scaleUp()

// Why a child exited
ExitReason::Normal          // exited via exit() / end of script
ExitReason::Signal          // terminated by a signal (SIGTERM etc.)
ExitReason::Killed          // force-killed with SIGKILL
ExitReason::Unknown         // status could not be determined

// Signal shortcuts (values map to POSIX signal numbers)
ProcessSignal::Stop         // SIGTERM — graceful stop
ProcessSignal::Kill         // SIGKILL — force kill
ProcessSignal::Reload       // SIGHUP  — reload
ProcessSignal::Usr1         // SIGUSR1
ProcessSignal::Usr2         // SIGUSR2

$child->pid            // int  — process ID
$child->createReason   // CreateReason
$child->running        // bool — false once the process has exited
$child->exitCode       // int  — exit code (populated after exit)
$child->exitReason     // ExitReason (populated after exit)

// child-worker.php
$ipc = fopen('php://fd/3', 'w');

fwrite($ipc, json_encode(['type' => 'started', 'pid' => getmypid()]) . "\n");

// ... do work ...

fwrite($ipc, json_encode(['type' => 'done', 'items_processed' => 1234]) . "\n");
fclose($ipc);

// supervisor
$sp->command('php child-worker.php')
   ->onChildMessage(function (Child $child, mixed $msg): void {
       echo "[{$child->pid}] {$msg['type']}\n";
   });