mpyw\Co\Internal\Pool

Complexity

Total Complexity

11

Size/Duplication

Total Lines	97
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Test Coverage

Coverage

100%

Importance

Changes	2
Bugs	0	Features	1

5 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	9	1
A	addCurl()	0	4	1
A	addDelay()	0	4	1
B	wait()	0	18	7
A	reserveHaltException()	0	4	1

<?php

namespace mpyw\Co\Internal;
use mpyw\Co\CURLException;
use mpyw\RuntimePromise\Deferred;

class Pool
{
    /**
     * Options.
     * @var CoOption
     */
    private $options;

    /**
     * cURL multi handle.
     * @var resource
     */
    private $mh;

    /**
     * Used for halting loop.
     * @var \RuntimeException
     */
    private $haltException;

    /**
     * cURL handle scheduler.
     * @var Scheduler
     */
    private $scheduler;

    /**
     * Delay controller.
     * @var Delayer
     */
    private $delayer;

    /**
     * Constructor.
     * Initialize cURL multi handle.
     * @param CoOption $options
     */
    public function __construct(CoOption $options)
    {
        $this->mh = curl_multi_init();
        $flags = (int)$options['pipeline'] + (int)$options['multiplex'] * 2;
        curl_multi_setopt($this->mh, CURLMOPT_PIPELINING, $flags);
        $this->options = $options;
        $this->scheduler = new Scheduler($options, $this->mh);
        $this->delayer = new Delayer;
    }

    /**
     * Call curl_multi_add_handle() or push into queue.
     * @param resource $ch
     * @param Deferred $deferred
     */
    public function addCurl($ch, Deferred $deferred = null)
    {
        $this->scheduler->add($ch, $deferred);
    }

    /**
     * Add delay.
     * @param int      $time
     * @param Deferred $deferred
     */
    public function addDelay($time, Deferred $deferred)
    {
        $this->delayer->add($time, $deferred);
    }

    /**
     * Run curl_multi_exec() loop.
     */
    public function wait()
    {
        curl_multi_exec($this->mh, $active); // Start requests.
        do {
            // if cURL handle is running, use curl_multi_select()
            // otherwise, just sleep until nearest time
            $this->scheduler->isEmpty()
                ? $this->delayer->sleep()
                : curl_multi_select($this->mh, $this->options['interval']) < 0
                  && usleep($this->options['interval'] * 1000000);
            curl_multi_exec($this->mh, $active);
            $this->scheduler->consume();
            $this->delayer->consume();
        } while (!$this->haltException && (!$this->scheduler->isEmpty() || !$this->delayer->isEmpty()));
        if ($this->haltException) {
            throw $this->haltException;
        }
    }

    /**
     * Used for halting loop.
     */
    public function reserveHaltException($e)
    {
        $this->haltException = $e;
    }
}