carno-php / process

src/Chips/Forking.php

<?php
/**
 * Forking daemon
 * User: moyo
 * Date: 28/12/2017
 * Time: 11:17 AM
 */

namespace Carno\Process\Chips;

use Carno\Process\Master;
use Carno\Process\Piping;
use Carno\Process\Program;
use Carno\Process\Progress;
use Carno\Promise\Promise;
use Swoole\Process as SWProcess;

The type Swoole\Process was not found. Maybe you did not declare it correctly or list all dependencies?

use Swoole\Server as SWServer;

The type Swoole\Server was not found. Maybe you did not declare it correctly or list all dependencies?

use Closure;
use Throwable;

trait Forking
{
    /**
     * @var Piping
     */
    private $forked = null;

    /**
     * @var bool
     */
    private $started = false;

    /**
     * @var Closure[]
     */
    private $events = [];

    /**
     * @var SWProcess
     */
    private $process = null;

    /**
     * @var bool
     */
    private $exiting = false;

    /**
     * @param int $event
     * @param Closure ...$actions
     */
    final public function on(int $event, Closure ...$actions) : void
    {
        $this->events[$event] = $actions;
    }

    /**
     * @return SWProcess
     */
    final public function process() : ?SWProcess
    {
        return $this->process;
    }

    /**
     * @return bool
     */
    final public function started() : bool
    {
        return $this->started;
    }

    /**
     * startup new process
     * @param object $server
     */
    final public function startup(object $server = null) : void
    {
        $this->process = new SWProcess(function (SWProcess $process) {
            // set process name
            if ($this->name) {
                @$process->name(sprintf('[process] %s', $this->name));

It seems like you do not handle an error condition for name(). This can introduce security issues, and is generally not recommended.

            }

            // progress notify
            Progress::started(getmypid(), $this->name);

            // start pipe reading
            $this->forked->reading();

            // trigger system startup
            $this->forked->bootstrap();

            // trigger event::program::started
            $this->action(Program::STARTED);

            // trigger user starting
            $this->starting();

It seems like starting() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        });

        // running
        $server instanceof SWServer ? $server->addProcess($this->process) : Master::watch($this->process->start());

        // flag
        $this->started = true;
    }

    /**
     * shutdown process
     * @param int $sig
     */
    final public function shutdown(int $sig = 0) : void
    {
        // check exiting state
        if ($this->exiting) {
            return;
        }

        $this->exiting = true;

        // be wait
        $wait = Promise::deferred();

        // user-land stopping
        $this->stopping($wait);

It seems like stopping() must be provided by classes using this trait. How about adding it as abstract method to this trait?

        // trigger event::program::stopped
        $this->action(Program::STOPPED);

        // waiting
        $wait->then(function () use ($sig) {
            $this->exited($sig);
        }, function (Throwable $e) use ($sig) {
            $this->exited($sig, $e);
        });
    }

    /**
     * @param int $sig
     * @param Throwable $e
     */
    final private function exited(int $sig = 0, Throwable $e = null) : void
    {
        Progress::exited(getmypid(), $sig, $e ? 1 : 0, $this->name);

        $this->process->exit();
    }

    /**
     * @param int $event
     */
    final private function action(int $event) : void
    {
        foreach ($this->events[$event] ?? [] as $program) {
            $program($this->forked);
        }
    }
}