library/Rediska/PubSub/Channel.php
protected function _execCommand($command, $channels)
    {
        // Group channels by connections
        $channelsByConnections = array();
        foreach($channels as $channel) {
            $hasSubscription = in_array($channel, $this->_subscriptions);

            if ($command == self::SUBSCRIBE) {
                if ($hasSubscription) {
                    throw new Rediska_PubSub_Exception("You already subscribed to $channel");
                } else {
                    $this->_subscriptions[] = $channel;
                }
            }

            if ($command == self::UNSUBSCRIBE) {
                if (!$hasSubscription) {
                    throw new Rediska_PubSub_Exception("You not subscribed to $channel");
                } else {
                    $key = array_search($channel, $this->_subscriptions);
                    unset($this->_subscriptions[$key]);
                }
            }

            $hasChannel = $this->_connections->hasChannel($channel);

            if (($command == self::SUBSCRIBE && !$hasChannel) || ($command == self::UNSUBSCRIBE && $hasChannel)) {
                switch ($command) {
                    case self::SUBSCRIBE:
                        $connection = $this->_connections->addChannel($channel);
                        break;
                    case self::UNSUBSCRIBE:
                        $connection = $this->_connections->removeChannel($channel);
                        break;
                }

                $connectionAlias = $connection->getAlias();
                if (!array_key_exists($connectionAlias, $channelsByConnections)) {
                    $channelsByConnections[$connectionAlias] = array();
                }
                $channelsByConnections[$connectionAlias][] = $channel;
            }
        }

        // Write commands to connections
        foreach($channelsByConnections as $connectionAlias => $channels) {
            $execCommand = array($command);
            foreach($channels as $channel) {
                $execCommand[] = $this->getRediska()->getOption('namespace') . $channel;
            }
            $connection = $this->_connections->getConnectionByAlias($connectionAlias);
            $exec = new Rediska_Connection_Exec($connection, $execCommand);
            $exec->write();
        }

        // Get responses from connections
        while (!empty($channelsByConnections)) {
            foreach ($channelsByConnections as $connectionAlias => $channels) {
                $connection = $this->_connections->getConnectionByAlias($connectionAlias);

                foreach($channels as $channel) {
                    $response = $this->_getResponseFromConnection($connection);

                    // TODO: May be timeout? Not data or server die()
                    if ($response === null) {
                        continue;
                    }

                    $channel = $response->getChannel();

                    if (($command == self::SUBSCRIBE && $response instanceof Rediska_PubSub_Response_Subscribe)
                     || ($command == self::UNSUBSCRIBE && $response instanceof Rediska_PubSub_Response_Unsubscribe)) {
                        $key = array_search($channel, $channelsByConnections[$connectionAlias]);

                        unset($channelsByConnections[$connectionAlias][$key]);

                        if (empty($channelsByConnections[$connectionAlias])) {
                            unset($channelsByConnections[$connectionAlias]);
                        }
                    } else if ($response instanceof Rediska_PubSub_Response_Message) {
                        $this->_addMessageToBuffer($response);
                    }
                }
            }
        }
    }
protected function _execCommand($command, $channels)
    {
        // Group channels by connections
        $channelsByConnections = array();
        foreach($channels as $channel) {
            $hasSubscription = in_array($channel, $this->_subscriptions);

            if ($command == self::SUBSCRIBE) {
                if ($hasSubscription) {
                    throw new Rediska_PubSub_Exception("You already subscribed to $channel");
                } else {
                    $this->_subscriptions[] = $channel;
                }
            }

            if ($command == self::UNSUBSCRIBE) {
                if (!$hasSubscription) {
                    throw new Rediska_PubSub_Exception("You not subscribed to $channel");
                } else {
                    $key = array_search($channel, $this->_subscriptions);
                    unset($this->_subscriptions[$key]);
                }
            }

            $hasChannel = $this->_connections->hasChannel($channel);

            if (($command == self::SUBSCRIBE && !$hasChannel) || ($command == self::UNSUBSCRIBE && $hasChannel)) {
                switch ($command) {
                    case self::SUBSCRIBE:
                        $connection = $this->_connections->addChannel($channel);
                        break;
                    case self::UNSUBSCRIBE:
                        $connection = $this->_connections->removeChannel($channel);
                        break;
                }

                $connectionAlias = $connection->getAlias();
                if (!array_key_exists($connectionAlias, $channelsByConnections)) {
                    $channelsByConnections[$connectionAlias] = array();
                }
                $channelsByConnections[$connectionAlias][] = $channel;
            }
        }

        // Write commands to connections
        foreach($channelsByConnections as $connectionAlias => $channels) {
            $execCommand = array($command);
            foreach($channels as $channel) {
                $execCommand[] = $this->getRediska()->getOption('namespace') . $channel;
            }
            $connection = $this->_connections->getConnectionByAlias($connectionAlias);
            $exec = new Rediska_Connection_Exec($connection, $execCommand);
            $exec->write();
        }

        // Get responses from connections
        while (!empty($channelsByConnections)) {
            foreach ($channelsByConnections as $connectionAlias => $channels) {
                $connection = $this->_connections->getConnectionByAlias($connectionAlias);

                foreach($channels as $channel) {
                    $response = $this->_getResponseFromConnection($connection);

                    // TODO: May be timeout? Not data or server die()
                    if ($response === null) {
                        continue;
                    }

                    $channel = $response->getChannel();

                    if (($command == self::SUBSCRIBE && $response instanceof Rediska_PubSub_Response_Subscribe)
                     || ($command == self::UNSUBSCRIBE && $response instanceof Rediska_PubSub_Response_Unsubscribe)) {
                        $key = array_search($channel, $channelsByConnections[$connectionAlias]);

                        unset($channelsByConnections[$connectionAlias][$key]);

                        if (empty($channelsByConnections[$connectionAlias])) {
                            unset($channelsByConnections[$connectionAlias]);
                        }
                    } else if ($response instanceof Rediska_PubSub_Response_Message) {
                        $this->_addMessageToBuffer($response);
                    }
                }
            }
        }
    }
Execute subscribe or unsubscribe command

@param string $command
@param array $channels