BitWasp\Bitcoin\Networking\Peer\Manager

Complexity

Total Complexity

16

Size/Duplication

Total Lines	192
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	8

Test Coverage

Coverage

83.56%

Importance

Changes	6
Bugs	1	Features	1

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	1
A	registerOutboundPeer()	0	12	1
A	registerInboundPeer()	0	9	1
A	registerListener()	0	8	1
A	connect()	0	4	1
A	connectToPeers()	0	9	2
A	getAnotherPeer()	0	20	2
A	attemptNextPeer()	0	23	1
B	connectNextPeer()	0	27	6

<?php

namespace BitWasp\Bitcoin\Networking\Peer;

use BitWasp\Bitcoin\Networking\Structure\NetworkAddressInterface;
use Evenement\EventEmitter;
use React\Promise\Deferred;
use React\Promise\RejectedPromise;
use React\Promise\Timer\TimeoutException;

class Manager extends EventEmitter
{
    /**
     * @var Connector
     */
    private $connector;

    /**
     * @var Peer[]
     */
    private $outPeers = [];

    /**
     * @var Peer[]
     */
    private $inPeers = [];

    /**
     * @var int
     */
    private $nOutPeers = 0;

    /**
     * @var int
     */
    private $nInPeers = 0;

    /**
     * Manager constructor.
     * @param Connector $connector
     */
    public function __construct(Connector $connector)
    {
        $this->connector = $connector;
    }

    /**
     * Store the newly connected peer, and trigger a new connection if they go away.
     *
     * @param Peer $peer
     * @return Peer
     */
    public function registerOutboundPeer(Peer $peer)
    {
        $next = $this->nOutPeers++;
        $peer->on('close', function ($peer) use ($next) {
            $this->emit('disconnect', [$peer]);
            unset($this->outPeers[$next]);
        });

        $this->outPeers[$next] = $peer;
        $this->emit('outbound', [$peer]);
        return $peer;
    }

    /**
     * @param Peer $peer
     */
    public function registerInboundPeer(Peer $peer)
    {
        $next = $this->nInPeers++;
        $this->inPeers[$next] = $peer;
        $peer->on('close', function () use ($next) {
            unset($this->inPeers[$next]);
        });
        $this->emit('inbound', [$peer]);
    }

    /**
     * @param Listener $listener
     * @return $this
     */
    public function registerListener(Listener $listener)
    {
        $listener->on('connection', function (Peer $peer) {
            $this->registerInboundPeer($peer);
        });

        return $this;
    }

    /**
     * @param NetworkAddressInterface $address
     * @return \React\Promise\PromiseInterface
     * @throws \Exception
     */
    public function connect(NetworkAddressInterface $address)
    {
        return $this->connector->connect($address);
    }

    /**
     * @param Locator $locator
     * @return \React\Promise\Promise|\React\Promise\PromiseInterface
     */
    public function getAnotherPeer(Locator $locator)
    {
        $deferred = new Deferred();

        // Otherwise, rely on the Locator.
        try {
            $deferred->resolve($locator->popAddress());
        } catch (\Exception $e) {
            $locator->queryDnsSeeds()->then(
                function () use ($deferred, $locator) {
                    $deferred->resolve($locator->popAddress());
                },
                function ($error) use ($deferred) {
                    return new RejectedPromise($error);
                }
            );
        }

        return $deferred->promise();
    }

    /**
     * @param Locator $locator
     * @return \React\Promise\Promise|\React\Promise\PromiseInterface
     */
    public function attemptNextPeer(Locator $locator)
    {
        $attempt = new Deferred();

        $this
            ->getAnotherPeer($locator)
            ->then(function (NetworkAddressInterface $address) use ($attempt, $locator) {
                return $this->connect($address)->then(
                    function (Peer $peer) use ($attempt) {
                        $this->registerOutboundPeer($peer);
                        $attempt->resolve($peer);
                        return $peer;
                    },
                    function (\Exception $error) use ($attempt) {
                        $attempt->reject($error);
                    }
                );
            }, function ($error) use ($attempt) {
                $attempt->reject($error);
            });

        return $attempt->promise();
    }

    /**
     * @param Locator $locator
     * @param int $retries
     * @return \React\Promise\PromiseInterface
     */
    public function connectNextPeer(Locator $locator, $retries = 5)
    {
        $errorBack = function ($error) use ($locator, $retries) {
            $allowContinue = false;
            if ($error instanceof \RuntimeException && $error->getMessage() === "Connection refused") {
                $allowContinue = true;
            }

            if ($error instanceof TimeoutException) {
                $allowContinue = true;
            }

            if (!$allowContinue) {
                throw $error;
            }

            if (0 >= $retries) {
                throw new \RuntimeException("Connection to peers failed: too many retries");
            }

            return $this->connectNextPeer($locator, $retries - 1);
        };

        return $this
            ->attemptNextPeer($locator)
            ->then(null, $errorBack);
    }

    /**
     * @param Locator $locator
     * @param int $n
     * @return \React\Promise\Promise
     */
    public function connectToPeers(Locator $locator, $n)
    {
        $peers = [];
        for ($i = 0; $i < $n; $i++) {
            $peers[$i] = $this->connectNextPeer($locator);
        }

        return \React\Promise\all($peers);
    }
}