2010-11-21 22:48:11 +00:00
parent 01018862f7
commit 952af43b97
23 changed files with 3314 additions and 0 deletions
@@ -0,0 +1,181 @@
+<?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();
+	}
+}
+?>