Tidal\WampWatch\Subscription\Collection

Complexity

Total Complexity

16

Size/Duplication

Total Lines	149
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes	4
Bugs	2	Features	2

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	1
A	addSubscription()	0	5	1
A	hasSubscription()	0	4	1
A	subscribe()	0	10	3
A	doSubscribe()	0	12	2
A	unsubscribe()	0	17	3
A	isSubscribed()	0	4	1
A	isSubscribing()	0	4	1
A	checkSubscribed()	0	13	3

<?php

/*
 * This file is part of the Tidal/WampWatch package.
 *   (c) 2016 Timo Michna <timomichna/yahoo.de>
 *
 *  For the full copyright and license information, please view the LICENSE
 *  file that was distributed with this source code.
 *
 */

namespace Tidal\WampWatch\Subscription;

use Thruway\Message\SubscribedMessage;
use React\Promise\Deferred;
use React\Promise\Promise;
use Tidal\WampWatch\ClientSessionInterface as ClientSession;
use Tidal\WampWatch\Util;

class Collection
{
    /**
     * The collection's WAMP client session.
     *
     * @var ClientSession
     */
    private $session;

    /**
     * @var array list of subscriptions with topic as key and subscription-id as value
     */
    private $subscriptions = [];

    /**
     * @var array list of subscriptions callbacks with topic as key
     */
    private $subscriptionCallbacks = [];

    /**
     * @var bool if the collection is successfully subscribed to all topics
     */
    private $isSubscribed = false;

    /**
     * @var bool if the collection is currently trying to subscribe to all topics
     */
    private $isSubscribing = false;

    /**
     * @var Deferred
     */
    private $subscriptionPromise;

    /**
     * Collection constructor.
     *
     * @param \Tidal\WampWatch\ClientSessionInterface $session
     */
    public function __construct(ClientSession $session)
    {
        $this->session = $session;
    }

    /**
     * @param string   $topic    the topic the subscription is for
     * @param callable $callback the callback for the topic
     */
    public function addSubscription($topic, callable $callback)
    {
        $this->subscriptions[$topic] = 0;
        $this->subscriptionCallbacks[$topic] = $callback;
    }

    /**
     * @return bool
     */
    public function hasSubscription()
    {
        return count($this->subscriptions) > 0;
    }

    /**
     * Subscribe to all topics added with 'addSubscription'.
     * Returns false if already subscribed or curretly subscribing.
     *
     * @return \React\Promise\Promise
     */
    public function subscribe()
    {
        if (!$this->isSubscribed() && !$this->isSubscribing()) {
            $this->isSubscribing = true;
            $this->subscriptionPromise = new Deferred();
            $this->doSubscribe();
        }

        return $this->subscriptionPromise->promise();
    }

    /**
     *
     */
    protected function doSubscribe()
    {
        foreach (array_keys($this->subscriptions) as $topic) {
            $this->session->subscribe($topic, $this->subscriptionCallbacks[$topic])
                ->done(function (SubscribedMessage $msg) use ($topic) {
                    $this->subscriptions[$topic] = $msg->getSubscriptionId();
                    $this->subscriptionPromise->notify($topic);

                    $this->checkSubscribed();
                });
        }
    }

    /**
     * @return \React\Promise\Promise|\React\Promise\PromiseInterface
     */
    public function unsubscribe()
    {
        $resolver = function (callable $resolve) {
            $resolve();
        };
        $promise = new  Promise($resolver);

        if ($this->isSubscribed()) {
            foreach ($this->subscriptions as $topic => $subId) {
                Util::unsubscribe($this->session, $subId);
            }

            $this->isSubscribed = false;
        }

        return $promise;
    }

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

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

    /**
     * Check if all subscriptions have been successfully confirmed.
     */
    protected function checkSubscribed()
    {
        foreach ($this->subscriptions as $topic => $subId) {
            if ($subId === 0) {
                return false;
            }
        }
        $this->isSubscribed = true;
        $this->isSubscribing = false;
        $this->subscriptionPromise->resolve($this->subscriptions);

        return true;
    }
}