rendered paste body<?class ioe { //in/out/event private $client = array(); //store our objects, clients, servers and pipes private $server = array(); private $eventQueue = array(); //[$time][$idx] = [$time, $obj, $event, $params]; private $eventMessy = false; //whether we need to resort the eventQueue or not private $callback = array(); private $doReload = false, $reloadCB = false, $reloadCBArgs = array(); public function connect($host, $port, $cb = array()) { //connect to a server $sock = @stream_socket_client("tcp://".$host.":".$port, $errno, $errstr, 1); //short timeouts keep the program moving if (!$sock) { if (isset($cb['connected'])) $this->event(0, $cb['connected']); return false; } else { if (!stream_set_blocking($sock, false)) die("ioe::connect unable to set non-blocking"); $cIdx = (int)$sock; $this->client[$cIdx] = new ioe_client($this, $sock, false, $cIdx, $cb); $this->event(0, array($this->client[$cIdx], "trigger"), array("connected")); return $this->client[$cIdx]->face; } } public function serv($port, $cb = array()) { //create a server, timeout for clients $socket = stream_socket_server("tcp://0.0.0.0:".$port, $errno, $errstr); if (!stream_set_blocking($socket, false)) die("ioe::serv unable to set non-blocking"); if (!$socket) { die("ioe::serv failed to open port ".$port.": $errstr ($errno)\n"); } $idx = (int)$socket; $this->server[$idx] = new ioe_server($this, $socket, $idx, $cb); return $this->server[$idx]->face; } public function event($time, $func, $args = array()) { //register an event if ($time < 1118000000) { //randomly large number to decide if they mean a timestamp or a number of seconds $time = microtime(true)+(float)$time; } $time = (float)$time; if (!isset($this->eventQueue[$time])) { $this->eventQueue[$time] = array(); $idx = 0; } else { end($this->eventQueue[$time]); //move the pointer $idx = key($this->eventQueue[$time])+1; } $this->eventQueue[$time][$idx] = array($time, $func, $args); $this->eventMessy = true; return array($time, $idx); } public function delEvent($event) { //remove an event list ($time, $idx) = $event; $time = (float)$time; if (isset($this->eventQueue[$time][$idx])) { if (count($this->eventQueue[$time]) == 1) unset($this->eventQueue[$time]); else unset($this->eventQueue[$time][$idx]); $event = false; return true; } $event = false; return false; } public function reload($callBack = false, $callBackArgs = array()) { $this->doReload = true; $this->reloadCB = $callBack; $this->reloadCBArgs = $callBackArgs; } private function doEvents() { //retrieve the next event from the list if (count($this->eventQueue) == 0) { return false; } do { if ($this->eventMessy) { //is the event list untidy? ksort($this->eventQueue); //sort by time asc $this->eventMessy = false; } reset($this->eventQueue); $time = key($this->eventQueue); if ($time-microtime(true) <= 0) { while (count($this->eventQueue[$time]) > 0) { $event = array_shift($this->eventQueue[$time]); list ($t, $func, $args) = $event; call_user_func_array($func, $args); } unset($this->eventQueue[$time]); } if (count($this->eventQueue) == 0) { //nothing left $this->eventMessy = false; return false; } reset($this->eventQueue); } while (key($this->eventQueue)-microtime(true) <= 0); $next = key($this->eventQueue)-microtime(true); if ($next < 0) return 0; else return $next; } public function go() { //this is the start of the core code do { $wait = $this->doEvents(); $socks = array(); foreach ($this->client as $s) $socks[] = $s->socket; foreach ($this->server as $s) $socks[] = $s->socket; if (count($socks) == 0) { if ($wait === false) { return 0; } else { usleep($wait*1000000); continue; } } if ($wait === false) $r = stream_select($socks, $w = null, $e = null, null); else $r = stream_select($socks, $w = null, $e = null, floor($wait), floor((($wait-(int)$wait)*1000000))); if ($r === false) die("ioe::go failed to stream_select"); while (count($socks) > 0) { $sock = array_pop($socks); //address each socket in turn $cIdx = (int)$sock; //handy reference pointer if (isset($this->server[$cIdx])) { $socket = stream_socket_accept($sock, 0); if ($socket === false) die("ioe::go wouldn't accept socket"); $nIdx = (int)$socket; $this->client[$nIdx] = new ioe_client($this, $socket, $this->server[$cIdx], $nIdx, false); $this->server[$cIdx]->trigger("connected", array($nIdx, $this->client[$nIdx]->face)); unset($socket); unset($nIdx); } elseif (isset($this->client[$cIdx])) { $this->client[$cIdx]->read(); } } $success = true; if ($this->doReload) { $this->doReload = false; $files = get_included_files(); foreach($files as $file) { if ($file == __FILE__) continue; if (runkit_lint_file($file)) { runkit_import($file); } else { $success = false; break; } } if ($this->reloadCB) { $args = $this->reloadCBArgs; array_unshift($args, $success); call_user_func_array($this->reloadCB, $args); } } } while(true); } public function disconnectClient($cIdx) { echo "ioe::disconnectClient ".$cIdx."\n"; if (isset($this->client[$cIdx])) { @fclose($this->client[$cIdx]->socket); } unset($this->client[$cIdx]); } public function closeServer($idx) { if (isset($this->server[$idx])) { @fclose($this->server[$idx]->socket); } unset($this->server[$idx]); }}class ioe_server { public $socket, $ioe; public $callback = array(); public $face; public $idx; public function trigger($evt, $args = array()) { if (!isset($this->callback[$evt])) return false; call_user_func_array($this->callback[$evt], $args); } function __construct(&$ioe, &$socket, $idx, $cb) { $this->ioe =& $ioe; $this->socket =& $socket; $this->idx = $idx; $this->callback = $cb; $this->face = new ioe_serverface($this); } function close() { $this->ioe->closeServer($this->idx); }}class ioe_serverface { private $obj; function __construct($obj) { $this->obj =& $obj; } public function close() { return $this->obj->close(); }}class ioe_client { public $ioe; public $socket, $cIdx; public $face; public $server; public function trigger($evt, $args = array()) { if (!isset($this->callback[$evt])) return false; call_user_func_array($this->callback[$evt], $args); } function __construct(&$ioe, &$socket, $server = false, $cIdx, $cb) { $this->ioe =& $ioe; $this->socket =& $socket; $this->server =& $server; $this->cIdx = $cIdx; $this->callback = $cb; $this->face = new ioe_clientface($this); } public function read() { $ret = stream_socket_recvfrom($this->socket, 16384); if (strlen($ret) == 0) { $this->disconnect(); return false; } if ($this->server) { $this->server->trigger("read", array($this->cIdx, $ret)); } else { $this->trigger("read", array($ret)); } return true; } public function write($str) { if (!fwrite($this->socket, $str)) { return false; } else { return true; } } public function disconnect() { if ($this->server) { $this->server->trigger("disconnected", array($this->cIdx)); } else { $this->trigger("disconnected"); } $this->ioe->disconnectClient($this->cIdx); }}class ioe_clientface { private $obj; function __construct($obj) { $this->obj =& $obj; } public function write($str) { return $this->obj->write($str); } public function disconnect() { return $this->obj->disconnect(); }}?>