Classes/Connection/ConnectionPool.php

<?php
/**
 * Connection pool class. Contains the SocketPool.
 * @author jpt
 * @package Connection
 * @depends Socket
 */
class Connection_ConnectionPool {

        /**
         * SocketPool instance.
         * @var Socket_SocketPool
         */
        protected $socketPool;

        /**
         * Contains all ConnectionHandler instances.
         * @var array
         */
        protected $connectionHandlers;

        /**
         * @var int Next ID for a new ConnectionHandler
         */
        protected $nextID;

        /**
         * Creates an Instance of SocketPool
         * @return void
         */
        function __construct($linebreak = "\r\n") {
                $this->connectionHandlers = array();
                $this->socketPool = new Socket_SocketPool($linebreak);
                $this->nextID = 1;
        }

        /**
         * Destroys the SocketPool
         * @return void
         */
        function __destruct() {
                unset($this->socketPool);
        }

        /**
         * Creates a new TcpConnection.
         * @param boolean $IPv6 will determine whether the socket uses IPv4 or IPv6.
         * @return Connection_ConnectionHandler
         */
        public function createTcpConnection($group = "", $protocol = "RAW", $IPv6 = FALSE) {
                $socket = $this->socketPool->createTcpSocket($IPv6);
                $connectionHandler = new Connection_ConnectionHandler($socket, $this->nextID, $group, $protocol);
                $this->addConnectionHandler($connectionHandler);
                return $connectionHandler;
        }

        /**
         * Adds a ConnectionHandler to the pool.
         * @param Connection_ConnectionHandler $add_connectionHandler
         * @return void
         */
        public function addConnectionHandler($add_connectionHandler) {
                array_push($this->connectionHandlers, $add_connectionHandler);
                $this->nextID++;
        }

        /**
         * Removes a ConnectionHandler from the pool.
         * @param Connection_ConnectionHandler $remove_connectionHandler
         * @return void
         */
        public function removeConnectionHandler($remove_connectionHandler) {
                foreach($this->connectionHandlers AS $key=>$connectionHandler) {
                        if($connectionHandler === $remove_connectionHandler) {
                                $this->socketPool->removeSocket($remove_connectionHandler->getSocket());
                                $remove_connectionHandler->close();
                                unset($this->connectionHandlers[$key]);
                        }
                }
        }

        /**
         * Returns ConnectionHandler for the given socket ressource.
         * @param ressource $socketRessource
         * @return Connection_ConnectionHandler
         */
        protected function getConnectionHandlerForSocketRessource($socketRessource) {
                foreach($this->connectionHandlers AS $connectionHandler) {
                        if($connectionHandler->getSocket() === $socketRessource) {
                                return $connectionHandler;
                        }
                }
        }

        /**
         * Calls select() on SocketPool and updates the ConnectionHandler.
         * Will also accept incoming connections and add them to the pool.
         * @return array An array containing the Connection_ConnectionHandler for each Socket with new data.
         * @throws Exception_GeneralException
         * @throws Exception_SocketException
         */
        public function select() {
                $read = array();
                $write = array();
                $except = array();
                foreach($this->connectionHandlers AS $connectionHandler) {
                        $connectionSocket = $connectionHandler->getSocket();
                        $read[] = $connectionSocket;
                        if($connectionHandler->canWrite() && $connectionHandler->isServer() === FALSE) $write[] = $connectionSocket;
                }
                $except = $read;

                $tempArray = array();
                $selectedSockets = $this->socketPool->select($read, $write, $except);
                foreach($selectedSockets AS $selectedType=>$selectedArray) { //read, write, except, this loop won't kill performance
                        foreach($selectedArray AS $socket) {
                                $connectionHandler = $this->getConnectionHandlerForSocketRessource($socket);
                                switch($selectedType) {
                                        case "read":
                                                if($connectionHandler->isServer() === FALSE) {
                                                        if($connectionHandler->readToBuffer() === FALSE) $this->removeConnectionHandler($connectionHandler);
                                                } else {
                                                        $acceptedSocket = $connectionHandler->accept();
                                                        $acceptedSocketHandler = new Connection_ConnectionHandler($acceptedSocket, $this->nextID, $connectionHandler->getGroup(), $connectionHandler->getProtocol());
                                                        $acceptedSocketHandler->hasBeenAccepted();
                                                        $this->addConnectionHandler($acceptedSocketHandler);
                                                }
                                                break;
                                        case "write":
                                                if($connectionHandler->writeFromBuffer() === FALSE) $this->removeConnectionHandler($connectionHandler);
                                                break;
                                        case "except":
                                                $connectionHandler->handleSocketError();
                                                break;
                                        default:
                                                throw new Exception_GeneralException("Unknown select type: '" . $selectedType . "'", 1289737080);
                                                break;
                                }
                                $tempArray[$selectedType][] = $connectionHandler;
                        }
                }
                return $tempArray;
        }

        /**
         * Writes the given data to all sockets in $group.
         * @param string $group
         * @param string $data
         * @throws Exception_SocketException
         * @return void
         */
        public function writeToGroup($group, $data) {
                foreach($this->connectionHandlers AS $connectionHandler) {
                        if($connectionHandler->getGroup() === $group && $connectionHandler->isServer() === FALSE) {
                                $connectionHandler->write($data);
                        }
                }
        }

        /**
         * Writes the given data to the socket with $id.
         * @param int $id
         * @param string $data
         */
        public function writeToID($id, $data) {
                foreach($this->connectionHandlers AS $connectionHandler) {
                        if($connectionHandler->getID() === $id && $connectionHandler->isServer() === FALSE) {
                                $connectionHandler->write($data);
                        }
                }
        }

        /**
         * @see Socket_SocketPool
         * @return int
         */
        public function countConnections() {
                return $this->socketPool->countSockets();
        }
}
?>