Comodojo\Daemon\Worker\Manager

Complexity

Total Complexity

32

Size/Duplication

Total Lines	253
Duplicated Lines	0 %

Test Coverage

Coverage

0%

Importance

Changes

0

12 Methods

Rating	Name	Duplication	Size	Complexity
A	get()	0	9	3
A	install()	0	22	2
A	isInstalled()	0	3	1
A	__construct()	0	9	1
A	setPid()	0	9	2
A	start()	0	58	5
A	running()	0	3	1
A	status()	0	15	3
B	stop()	0	27	7
A	pause()	0	11	3
A	resume()	0	11	3
A	getStatus()	0	3	1

<?php namespace Comodojo\Daemon\Worker;

use \Comodojo\Daemon\Daemon;
use \Comodojo\Daemon\Utils\ProcessTools;
use \Comodojo\Daemon\Utils\PosixSignals;
use \Comodojo\Foundation\Events\EventsTrait;
use \Comodojo\Foundation\Logging\LoggerTrait;
use \Comodojo\Foundation\Events\Manager as EventsManager;
use \Comodojo\Foundation\DataAccess\IteratorTrait;
use \Comodojo\Foundation\DataAccess\CountableTrait;
use \Psr\Log\LoggerInterface;
use \Iterator;
use \Countable;
use \Exception;

/**
 * @package     Comodojo Daemon
 * @author      Marco Giovinazzi <marco.giovinazzi@comodojo.org>
 * @license     MIT
 *
 * LICENSE:
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

 class Manager implements Iterator, Countable {

    use IteratorTrait;
    use CountableTrait;
    use EventsTrait;
    use LoggerTrait;

    /**
     * The stack :)
     *
     * @var array
     */
    private $data = [];

    /**
     * Pointer to parent daemon
     *
     * @var Daemon
     */
    private $daemon;

    /**
     * Manager constructor
     *
     * @param LoggerInterface $logger;
     * @param EventsManager $events;
     * @param Daemon $daemon;
     */
    public function __construct(
        LoggerInterface $logger,
        EventsManager $events,
        Daemon $daemon
    ) {

        $this->logger = $logger;
        $this->events = $events;
        $this->daemon = $daemon;

    }

    /**
     * Install a worker into the stack
     *
     * @param WorkerInterface $worker
     * @param int $looptime
     * @param bool $forever
     * @return Manager
     */
    public function install(
        WorkerInterface $worker,
        $looptime = 1,
        $forever = false
    ) {

        $name = $worker->getName();

        if ( $this->isInstalled($name) ) {
            throw new Exception("Worker already installed");
        }

        $w = Worker::create()
            ->setInstance($worker)
            ->setLooptime($looptime)
            ->setForever($forever)
            ->setInputChannel(new SharedMemory((int) '1'.hexdec($worker->getId())))
            ->setOutputChannel(new SharedMemory((int) '2'.hexdec($worker->getId())));

        $this->data[$name] = $w;

        return $this;

    }

    public function setPid($name, $pid) {

        if ( !$this->isInstalled($name) ) {
            throw new Exception("Worker not installed");
        }

        $this->data[$name]->setPid($pid);

        return $this;

    }

    public function get($name = null) {

        if ( is_null($name) ) return $this->data;

        if ( !$this->isInstalled($name) ) {
            throw new Exception("Worker not installed");
        }

        return $this->data[$name];

    }

    public function isInstalled($name) {

        return array_key_exists($name, $this->data);

    }

    public function start($name, $restart = false) {

        // re-install shmop (if restart)
        if ( $restart === true ) {
            $worker = $this->get($name);
            // close the previous channels
            $worker->getInputChannel()->close();
            $worker->getOutputChannel()->close();
            // open brand new channels
            $worker->setInputChannel(new SharedMemory((int) '1'.hexdec($worker->getInstance()->getId())))
                ->setOutputChannel(new SharedMemory((int) '2'.hexdec($worker->getInstance()->getId())));
        }

        // fork worker
        $pid = pcntl_fork();

        if ( $pid == -1 ) {
            $this->logger->error("Could not create worker $name (fork error)");
            $daemon->end(1);

The variable $daemon seems to be never defined.

        }

        if ( $pid ) {
            $this->logger->notice("Worker $name created with pid $pid");
            $this->setPid($name, $pid);
            return;
        }

        // get daemon and worker
        $worker = $this->get($name);
        $daemon = $this->daemon;

        // update pid reference
        $daemon->setPid(ProcessTools::getPid());

        // cleanup events
        $daemon->getSignals()->any()->setDefault();

        // unmask signals (if restart)
        if ( $restart === true ) {
            $daemon->getSignals()->any()->unmask();
        }

        // inject events, logger and signals
        $logger = $daemon->getLogger()->withName($name);

The method withName() does not exist on Psr\Log\LoggerInterface. It seems like you code against a sub-type of Psr\Log\LoggerInterface such as Monolog\Logger.

        // $events = new EventsManager($logger);
        $events = clone $this->getEvents();
        $signals = new PosixSignals;
        $worker->getInstance()->setLogger($logger);
        $worker->getInstance()->setEvents($events);
        $worker->getInstance()->setSignals($signals);

        $loop = new Loop($worker);

It seems like $worker can also be of type array; however, parameter $worker of Comodojo\Daemon\Worker\Loop::__construct() does only seem to accept Comodojo\Daemon\Worker\Worker, maybe add an additional type check?

        $daemon->declass();

        $loop->start();

        $daemon->end(0);

    }

    public function stop($name = null) {

        foreach ( $this->data as $wname => $worker ) {

            $wpid = $worker->getPid();

            if ( is_null($name) || $name == $wname ) {

                // fix the wait time;
                $time = time() + 5;

                // try to gently ask the worker to close
                $worker->getOutputChannel()->send('stop');

                while ( time() < $time ) {

                    if ( !$this->running($wpid) ) break;
                    usleep(20000);

                }

                // close the shared memory block
                $worker->getInputChannel()->close();
                $worker->getOutputChannel()->close();

                // terminate the worker if still alive
                if ( $this->running($wpid) ) ProcessTools::term($wpid, 5, SIGTERM);

            }

        }

    }

    public function pause($name = null) {

        if ( empty($name) ) {
            $result = [];
            foreach ( $this->data as $name => $worker ) {
                $result[$name] = $worker->getOutputChannel()->send('pause') > 0;
            }
            return $result;
        }

        return $this->get($name)->getOutputChannel()->send('pause') > 0;

    }

    public function resume($name = null) {

        if ( empty($name) ) {
            $result = [];
            foreach ( $this->data as $name => $worker ) {
                $result[$name] = $worker->getOutputChannel()->send('resume') > 0;
            }
            return $result;
        }

        return $this->get($name)->getOutputChannel()->send('resume') > 0;

    }

    public function running($pid) {

        return ProcessTools::isRunning($pid);

    }

    public function status($name = null) {

        if ( $name === null ) {

            $result = [];
            foreach ( $this->data as $name => $worker ) {
                $result[$name] = $this->getStatus($worker);
            }
            return $result;

        }

        $worker = $this->get($name);

        return $this->getStatus($worker);

It seems like $worker can also be of type array; however, parameter $worker of Comodojo\Daemon\Worker\Manager::getStatus() does only seem to accept Comodojo\Daemon\Worker\Worker, maybe add an additional type check?

    }

    private function getStatus(Worker $worker) {

        return $worker->getInputChannel()->read();

    }

 }