Socket Implementation

socket.php

<?php

//http://code.google.com/p/phpsocketdaemon/source/browse/trunk/socket.php

namespace Atom;

class Socket extends \Atom\Design\Creational\Factory {

	// Make static access available.
}

namespace Atom\Socket;

abstract class Driver {

	protected $socket;
	protected $address;
	protected $localAddress;
	protected $port;
	protected $localPort;
	protected $domain;
	protected $type;
	protected $protocol;

	protected $readBuffer = '';
	protected $writeBuffer = '';
	protected $errorBuffer = '';

	public function __construct($address, $port, $domain = AF_INET, $type = SOCK_STREAM, $protocol = SOL_TCP)
	{
		$this->address  = $address;
		$this->port     = $port;
		$this->domain   = $domain;
		$this->type     = $type;
		$this->protocol = $protocol;

		$this->socket = socket_create($domain, $type, $protocol);

		if($this->socket === false)
		{
			throw new \RuntimeException('Unable to create socket');
		}

		if(!socket_set_option($this->socket, SOL_SOCKET, SO_REUSEADDR, 1))
		{
			throw new \RuntimeException('Could not configure socket');
		}

		if(!socket_bind($this->socket, $address, $port))
		{
			throw new \RuntimeException('Could not bind socket to ['.$address.':'.$port.']');
		}

		if(!socket_getsockname($this->socket, $this->localAddress, $this->localPort))
		{
			throw new \RuntimeException('Could not retrieve local address and port: '.socket_strerror(socket_last_error($this->socket)));
		}

		$this->allowBlocking(false);
	}

	public function __destruct()
	{
		$this->close();
	}

	public function getError()
	{
		$error = socket_strerror(socket_last_error($this->socket));
		         socket_clear_error($this->socket);

		return $error;
	}

	public function close()
	{
		method_exists($this. 'close_callback') and $this->close_callback();

		socket_shutdown($this->socket, 2);
		socket_close($this->socket);

		$this->socket = null;
	}

	public function write($buffer)
	{
		method_exists($this. 'write_callback') and $this->write_callback($buffer);

		socket_write($this->socket, $buffer, mb_strlen($buffer));

		return $this;
	}

	public function read($length = 1024)
	{
		method_exists($this. 'read_callback') and $this->read_callback($length);

		return socket_read($this->socket, $length, PHP_BINARY_READ);
	}

	public function connect($address, $port)
	{
		$this->address = $address;
		$this->port    = $port;

		if(!socket_connect($this->socket, $address, $port)
		{
			throw new \RuntimeException('Could not connect to '.$address.':'.$port.': '.$this->getError());
		}

		method_exists($this. 'connect_callback') and $this->connect_callback($address, $port);

		return $this;
	}

	public function listen($backLog = 128)
	{
		if(!socket_listen($this->socket, $backLog))
		{
			throw new \RuntimeException('Could not listen to '.$address.':'.$port.': '.$this->getError());
		}

		method_exists($this. 'listen_callback') and $this->listen_callback($backLog);

		return $this;
	}

	public function accept()
	{
		$client = socket_listen($this->socket, $backLog);

		if(!$client)
		{
			throw new \RuntimeException('Could not listen to '.$address.':'.$port.': '.$this->getError());
		}

		return $client;
	}

	public function allowBlocking($block = true)
	{
		$result = $block ? socket_set_block($this->socket) : socket_set_nonblock($this->socket);

		if(!$result)
		{
			throw new \RuntimeException('Could not set block: '.$this->getError());
		}

		return $this;
	}

	public function setTimeout($seconds, $milliseconds)
	{
		if(!socket_set_option($this->socket, SOL_SOCKET, SO_RCVTIMEO, array('sec' => $seconds, 'usec' => $milliseconds)))
		{
			throw new \RuntimeException('Could not set timeout: '.$this->getError());
		}

		return $this;
	}

	public function reuseAddress($reuse = true)
	{
		if(!socket_set_option($this->socket, SOL_SOCKET, SO_REUSEADDR, $reuse))
		{
			throw new \RuntimeException('Could not set reuse address: '.$this->getError());
		}

		return $this;
	}
}

namespace Atom\Socket\Driver;

class Client extends \Atom\Socket\Driver {

	public $connected  = false;

	public function connect($address, $port)
	{
		try
		{
			parent::connect($address, $port);
		}
		catch(\RuntimeException $e)
		{
			return false;
		}

		$this->connected = true;

		return $this;
	}
}

namespace Atom\Socket\Driver;

class Server extends \Atom\Socket\Driver {

	protected $clientClass;

	public function __construct($address, $port, $domain = AF_INET, $type = SOCK_STREAM, $protocol = SOL_TCP)
	{
		parent::__construct($address, $port, $domain, $type, $protocol);

		$this->clientclass = '\\Atom\\Socket\\Driver\\Client';

		$this->listen();
	}

	public function accept()
	{
		$client = new $this->clientClass(parent::accept());

		method_exists($this. 'accept_callback') and $this->accept_callback($client);

		return $client;
	}

	public function setClientClass($clientClass)
	{
		if(!$clientClass instanceof \Atom\Socket\Driver\Client)
		{
			throw new \InvalidArgumentException('Client class must extend \\Atom\\Socket\\Driver\\Client');
		}

		$this->clientClass = $clientClass;

		return $this;
	}
}

namespace Atom\Socket\Driver;

class Daemon extends \Atom\Socket\Driver\Server {

	// Daemonize functionality... loop.
}