AsyncSockets\RequestExecutor\Pipeline\Pipeline

Complexity

Total Complexity

6

Size/Duplication

Total Lines	74
Duplicated Lines	0 %

Coupling/Cohesion

Components	2
Dependencies	2

Test Coverage

Coverage

100%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	9	1
A	disconnectSockets()	0	4	1
A	process()	0	13	4

<?php
/**
 * Async sockets
 *
 * @copyright Copyright (c) 2015-2017, Efimov Evgenij <edefimov.it@gmail.com>
 *
 * This source file is subject to the MIT license that is bundled
 * with this source code in the file LICENSE.
 */
namespace AsyncSockets\RequestExecutor\Pipeline;

use AsyncSockets\RequestExecutor\Metadata\RequestDescriptor;
use AsyncSockets\RequestExecutor\Metadata\SocketBag;

/**
 * Class Pipeline
 */
class Pipeline
{
    /**
     * DisconnectStage
     *
     * @var PipelineStageInterface
     */
    private $disconnectStage;

    /**
     * Connect stage
     *
     * @var PipelineStageInterface
     */
    private $connectStage;

    /**
     * PipelineStageInterface
     *
     * @var PipelineStageInterface[]
     */
    private $stages;

    /**
     * Pipeline constructor
     *
     * @param PipelineStageInterface   $connectStage Connect stage
     * @param PipelineStageInterface[] $stages Pipeline stages
     * @param PipelineStageInterface   $disconnectStage Disconnect stages
     */
    public function __construct(
        PipelineStageInterface $connectStage,
        array $stages,
        PipelineStageInterface $disconnectStage
    ) {
        $this->connectStage    = $connectStage;
        $this->stages          = $stages;
        $this->disconnectStage = $disconnectStage;
    }

    /**
     * Process I/O operations on sockets
     *
     * @param SocketBag $socketBag Socket bag
     *
     * @return void
     * @throws \Exception
     */
    public function process(SocketBag $socketBag)
    {
        do {
            $activeOperations = $this->connectStage->processStage($socketBag->getItems());
            if (!$activeOperations) {
                break;
            }

            foreach ($this->stages as $stage) {
                $activeOperations = $stage->processStage($activeOperations);
            }
        } while (true);
    }

    /**
     * Disconnect given list of sockets
     *
     * @param RequestDescriptor[] $items Sockets' operations
     *
     * @return void
     */
    public function disconnectSockets(array $items)
    {
        $this->disconnectStage->processStage($items);
    }
}