File Source for nanoserv.php

<?php
/**
*
* nanoserv - a sockets daemon toolkit for PHP 5.1+
*
* Copyright (C) 2004-2010 Vincent Negrier aka. sIX <six at aegis-corp.org>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*
* @package nanoserv
* @subpackage Core
*/
/**
* nanoserv current version number
* @var string
*/
define("NS_VERSION", "2.1.1-dev");
/**
* Base exception class
*
* @package nanoserv
* @subpackage Core
* @since 2.0
*/
abstract class NS_Exception extends Exception {
public $addr;
public function __construct($errmsg, $errno, $addr) {
parent::__construct($errmsg, $errno);
$this->addr = $addr;
}
}
/**
* Server exception class
*
* @package nanoserv
* @subpackage Core
* @since 2.0
*/
class NS_Server_Exception extends NS_Exception {
public $listener;
public function __construct($errmsg, $errno, $addr, NS_Listener $listener = NULL) {
parent::__construct($errmsg, $errno, $addr);
$this->listener = $listener;
}
}
/**
* Client exception class
*
* @package nanoserv
* @subpackage Core
* @since 2.0
*/
class NS_Client_Exception extends NS_Exception {
public $handler;
public function __construct($errmsg, $errno, $addr, NS_Handler $handler = NULL) {
parent::__construct($errmsg, $errno, $addr);
$this->handler = $handler;
}
}
/**
* Base socket class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_Socket {
/**
* Maximum number of bytes read by Read()
* @var int
*/
const DEFAULT_READ_LENGTH = 16384;
/**
* Internal Socket unique ID
* @var int
*/
public $id;
/**
* Socket stream descriptor
* @var resource
*/
public $fd;
/**
* Is the socket connected ?
* @var bool
*/
public $connected = false;
/**
* Is the socket waiting to be connected ?
* @var bool
*/
public $pending_connect = false;
/**
* Is the socket waiting for ssl/tls handshake ?
* @var bool
*/
public $pending_crypto = false;
/**
* Is the socket blocked ?
* @var bool
*/
public $blocked = false;
/**
* Should we block reading from this socket ?
* @var bool
*/
public $block_reads = false;
/**
* Stream context
* @var resource
*/
protected $context;
/**
* Crypto type
* @var int
*/
public $crypto_type;
/**
* Attached handler
* @var NS_Connection_Handler
*/
public $handler;
/**
* Static instance counter
* @var int
*/
private static $sck_cnt;
/**
* NS_Socket contructor
*
* @param resource $fd
*/
public function __construct($fd = false, $crypto_type = false) {
if ($fd === false) {
$this->context = stream_context_create();
} else {
$this->fd = $fd;
$this->connected = true;
$this->Set_Blocking(false);
$this->Set_Timeout(0);
if ($crypto_type) $this->crypto_type = $crypto_type;
}
$this->id = ++NS_Socket::$sck_cnt;
}
/**
* Get stream options
*
* @return array
* @since 0.9
*/
public function Get_Options() {
if ($this->fd) {
return stream_context_get_options($this->fd);
} else {
return stream_context_get_options($this->context);
}
}
/**
* Set a stream context option
*
* @param string $wrapper
* @param string $opt
* @param mixed $val
* @return bool
* @since 0.9
*/
public function Set_Option($wrapper, $opt, $val) {
if ($this->fd) {
return stream_context_set_option($this->fd, $wrapper, $opt, $val);
} else {
return stream_context_set_option($this->context, $wrapper, $opt, $val);
}
}
/**
* Set timeout
*
* @param int $timeout
* @return bool
* @since 0.9
*/
protected function Set_Timeout($timeout) {
return stream_set_timeout($this->fd, $timeout);
}
/**
* Sets wether the socket is blocking or not
*
* @param bool $block
* @return bool
* @since 0.9
*/
protected function Set_Blocking($block) {
return stream_set_blocking($this->fd, $block);
}
/**
* Flag the socket so that the main loop won't read from it even if data is available.
*
* This can be used to implement flow control when proxying data between two asymetric connections for example.
*
* @param bool $block
* @return bool the previous status
* @since 2.0.3
*/
public function Block_Reads($block) {
$ret = $this->block_reads;
$this->block_reads = $block;
return $ret;
}
/**
* Set the stream write buffer (PHP defaults to 8192 bytes)
*
* @param int $buffer_size
* @return int
* @since 2.0
*/
public function Set_Write_Buffer($buffer_size) {
return stream_set_write_buffer($this->fd, $buffer_size);
}
/**
* Enable or disable ssl/tls crypto on the socket
*
* @param bool $enable
* @param int $type
* @return mixed
* @since 0.9
*/
public function Enable_Crypto($enable = true, $type = false) {
if ($type) $this->crypto_type = $type;
$ret = @stream_socket_enable_crypto($this->fd, $enable, $this->crypto_type);
$this->pending_crypto = $ret === 0;
return $ret;
}
/**
* Setup crypto if needed
*
* @return bool
* @since 0.9
*/
public function Setup() {
if (isset($this->crypto_type)) return $this->Enable_Crypto();
return true;
}
/**
* Get local socket name
*
* @return string
* @since 0.9
*/
public function Get_Name() {
return stream_socket_get_name($this->fd, false);
}
/**
* Get remote socket name
*
* @return string
* @since 0.9
*/
public function Get_Peer_Name() {
return stream_socket_get_name($this->fd, true);
}
/**
* Read data from the socket and return it
*
* @param int $length maximum read length
* @return string
* @since 0.9
*/
public function Read() {
return fread($this->fd, self::DEFAULT_READ_LENGTH);
}
/**
* Read data from a non connected socket and return it
*
* @param string &$addr contains the message sender address upon return
* @param int $len maximum read length
* @return string
* @since 0.9.61
*/
public function Read_From(&$addr, $len = 16384) {
return stream_socket_recvfrom($this->fd, $len, NULL, $addr);
}
/**
* Write data to the socket
*
* write returns the number of bytes written to the socket
*
* @param string $data
* @return int
* @since 0.9
*/
public function Write($data) {
$nb = fwrite($this->fd, $data);
if (isset($data[$nb])) $this->blocked = true;
return $nb;
}
/**
* Write data to a non connected socket
*
* @param string $to in the form of "<ip_address>:<port>"
* @param string $data
* @return int
* @since 0.9.61
*/
public function Write_To($to, $data) {
return stream_socket_sendto($this->fd, $data, NULL, $to);
}
/**
* Write data from stream to socket
*
* returns the number of bytes read from the stream and written to the socket
*
* @param resource $stream
* @param int $len maximum length (bytes) to read/write
* @return int
* @since 2.1
*/
public function Write_From_Stream($stream, $len = 16384) {
return stream_copy_to_stream($stream, $this->fd, $len);
}
/**
* Query end of stream status
*
* @return bool
* @since 0.9
*/
public function Eof() {
$fd = $this->fd;
if (!is_resource($fd)) return true;
stream_socket_recvfrom($fd, 1, STREAM_PEEK);
return feof($fd);
}
/**
* Close the socket
* @since 0.9
*/
public function Close() {
@fclose($this->fd);
$this->connected = $this->pending_connect = false;
}
/**
* NS_Socket destructor
*/
public function __destruct() {
Nanoserv::Free_Write_Buffers($this->id);
$this->Close();
}
}
/**
* Server socket class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_Server_Socket extends NS_Socket {
/**
* Listen address (format is 'proto://addr:port')
* @var string
*/
public $address;
/**
* Real listen address (format is 'proto://addr:port')
* @var string
*/
private $real_address;
/**
* NS_Server_Socket constructor
*/
public function __construct($addr) {
parent::__construct();
$this->address = $addr;
$proto = strtolower(strtok($addr, ":"));
if (($proto === "udp") || ($proto === "unix")) {
$this->real_address = $addr;
} else {
$this->real_address = "tcp:" . strtok("");
if ($proto !== "tcp") switch ($proto) {
case "ssl": $this->crypto_type = STREAM_CRYPTO_METHOD_SSLv23_SERVER; break;
case "tls": $this->crypto_type = STREAM_CRYPTO_METHOD_TLS_SERVER; break;
case "sslv2": $this->crypto_type = STREAM_CRYPTO_METHOD_SSLv2_SERVER; break;
case "sslv3": $this->crypto_type = STREAM_CRYPTO_METHOD_SSLv3_SERVER; break;
default:
if (defined($cname = "STREAM_CRYPTO_METHOD_".strtoupper($proto)."_SERVER")) {
$this->crypto_type = constant($cname);
} else {
throw new NS_Server_Exception("unknown transport/crypto type '{$proto}'");
}
}
}
}
/**
* Start listening and accepting connetions
*
* @return bool
* @since 0.9
*/
public function Listen($bind_only = false) {
$errno = $errstr = false;
$this->fd = @stream_socket_server($this->real_address, $errno, $errstr, STREAM_SERVER_BIND | ($bind_only ? 0 : STREAM_SERVER_LISTEN), $this->context);
if ($this->fd === false) {
throw new NS_Server_Exception("cannot listen to {$this->real_address}: {$errstr}", $errno, $this->real_address);
}
$this->Set_Blocking(false);
$this->Set_Timeout(0);
return true;
}
/**
* Accept connection
*
* @return resource
* @since 0.9
*/
public function Accept() {
return @stream_socket_accept($this->fd, 0);
}
}
/**
* Client socket class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_Client_Socket extends NS_Socket {
/**
* Connect timeout (seconds)
* @var int
*/
const CONNECT_TIMEOUT = 10;
/**
* Peer address (format is 'proto://addr:port')
* @var string
*/
public $address;
/**
* Connect timeout (timestamp)
* @var int
*/
public $connect_timeout;
/**
* NS_Client_Socket constructor
*/
public function __construct($addr) {
parent::__construct();
$this->address = $addr;
$proto = strtolower(strtok($addr, ":"));
$s = strtok("");
if (($proto === "udp") || ($proto === "unix")) {
$this->real_address = $addr;
} else {
$this->real_address = "tcp:" . $s;
if ($proto != "tcp") switch ($proto) {
case "ssl": $this->crypto_type = STREAM_CRYPTO_METHOD_SSLv23_CLIENT; break;
case "tls": $this->crypto_type = STREAM_CRYPTO_METHOD_TLS_CLIENT; break;
case "sslv2": $this->crypto_type = STREAM_CRYPTO_METHOD_SSLv2_CLIENT; break;
case "sslv3": $this->crypto_type = STREAM_CRYPTO_METHOD_SSLv3_CLIENT; break;
default: if (defined($cname = "STREAM_CRYPTO_METHOD_".strtoupper($proto)."_CLIENT")) $this->crypto_type = constant($cname);
}
}
}
/**
* Connect to the peer address
*
* @param int $timeout connection timeout in seconds
* @return bool
* @since 0.9
*/
public function Connect($timeout = false) {
$errno = $errstr = false;
$this->fd = @stream_socket_client($this->real_address, $errno, $errstr, 3, STREAM_CLIENT_ASYNC_CONNECT | STREAM_CLIENT_CONNECT, $this->context);
if ($this->fd === false) {
throw new <a href="../nanoserv/Core/NS_Client_Exception.html">NS_Client_Exception</a>("cannot connect to {$this->real_address}: {$errstr}", $errno, $this->real_address);
}
if ($timeout === false) $timeout = self::CONNECT_TIMEOUT;
$this->connect_timeout = microtime(true) + $timeout;
$this->pending_connect = true;
$this->connected = false;
$this->Set_Blocking(false);
$this->Set_Timeout(0);
return true;
}
}
/**
* IPC Socket class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_IPC_Socket extends NS_Socket {
/**
* Maximum size of inter process communication packets
* @var int
*/
const IPC_MAX_PACKET_SIZE = 1048576;
/**
* pid number of the remote forked process
* @var int
*/
public $pid;
/**
* IPC Socket constructor
*
* @param resource $fd
* @param int $pid
*/
public function __construct($fd, $pid=false) {
parent::__construct($fd);
$this->Set_Write_Buffer(self::IPC_MAX_PACKET_SIZE);
$this->pid = $pid;
}
/**
* Read data from IPC socket
*
* @return string
* @since 0.9
*/
public function Read() {
return fread($this->fd, self::IPC_MAX_PACKET_SIZE);
}
/**
* Creates a pair of connected, indistinguishable pipes
*
* Returns an array of two NS_IPC_Socket objects
*
* @param int $domain
* @param int $type
* @param int $proto
* @return array
* @since 0.9
*/
static public function Pair($domain = STREAM_PF_UNIX, $type = STREAM_SOCK_DGRAM, $proto = 0) {
list($s1, $s2) = stream_socket_pair($domain, $type, $proto);
return array(new NS_IPC_Socket($s1), new NS_IPC_Socket($s2));
}
/**
* Ask the master process for object data
*
* @param array $request
* @param bool $need_response
* @return mixed
* @since 0.9
*/
public function Ask_Master($request, $need_response = true) {
$this->Write(serialize($request));
if (!$need_response) return;
$rfd = array($this->fd);
$dfd = array();
if (@stream_select($rfd, $dfd, $dfd, 600)) return unserialize($this->Read());
}
}
/**
* Timer class
*
* Do not instanciate NS_Timer but use the Nanoserv::New_Timer() method instead
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_Timer {
/**
* System time for timer activation
* @var float
*/
public $microtime;
/**
* Timer callback
* @var mixed
*/
public $callback;
/**
* Timer status
* @var bool
*/
public $active = true;
/**
* NS_Timer constructor
*
* @param float $time
* @param mixed $callback
* @since 0.9
* @see Nanoserv::New_Timer()
*/
public function __construct($time, $callback) {
$this->microtime = $time;
$this->callback = $callback;
}
/**
* Activate timer
*
* Timers are activated by default, and Activate should only be used after a call do Deactivate()
*
* @see NS_Timer::Deactivate()
*/
public function Activate() {
$this->active = true;
}
/**
* Deactivate timer
*/
public function Deactivate() {
$this->active = false;
}
}
/**
* Write buffer interface
*/
interface NS_I_Write_Buffer {
/**
* Setup a new write buffer
*
* @param NS_Socket $socket
* @param mixed $data
* @param mixed $callback
*/
public function __construct(NS_Socket $socket, $data, $callback = false);
/**
* Get availability of data
*
* @return bool
* @since 0.9
*/
public function Waiting_Data();
/**
* Write data to socket and advance buffer pointer
*
* @param int $length
*/
public function Write($length = NULL);
}
/**
* Write buffer base class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
abstract class NS_Write_Buffer {
/**
* Attached socket
* @var NS_Socket
*/
public $socket;
/**
* Buffered data
* @var string
*/
protected $data;
/**
* End-of-write Callback
* @var mixed
*/
protected $callback = false;
/**
* NS_Write_Buffer constructor
*
* @param NS_Socket $socket
* @param mixed $data
* @param mixed $callback
*/
public function __construct(NS_Socket $socket, $data, $callback = false) {
$this->socket = $socket;
$this->data = $data;
$this->callback = $callback;
}
/**
* NS_Write_Buffer destructor
*/
public function __destruct() {
if ($this->callback) call_user_func($this->callback, $this->Waiting_Data());
}
}
/**
* Static write buffer class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_Static_Write_Buffer extends NS_Write_Buffer implements NS_I_Write_Buffer {
/**
* Buffered data pointer
* @var int
*/
private $pointer = 0;
/**
* Get availability of data
*
* @return bool
* @since 0.9
*/
public function Waiting_Data() {
return isset($this->data[$this->pointer]);
}
/**
* Write data to socket and advance buffer pointer
*
* @param int $length
* @since 1.1
*/
public function Write($length = 16384) {
$this->pointer += $this->socket->Write(substr($this->data, $this->pointer, $length));
}
}
/**
* Stream write buffer class
*
* @package nanoserv
* @subpackage Core
* @since 2.1
*/
class NS_Stream_Write_Buffer extends NS_Write_Buffer implements NS_I_Write_Buffer {
/**
* Get availability of data from stream
*
* @return bool
* @since 0.9
*/
public function Waiting_Data() {
return !@feof($this->data);
}
/**
* Read data from stream and write it to socket
*
* @param int $length
* @since 1.1
*/
public function Write($length = 16384) {
return $this->socket->Write_From_Stream($this->data, $length);
}
}
/**
* Base handler class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
abstract class NS_Handler {
/**
* Attached socket
* @var NS_Socket
*/
public $socket;
/**
* Set a stream context option
*
* @param string $wrapper
* @param string $opt
* @param mixed $val
* @return bool
* @since 0.9
*/
public function Set_Option($wrapper, $opt, $val) {
return $this->socket->Set_Option($wrapper, $opt, $val);
}
}
/**
* Datagram listener / handler class
*
* @package nanoserv
* @subpackage Core
* @since 0.9.61
*/
abstract class NS_Datagram_Handler extends NS_Handler {
/**
* Is the listener active ?
* @var bool
*/
public $active = false;
/**
* NS_Datagram_Handler constructor
*
* @param string $addr
* @param string $handler_classname
* @param mixed $handler_options
*/
public function __construct($addr) {
$this->socket = new NS_Server_Socket($addr);
}
/**
* Activate the listener
*
* @return bool
* @since 0.9.61
*/
public function Activate() {
try {
if ($ret = $this->socket->Listen(true)) $this->active = true;
return $ret;
} catch (NS_Server_Exception $e) {
throw new NS_Server_Exception($e->getMessage(), $e->getCode(), $e->addr, $this);
}
}
/**
* Deactivate the listener
* @since 0.9.61
*/
public function Deactivate($close_socket = true) {
if ($close_socket) {
$this->socket->Close();
}
$this->active = false;
}
/**
* Send data over the connection
*
* @param string $to in the form of "<ip_address>:<port>"
* @param string $data
* @return int
* @since 0.9.61
*/
public function Write($to, $data) {
return $this->socket->Write_To($to, $data);
}
/**
* Event called on data reception
*
* @param string $from
* @param string $data
* @since 0.9.61
*/
public function on_Read($from, $data) {
}
/**
* NS_Datagram_Handler destructor
*/
public function __destruct() {
$this->Deactivate();
}
}
/**
* Connection handler class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
abstract class NS_Connection_Handler extends NS_Handler {
/**#@+
* Cause of connection failure
* @var int
*/
const FAIL_CONNREFUSED = 1;
const FAIL_TIMEOUT = 2;
const FAIL_CRYPTO = 3;
/**#@-*/
/**
* Send data over the connection
*
* @param string $data
* @param mixed $callback
* @return NS_Static_Write_Buffer
* @since 0.9
*/
public function Write($data, $callback=false) {
return Nanoserv::New_Static_Write_Buffer($this->socket, $data, $callback);
}
/**
* Send open stream over the connection
*
* @param resource $stream
* @param mixed $callback
* @return NS_Stream_Write_Buffer
* @since 2.1
*/
public function Write_Stream($stream, $callback=false) {
return Nanoserv::New_Stream_Write_Buffer($this->socket, $stream, $callback);
}
/**
* Connect
*
* @param int $timeout timeout in seconds
* @since 0.9
*/
public function Connect($timeout=false) {
try {
$this->socket->Connect($timeout);
} catch (NS_Client_Exception $e) {
Nanoserv::Free_Connection($this);
throw new NS_Client_Exception($e->getMessage(), $e->getCode(), $e->addr, $this);
}
}
/**
* Disconnect
*/
public function Disconnect() {
$this->socket->Close();
Nanoserv::Free_Connection($this);
}
/**
* Event called on received connection
* @since 0.9
*/
public function on_Accept() {
}
/**
* Event called on established connection
* @since 0.9
*/
public function on_Connect() {
}
/**
* Event called on failed connection
*
* @param int $failcode see NS_Connection_Handler::FAIL_* constants
* @since 0.9
*/
public function on_Connect_Fail($failcode) {
}
/**
* Event called on disconnection
* @since 0.9
*/
public function on_Disconnect() {
}
/**
* Event called on data reception
*
* @param string $data
* @since 0.9
*/
public function on_Read($data) {
}
/**
* Event called before forking
*
* @since 2.0
*/
public function on_Fork_Prepare() {
}
/**
* Event called after forking, both on master and child processes
*
* @since 2.0
*/
public function on_Fork_Done() {
}
}
/**
* Listener class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class NS_Listener {
/**
* Attached socket
* @var NS_Server_Socket
*/
public $socket;
/**
* Name of the handler class
* @var string
* @see NS_Connetion_Handler
*/
public $handler_classname;
/**
* Handler options
*
* this is passed as the first constructor parameter of each spawned connection handlers
*
* @var mixed
*/
public $handler_options;
/**
* Is the listener active ?
* @var bool
*/
public $active = false;
/**
* If set the listener will fork() a new process for each accepted connection
* @var bool
*/
public $forking = false;
/**
* NS_Listener constructor
*
* @param string $addr
* @param string $handler_classname
* @param mixed $handler_options
*/
public function __construct($addr, $handler_classname, $handler_options=false, $forking=false) {
$this->socket = new NS_Server_Socket($addr);
$this->handler_classname = $handler_classname;
$this->handler_options = $handler_options;
$this->forking = ($forking && is_callable("pcntl_fork"));
}
/**
* Set a stream context option
*
* @param string $wrapper
* @param string $opt
* @param mixed $val
* @return bool
* @since 0.9
*/
public function Set_Option($wrapper, $opt, $val) {
return $this->socket->Set_Option($wrapper, $opt, $val);
}
/**
* Sets wether the listener should fork() a new process for each accepted connection
*
* @param bool $forking
* @return bool
* @since 0.9
*/
public function Set_Forking($forking=true) {
if ($forking && !is_callable("pcntl_fork")) return false;
$this->forking = $forking;
return true;
}
/**
* Activate the listener
*
* @return bool
* @since 0.9
*/
public function Activate() {
try {
if ($ret = $this->socket->Listen()) $this->active = true;
return $ret;
} catch (<a href="../nanoserv/Core/NS_Server_Exception.html">NS_Server_Exception</a> $e) {
throw new <a href="../nanoserv/Core/NS_Server_Exception.html">NS_Server_Exception</a>($e->getMessage(), $e->getCode(), $e->addr, $this);
}
}
/**
* Deactivate the listener
* @since 0.9
*/
public function Deactivate() {
$this->socket->Close();
$this->active = false;
}
/**
* NS_Listener destructor
*/
public function __destruct() {
$this->Deactivate();
}
}
/**
* Shared object class for inter-process communications
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
class <a href="../nanoserv/Core/NS_Shared_Object.html">NS_Shared_Object</a> {
/**
* caller process pid
* @var int
*/
static public $caller_pid;
/**
* shared object unique identifier
* @var int
*/
public $_oid;
/**
* wrapped object
* @var object
*/
private $wrapped;
/**
* static instance counter
* @var int
*/
static public $shared_count = 0;
/**
/**
* NS_Shared_Object constructor
*
* If $o is omited, a new StdClass object will be created and wrapped
*
* @param object $o
*/
public function __construct($o=false) {
if ($o === false) $o = new StdClass();
$this->_oid = ++self::$shared_count;
$this->wrapped = $o;
}
public function __get($k) {
if (Nanoserv::$child_process) {
return Nanoserv::$master_pipe->Ask_Master(array("oid" => $this->_oid, "action" => "G", "var" => $k));
} else {
return $this->wrapped->$k;
}
}
public function __set($k, $v) {
if (<a href="../nanoserv/Core/Nanoserv.html">Nanoserv</a>::$child_process) {
Nanoserv::$master_pipe->Ask_Master(array("oid" => $this->_oid, "action" => "S", "var" => $k, "val" => $v), false);
} else {
$this->wrapped->$k = $v;
}
}
public function __call($m, $a) {
if (<a href="../nanoserv/Core/Nanoserv.html">Nanoserv</a>::$child_process) {
return Nanoserv::$master_pipe->Ask_Master(array("oid" => $this->_oid, "action" => "C", "func" => $m, "args" => $a));
} else {
return call_user_func_array(array($this->wrapped, $m), $a);
}
}
}
/**
* Server / multiplexer class
*
* @package nanoserv
* @subpackage Core
* @since 0.9
*/
final class Nanoserv {
/**
* nanoserv current version number
* @var string
*/
const VERSION = "2.1.1-dev";
/**
* Registered listeners
* @var array
*/
static private $listeners = array();
/**
* Write buffers
* @var array
*/
static private $write_buffers = array();
/**
* Active connections
* @var array
*/
static private $connections = array();
/**
* Active datagram handlers
* @var array
*/
static private $dgram_handlers = array();
/**
* Shared objects
* @var array
*/
static private $shared_objects = array();
/**
* Forked process pipes
* @var array
*/
static private $forked_pipes = array();
/**
* Timers
* @var array
*/
static private $timers = array();
/**
* Timers updated
* @var bool
*/
static private $timers_updated = false;
/**
* Number of active connection handler processes
* @var int
*/
static public $nb_forked_processes = 0;
/**
* Maximum number of active children before incoming connections get delayed
* @var int
*/
static public $max_forked_processes = 64;
/**
* Are we master or child process ?
* @var bool
*/
static public $child_process = false;
/**
* Forked server handled connection
* @var NS_Connection_Handler
*/
static private $forked_connection;
/**
* Forked server pipe to the master process
* @var NS_Socket
*/
static public $master_pipe;
/**
* Class Nanoserv should not be instanciated but used statically
*/
private function __construct() {
}
/**
* Register a new listener
*
* For consistency New_Listener() will also wrap Nanoserv::New_Datagram_Handler() if the given addr is of type "udp"
*
* @param string $addr
* @param string $handler_classname
* @param mixed $handler_options
* @return NS_Listener
* @see NS_Listener
* @see NS_Datagram_Handler
* @since 0.9
*/
static public function New_Listener($addr, $handler_classname, $handler_options=false) {
if (strtolower(strtok($addr, ":")) == "udp") {
$l = self::New_Datagram_Handler($addr, $handler_classname);
} else {
$l = new <a href="../nanoserv/Core/NS_Listener.html">NS_Listener</a>($addr, $handler_classname, $handler_options);
self::$listeners[] = $l;
}
return $l;
}
/**
* Deactivate and free a previously registered listener
*
* For consistency Free_Listener() will also wrap Nanoserv::Free_Datagram_Handler() if the given object is an instance of NS_Datagram_Handler
*
* @param NS_Listener $l
* @return bool
* @see NS_Listener
* @see NS_Datagram_Handler
* @since 0.9
*/
static public function Free_Listener($l) {
if ($l instanceof NS_Listener) {
foreach (self::$listeners as $k => $v) if ($v === $l) {
unset(self::$listeners[$k]);
return true;
}
} else if ($l instanceof <a href="../nanoserv/Core/NS_Datagram_Handler.html">NS_Datagram_Handler</a>) {
return self::Free_Datagram_Handler($l);
}
return false;
}
/**
* Register a new static write buffer
*
* This method is used by NS_Connection_Handler::Write() and should not be
* called unless you really know what you are doing
*
* @param NS_Socket $socket
* @param string $data
* @param mixed $callback
* @return NS_Static_Write_Buffer
* @see NS_Connection_Handler::Write()
* @since 0.9
*/
static public function New_Static_Write_Buffer(NS_Socket $socket, $data, $callback=false) {
$wb = new <a href="../nanoserv/Core/NS_Static_Write_Buffer.html">NS_Static_Write_Buffer</a>($socket, $data, $callback);
$wb->Write();
if ($wb->Waiting_Data()) {
self::$write_buffers[$socket->id][] = $wb;
}
return $wb;
}
/**
* Register a new static write buffer
*
* This method is used by NS_Connection_Handler::Write_Stream() and should not be
* called unless you really know what you are doing
*
* @param NS_Socket $socket
* @param resource $stream
* @param mixed $callback
* @return NS_Stream_Write_Buffer
* @see NS_Connection_Handler::Write_Stream()
* @since 0.9
*/
static public function New_Stream_Write_Buffer(NS_Socket $socket, $data, $callback=false) {
$wb = new NS_Stream_Write_Buffer($socket, $data, $callback);
$wb->Write();
if ($wb->Waiting_Data()) {
self::$write_buffers[$socket->id][] = $wb;
}
return $wb;
}
/**
* Free a registered write buffer
*
* @param int $sid socket id
* @since 0.9
*/
static public function Free_Write_Buffers($sid) {
unset(self::$write_buffers[$sid]);
}
/**
* Register a new outgoing connection
*
* @param string $addr
* @param string $handler_classname
* @param mixed $handler_options
* @return NS_Connection_Handler
* @see NS_Connection_Handler
* @since 0.9
*/
static public function New_Connection($addr, $handler_classname, $handler_options=false) {
$sck = new NS_Client_Socket($addr);
$h = new $handler_classname($handler_options);
$h->socket = $sck;
self::$connections[$sck->id] = $h;
return $h;
}
/**
* Free an allocated connection
*
* @param NS_Connection_Handler $h
* @return bool
* @since 0.9
*/
static public function Free_Connection(NS_Connection_Handler $h) {
$so = $h->socket;
unset(self::$connections[$so->id]);
self::Free_Write_Buffers($so->id);
$so->pending_connect = $so->pending_crypto = $so->connected = false;
if (self::$child_process && (self::$forked_connection === $h)) exit();
return true;
}
/**
* Register a new datagram (udp) handler
*
* @param string $addr
* @param string $handler_classname
* @return NS_Datagram_Handler
* @see NS_Datagram_Handler
* @since 0.9.61
*/
static public function New_Datagram_Handler($addr, $handler_classname) {
$h = new $handler_classname($addr);
self::$dgram_handlers[$h->socket->id] = $h;
return $h;
}
/**
* Deactivate and free a datagram handler
*
* @param NS_Datagram_Handler $h
* @return bool
* @since 0.9.61
*/
static public function Free_Datagram_Handler(NS_Datagram_Handler $h) {
unset(self::$dgram_handlers[$h->socket->id]);
return true;
}
/**
* Register a new shared object
*
* shared objects allow forked processes to use objects stored on the master process
* if $o is ommited, a new StdClass empty object is created
*
* @param object $o
* @return NS_Shared_Object
* @since 0.9
*/
static public function New_Shared_Object($o = false) {
$shr = new <a href="../nanoserv/Core/NS_Shared_Object.html">NS_Shared_Object</a>($o);
self::$shared_objects[$shr->_oid] = $shr;
return $shr;
}
/**
* Free a shared object
*
* @param NS_Shared_Object $o
* @since 0.9
*/
static public function Free_Shared_Object(NS_Shared_Object $o) {
unset(self::$shared_objects[$o->_oid]);
}
/**
* Register a new timer callback
*
* @param float $delay specified in seconds
* @param mixed $callback may be "function" or array($obj, "method")
* @return NS_Timer
* @since 0.9
*/
static public function New_Timer($delay, $callback) {
$t = new NS_Timer(microtime(true) + $delay, $callback);
self::$timers[] = $t;
self::$timers_updated = true;
return $t;
}
/**
* Clear all existing timers
*
* @return int number of timers cleared
* @since 2.0
*/
static public function Clear_Timers() {
$ret = count(self::$timers);
self::$timers = array();
return $ret;
}
/**
* Get all registered NS_Connection_Handler objects
*
* Note: connections created by fork()ing listeners can not be retreived this way
*
* @param bool $include_pending_connect
* @return array
* @since 0.9
*/
static public function Get_Connections($include_pending_connect=false) {
$ret = array();
foreach (self::$connections as $c) if ($c->socket->connected || $include_pending_connect) $ret[] = $c;
return $ret;
}
/**
* Get all registered NS_Listener objects
*
* @param bool $include_inactive
* @return array
* @since 0.9
*/
static public function Get_Listeners($include_inactive=false) {
$ret = array();
foreach (self::$listeners as $l) if ($l->active || $include_inactive) $ret[] = $l;
return $ret;
}
/**
* Get all registered NS_Timer objects
*
* @param bool $include_inactive
* @return array
* @since 2.0.1
*/
static public function Get_Timers($include_inactive=false) {
$ret = array();
foreach (self::$timers as $t) if ($t->active || $include_inactive) $ret[] = $t;
return $ret;
}
/**
* Set the maximum number of allowed children processes before delaying incoming connections
*
* Note: this setting only affect and applies to forking listeners
*
* @param int $i
* @since 2.0
*/
static public function Set_Max_Children($i) {
self::$max_forked_processes = $i;
}
/**
* Flush all write buffers
*
* @since 2.0
*/
static public function Flush_Write_Buffers() {
while (self::$write_buffers) {
self::Run(1);
}
}
/**
* Fork and setup IPC sockets
*
* @return int the pid of the created process, 0 if child process
* @since 0.9.63
*/
static public function Fork() {
if ($has_shared = (<a href="../nanoserv/Core/NS_Shared_Object.html">NS_Shared_Object</a>::$shared_count > 0)) {
list($s1, $s2) = NS_IPC_Socket::Pair();
}
$pid = pcntl_fork();
if ($pid === 0) {
self::$child_process = true;
if ($has_shared) {
self::$master_pipe = $s2;
}
} else if ($pid > 0) {
++self::$nb_forked_processes;
if ($has_shared) {
$s1->pid = $pid;
self::$forked_pipes[$pid] = $s1;
}
}
return $pid;
}
/**
* Enter main loop
*
* The <var>$time</var> parameter can have different meanings:
* <ul>
* <li>int or float > 0 : the main loop will run once and will wait for activity for a maximum of <var>$time</var> seconds</li>
* <li>0 : the main loop will run once and will not wait for activity when polling, only handling waiting packets and timers</li>
* <li>int or float < 0 : the main loop will run for -<var>$time</var> seconds exactly, whatever may happen</li>
* <li>NULL : the main loop will run forever</li>
* </ul>
*
* @param float $time how much time should we run, if omited nanoserv will enter an endless loop
* @param array $user_streams if specified, user streams will be polled along with internal streams
* @return array the user streams with pending data
* @since 0.9
*/
static public function Run($time = NULL, array $user_streams = NULL) {
$tmp = 0;
$ret = array();
if (isset($time)) {
if ($time < 0) {
$poll_max_wait = -$time;
$exit_mt = microtime(true) - $time;
} else {
$poll_max_wait = $time;
$exit = true;
}
} else {
$poll_max_wait = 60;
$exit = false;
}
do {
$t = microtime(true);
// Timers
if (self::$timers_updated) {
usort(self::$timers, function(NS_Timer $a, NS_Timer $b) { return $a->microtime > $b->microtime; });
self::$timers_updated = false;
}
$next_timer_md = NULL;
if (self::$timers) foreach (self::$timers as $k => $tmr) {
if ($tmr->microtime > $t) {
$next_timer_md = $tmr->microtime - $t;
break;
} else if ($tmr->active) {
$tmr->Deactivate();
call_user_func($tmr->callback);
}
unset(self::$timers[$k]);
}
if (self::$timers_updated) {
$t = microtime(true);
usort(self::$timers, function(NS_Timer $a, NS_Timer $b) { return $a->microtime > $b->microtime; });
foreach (self::$timers as $tmr) {
if ($tmr->microtime > $t) {
$next_timer_md = $tmr->microtime - $t;
break;
}
}
self::$timers_updated = false;
}
// Write buffers to non blocked sockets
foreach (self::$write_buffers as $write_buffers) {
if (!$write_buffers || $write_buffers[0]->socket->blocked || !$write_buffers[0]->socket->connected) continue;
foreach ($write_buffers as $wb) {
while ($wb->Waiting_Data() && !$wb->socket->blocked) {
$wb->Write();
if (!$wb->Waiting_Data()) {
array_shift(self::$write_buffers[$wb->socket->id]);
if (!self::$write_buffers[$wb->socket->id]) self::Free_Write_Buffers($wb->socket->id);
break;
}
}
}
}
$handler = $so = $write_buffers = $l = $c = $wbs = $wb = $data = $so = NULL;
// Prepare socket arrays
$fd_lookup_r = $fd_lookup_w = $rfd = $wfd = $efd = array();
foreach (self::$listeners as $l) if (($l->active) && ((!$l->forking) || (self::$nb_forked_processes <= self::$max_forked_processes))) {
$fd = $l->socket->fd;
$rfd[] = $fd;
$fd_lookup_r[(int)$fd] = $l;
}
$next_conn_timeout_mt = NULL;
foreach (self::$connections as $c) {
$so = $c->socket;
if ($so->pending_crypto) {
$cr = $so->Enable_Crypto();
if ($cr === true) {
$c->on_Accept();
} else if ($cr === false) {
$c->on_Connect_Fail(<a href="../nanoserv/Core/NS_Connection_Handler.html">NS_Connection_Handler</a>::FAIL_CRYPTO);
self::Free_Connection($c);
} else {
$fd = $so->fd;
$rfd[] = $fd;
$fd_lookup_r[(int)$fd] = $c;
}
} else if ($so->connected) {
if (!$so->block_reads) {
$fd = $so->fd;
$rfd[] = $fd;
$fd_lookup_r[(int)$fd] = $c;
}
} else if ($so->connect_timeout < $t) {
$c->on_Connect_Fail(<a href="../nanoserv/Core/NS_Connection_Handler.html">NS_Connection_Handler</a>::FAIL_TIMEOUT);
self::Free_Connection($c);
} else if ($so->pending_connect) {
$fd = $so->fd;
$wfd[] = $fd;
$fd_lookup_w[(int)$fd] = $c;
if (!$next_conn_timeout_mt || ($sc->connect_timeout < $next_conn_timeout_mt)) {
$next_conn_timeout_mt = $sc->connect_timeout;
}
}
}
if (self::$dgram_handlers) foreach (self::$dgram_handlers as $l) if ($l->active) {
$fd = $l->socket->fd;
$rfd[] = $fd;
$fd_lookup_r[(int)$fd] = $l;
}
foreach (self::$write_buffers as $wbs) if ($wbs[0]->socket->blocked) {
$fd = $wbs[0]->socket->fd;
$wfd[] = $fd;
$fd_lookup_w[(int)$fd] = self::$connections[$wbs[0]->socket->id];
}
if (self::$forked_pipes) foreach (self::$forked_pipes as $fp) {
$fd = $fp->fd;
$rfd[] = $fd;
$fd_lookup_r[(int)$fd] = $fp;
}
if (isset($user_streams)) {
foreach ((array)$user_streams[0] as $tmp_r) $rfd[] = $tmp_r;
foreach ((array)$user_streams[1] as $tmp_w) $wfd[] = $tmp_w;
}
// Main select
$wait_mds = array($poll_max_wait);
if (isset($next_timer_md)) $wait_mds[] = $next_timer_md;
if (isset($exit_mt)) $wait_mds[] = $exit_mt - $t;
if (isset($next_conn_timeout_mt)) $wait_mds[] = $next_conn_timeout_mt - $t;
$wait_md = min($wait_mds);
$tv_sec = (int)$wait_md;
$tv_usec = ($wait_md - $tv_sec) * 1000000;
if (($rfd || $wfd) && (@stream_select($rfd, $wfd, $efd, $tv_sec, $tv_usec))) {
foreach ($rfd as $act_rfd) {
$handler = $fd_lookup_r[(int)$act_rfd];
$so = $handler->socket;
if ($handler instanceof <a href="../nanoserv/Core/NS_Connection_Handler.html">NS_Connection_Handler</a>) {
if ($so->pending_crypto) {
$cr = $so->Enable_Crypto();
if ($cr === true) {
$handler->on_Accept();
} else if ($cr === false) {
$handler->on_Connect_Fail(<a href="../nanoserv/Core/NS_Connection_Handler.html">NS_Connection_Handler</a>::FAIL_CRYPTO);
self::Free_Connection($handler);
}
} else if (!$so->connected) {
continue;
}
$data = $so->Read();
if (($data === "") || ($data === false)) {
if ($so->Eof()) {
// Disconnected socket
$handler->on_Disconnect();
self::Free_Connection($handler);
}
} else {
// Data available
$handler->on_Read($data);
}
} else if ($handler instanceof <a href="../nanoserv/Core/NS_Datagram_Handler.html">NS_Datagram_Handler</a>) {
$from = "";
$data = $so->Read_From($from);
$handler->on_Read($from, $data);
} else if ($handler instanceof <a href="../nanoserv/Core/NS_Listener.html">NS_Listener</a>) {
while ($fd = $so->Accept()) {
// New connection accepted
$sck = new <a href="../nanoserv/Core/NS_Socket.html">NS_Socket</a>($fd, $so->crypto_type);
$hnd = new $handler->handler_classname($handler->handler_options);
$hnd->socket = $sck;
if ($handler->forking) {
$hnd->on_Fork_Prepare();
if (self::Fork() === 0) {
$hnd->on_Fork_Done();
self::$write_buffers = self::$listeners = array();
self::$connections = array($sck->id => $hnd);
self::$forked_connection = $hnd;
self::Clear_Timers();
if ($sck->Setup()) {
$hnd->on_Accept();
}
$handler = $hnd = $sck = $l = $c = $wbs = $wb = $fd_lookup_r = $fd_lookup_w = false;
break;
}
$hnd->on_Fork_Done();
if (self::$nb_forked_processes >= self::$max_forked_processes) break;
} else {
self::$connections[$sck->id] = $hnd;
if ($sck->Setup()) {
$hnd->on_Accept();
}
}
$sck = $hnd = NULL;
}
} else if ($handler instanceof <a href="../nanoserv/Core/NS_IPC_Socket.html">NS_IPC_Socket</a>) {
while ($ipcm = $handler->Read()) {
if ((!$ipcq = unserialize($ipcm)) || (!is_object($o = self::$shared_objects[$ipcq["oid"]]))) continue;
switch ($ipcq["action"]) {
case "G":
$handler->Write(serialize($o->$ipcq["var"]));
break;
case "S":
$o->$ipcq["var"] = $ipcq["val"];
break;
case "C":
<a href="../nanoserv/Core/NS_Shared_Object.html">NS_Shared_Object</a>::$caller_pid = $handler->pid;
$handler->Write(serialize(call_user_func_array(array($o, $ipcq["func"]), $ipcq["args"])));
break;
}
}
$o = $ipcq = $ipcm = NULL;
} else if (!isset($handler)) {
// User stream
$ret[0][] = $act_rfd;
}
}
foreach ($wfd as $act_wfd) {
$handler = $fd_lookup_w[$act_wfd];
$so = $handler->socket;
if (!isset($handler)) {
// User stream
$ret[1][] = $act_wfd;
} else if ($so->connected) {
// Unblock buffered write
if ($so->Eof()) {
$handler->on_Disconnect();
self::Free_Connection($handler);
} else {
$so->blocked = false;
}
} else if ($so->pending_connect) {
// Pending connect
if ($so->Eof()) {
$handler->on_Connect_Fail(<a href="../nanoserv/Core/NS_Connection_Handler.html">NS_Connection_Handler</a>::FAIL_CONNREFUSED);
self::Free_Connection($handler);
} else {
$so->Setup();
$so->connected = true;
$so->pending_connect = false;
$handler->on_Connect();
}
}
}
}
if (self::$nb_forked_processes && !self::$child_process) while ((($pid = pcntl_wait($tmp, WNOHANG)) > 0) && self::$nb_forked_processes--) unset(self::$forked_pipes[$pid]);
if ($ret) {
return $ret;
} else if (isset($exit_mt)) {
$exit = $exit_mt <= $t;
}
} while (!$exit);
}
}

Source for file nanoserv.php