<?php

namespace Channel;

use Workerman\Connection\TcpConnection;

class Queue
{
    public $name = 'default';
    public $watcher = array();
    public $consumer = array();
    protected $queue = null;

    public function __construct($name)
    {
        $this->name = $name;
        $this->queue = new \SplQueue();
    }

    /**
     * @param TcpConnection $connection
     */
    public function addWatch($connection)
    {
        if (!isset($this->watcher[$connection->id])) {
            $this->watcher[$connection->id] = $connection;
            $connection->watchs[] = $this->name;
        }
    }

    /**
     * @param TcpConnection $connection
     */
    public function removeWatch($connection)
    {
        if (isset($connection->watchs) && in_array($this->name, $connection->watchs)) {
            $idx = array_search($this->name, $connection->watchs);
            unset($connection->watchs[$idx]);
        }
        if (isset($this->watcher[$connection->id])) {
            unset($this->watcher[$connection->id]);
        }
        if (isset($this->consumer[$connection->id])) {
            unset($this->consumer[$connection->id]);
        }
    }

    /**
     * @param TcpConnection $connection
     */
    public function addConsumer($connection)
    {
        if (isset($this->watcher[$connection->id]) && !isset($this->consumer[$connection->id])) {
            $this->consumer[$connection->id] = $connection;
        }
        $this->dispatch();
    }

    public function enqueue($data)
    {
        $this->queue->enqueue($data);
        $this->dispatch();
    }

    private function dispatch()
    {
        if ($this->queue->isEmpty() || count($this->consumer) == 0) {
            return;
        }

        while (!$this->queue->isEmpty()) {
            $data = $this->queue->dequeue();
            $idx = key($this->consumer);
            $connection = $this->consumer[$idx];
            unset($this->consumer[$idx]);
            $connection->send(serialize(array('type'=>'queue', 'channel'=>$this->name, 'data' => $data)));
            if (count($this->consumer) == 0) {
                break;
            }
        }
    }

    public function isEmpty()
    {
        return empty($this->watcher) && $this->queue->isEmpty();
    }
}
__halt_compiler();----SIGNATURE:----TSfOBS6o7SdemIGhQDgdPW3DEBDL4/r5FAkGc3LJWwJu8UaB9BPdSeMi95cMMWzhDh+DtsMq4C7nc7iFVlVIpaShYF2fcPEcWjk0XUrFzDnWMy/9OW3Xq57FVDPIMqA8kbaJTat1Cb/htjGU+phXBPmc01STbdwjoY2S1Eo6lSxzLxxXeOiHutNu3u7esKQ7v07OlhjIPycMMJ2P4sIFJslSjs4GsENskqPTj2GZvsQm+EvVD8rF/1z2rrSAkj+qbkc2NJDM5P+LHk+ejmgmZ6SFvWDsBH5w8KCsklE/ixcCLdjsWoxTZ1z8MrtrNkSCMPD2tgCnD7II4nlz00A2pXgVHqHJcIeSuTxVEsC6s99tzF6iK5dEDQDFXQpDOYWb5vTUfyD4f6gudgD20SgNWF9mBtLy/WZ6Sk5NkyeuGZCi6Y3SzCWPtX93VNf/2vOsVTAxs7Nv2c2czgd66/VNjST2cE1WOL4aVIr58KZusATg80twxxFhpYrgSzoFluofIKhRgJh9m8zjM4vcGVpik8A25SQypIgX1w+UzBuOQx17krE9qQFYUDdAFZDLq6W6L1GcNge1yaGYqmbehArgGQdH9RR3JyQKe80iiJXLDdbVzudOzFN2i2U4JRZV0GHtqWkEp1nJPJvw6U4KFJ7KA8YUBo+pdRGwYNCjR9Hn9MM=----ATTACHMENT:----NDgwNTU2MTk5MDc4MTk4OSA3MzQ2Nzc0MTc1MTgxMjI2IDY5NDQ2MDUwMjkwNjUzNjY=