Skip to content

Instantly share code, notes, and snippets.

@egm0121

egm0121/CometServer.php

Last active December 11, 2015 12:48
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save egm0121/4603418 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save egm0121/4603418 to your computer and use it in GitHub Desktop.
Download ZIP
http streaming tracker
Raw
clientCountHandler.php
<?php
/**
* @author giulio
* @copyright 2010
*/
class clientCountHandler {
protected $server;
protected $redis;
public function __construct($serverSingleton){
$this->server= $serverSingleton;
$this->redis = new Redis();
$this->redis->connect('10.64.98.222', 6379);
$this->redis->set('simpleLiveCounter',0);
$this->redis->del('simpleLoggedList');
}
public function onconnect($params){
print_r("\r\n client connected \r\n");
if(is_array($params['reqArray']) ){
$this->redis->lPush('simpleLoggedList',json_encode(array(implode('#',$params['reqArray']['headers']),$params['reqArray']['params'],$params['reqArray']['ip'])));
$this->redis->incr('simpleLiveCounter');
}
//if( in_array('users',$evtArray)){
//$this->server->publishToSocket($params['socket'],array('list'=>$this->getUserList(),'action'=>'userList','evtName'=>'users'));
//}
}
public function onpublish($data){
$actionName = $data['action'].'Action';
print_r("\r\n client publish \r\n");
}
public function onclose($params){
if(is_array($params['reqArray']) ){
$this->redis->lRem('simpleLoggedList', json_encode(array(implode('#',$params['reqArray']['headers']),$params['reqArray']['params'],$params['reqArray']['ip'])),0);
$this->redis->decr('simpleLiveCounter');
}
print_r("\r\n client close \r\n");
}
}
Raw
CometServer.php
<?php
//no limite di tempo
set_time_limit(0);
//error_reporting(E_RECOVERABLE_ERROR);
//include 'ACometResponse.php';
class CometServer {
private $storageIdCount = 0;
private $serverSocket;
private $dispatchQueue;
private $publishQueue;
private $clientList;
private $logPath;
private $dispatchDelay;
private $isDebug = false;
private $debugCount;
private $timeout ;
private $countClient ;
private $emptyClients ;
private $isPubSub ;
private $msgBuffer ;
private $msgId ;
private $eventHandlers;
static private $instance;
static public function getInstance($opts = false){
return self::$instance instanceof self ? self::$instance : self::$instance = new self($opts);
}
private function __clone(){}
private function __construct( $opts){
$bindIp = $opts['ip'] ? $opts['ip'] : '127.0.0.2';
$bindPort = $opts['port']? $opts['port'] : 80;
$this->timeout = $opts['timeout'] ? $opts['timeout'] : 20 ;
$this->isDebug = $opts['debug'] ? true : false ;
$this->serverSocket = socket_create(AF_INET,SOCK_STREAM,SOL_TCP);
//$this->logPath = dirname(__FILE__).DIRECTORY_SEPARATOR.'log.txt';
$this->clientList = array();
$this->publishQueue = new SplQueue();
$this->isPubSub = $opts['pubSub'] ?true:false;
$this->msgBuffer = array();
$this->msgId = 1;
$this->eventHandlers = array('onconnect'=>array(),
'onpublish'=>array(),
'onclose'=>array() );
if(!socket_set_option($this->serverSocket, SOL_SOCKET, SO_REUSEADDR, 1)) {
echo "**".socket_strerror(socket_last_error($this->serverSocket));
exit;
}
if(!socket_bind($this->serverSocket,$bindIp,$bindPort)){
throw new Exception("couldn't bind to ip {$bindIp} port {$bindPort} ");
}
if(!socket_listen($this->serverSocket,SOMAXCONN)){
throw new Exception('couldn\'t listen on the adress/port');
}
socket_set_nonblock($this->serverSocket);
$this->emptyClients[]=$this->serverSocket;
echo "Welcome to micro Comet Socket Server!\n (mode pubSub:{$this->isPubSub})\n";
$this->testLoopNB = 0;
$this->testCn =0;
$this->testStart;
declare(ticks = 1);
// signal handler function
echo "Installing signal handler...\n";
// setup signal handlers
pcntl_signal(SIGTERM, array($this,"sigHandler"));
pcntl_signal(SIGHUP, array($this,"sigHandler"));
pcntl_signal(SIGINT, array($this,"sigHandler"));
pcntl_signal(SIGUSR1, array($this,"sigHandler"));
}
protected function sigHandler($signo)
{
echo "exit----";
var_dump($this->serverSocket);
switch ($signo) {
case SIGTERM:
socket_close($this->serverSocket);
case SIGINT:
socket_close($this->serverSocket);
case SIGHUP:
socket_close($this->serverSocket);
case SIGUSR1:
default:
// handle all other signals
}
var_dump($this->serverSocket);
var_dump(socket_write($this->serverSocket,'00000'));
unset($this->serverSocket);
die();
}
public function runMain(){
while(true){
usleep(800000);// to offload the cpu a bit
if($this->dispatchDelay + 60 < microtime(true)){
$this->logMemUsage();
var_dump(count($this->emptyClients));
//resets the timer
$this->dispatchDelay = microtime(true);
}
// accepts new connections
//$this->acceptConnection();
//check writing sockets
$this->checkReadSocket();
$this->processPublishRequests();
//loop trougth sockets
$this->cleanExpiredClients();
$this->testLoopNb ++;
}
}
private function getStorageId(){
return $this->storageIdCount++;
}
public function acceptConnection(){
$new_socket = socket_accept($this->serverSocket);
$this->emptyClients[] = $new_socket;
}
public function initScriptSocket($new_socket,$reqArray){
$buffer = '';
$responseOut = $this->startChunked().$this->addChunk('<html><body><br>'.str_pad($str,2048,'e')).$this->addChunk($this->injectDomain($reqArray['headers']['Host']));
//$mergeArray = new AppendIterator(array());
//
// foreach($reqArray['evtName'] as $channel){
// if($this->msgBuffer[$channel] ){ $mergeArray->append($this->msgBuffer[$channel]);}
// }
// $mergeArray =iterator_to_array($mergeArray) ;
//
// if(false){
//
// foreach($mergeArray as $msgArray){
// $buffer .= $this->renderMessage($msgArray,$reqArray['params']['token']);
// }
// $responseOut.= $this->addChunk($buffer);
// }
// var_dump("write :\r\n".$responseOut." bytes to ".$new_socket);
socket_write($new_socket,$responseOut);
}
public function cleanExpiredClients(){
foreach($this->clientList as $evtName => &$cByType){
foreach($cByType as $k => &$client){
$time=time();
if($client['ts'] + $this->timeout <= $time ){
$this->trigger('onclose',array('socket'=>array($client['sock'],$client['params']['token']),'reqArray'=>$client));
socket_write($client['sock'],$this->endChunked());
socket_close($client['sock']);
if($this->isDebug){
echo "|-closing connection {$i}... \n ";
}
unset($cByType[$k]);
}
}
}
}
public function checkReadSocket(){
//copy socket_select accept array by ref...
$socketToRead = $this->emptyClients;
if(@socket_select($socketToRead, $write = NULL, $socket4Exception, 0) > 0){
foreach($socketToRead as $key => &$socket){
if($socket == $this->serverSocket){
$this->acceptConnection();
continue;
}
$reqArray = $this->readRequest($socket);
if($reqArray == false && socket_write($socket,$this->addChunk('up')) === false){
echo "\r\nEMPTY close {$socket}\r\n";
foreach($this->clientList as $evtName => &$cByType){
foreach($cByType as $k => &$client){
if($client['sock'] === $socket){
$this->trigger('onclose',array('socket'=>array($client['sock'],$client['params']['token']),'reqArray'=>$client));
unset($this->clientList[$cByType][$k]);
}
}
}
socket_close($socket);
unset($this->emptyClients[$key]);
}
//if the first is a publish we have a publish rqst with some events...
if($reqArray[0]['cType'] == 'publish'){
foreach( $reqArray as $evtArray){
$this->publishMessage($evtArray);
}
socket_write($socket,$this->pubResponse(json_encode($reqArray['params'])));
socket_close($socket);
if($this->isDebug) echo '|new publisher.connection deferred '."\r\n";
unset($this->emptyClients[$key]);
} else {
$reqArray = $reqArray[0];
if($reqArray['evtName'] != ''){
foreach($reqArray['evtName'] as $channel){
$this->clientList[$channel] ? true : $this->clientList[$channel] = array();
$this->clientList[$channel][] = array('sock'=>$socket,
'ts'=>time(),
'params'=>$reqArray['params'],
'headers' => $reqArray['headers'],
'ip' => $reqArray['ip'],
'deferred'=>1);
if($this->isDebug)echo 'registred to '.$channel."\r\n";
$this->initScriptSocket($socket,$reqArray);//initialize connection send js domain settings
$this->trigger('onconnect',array('socket'=>array($socket,$reqArray['params']['token']),'reqArray'=>$reqArray));
}
}
if($this->isDebug) echo '|new client.connection deferred '."\r\n";
}
}
}
}
public function publishMessage($reqArray){
$evtName = $reqArray['params']['evtName'];
if( $this->isDebug ) echo "publish message of type ".$evtName;
$this->trigger('onpublish',$reqArray['params']);
if(!$this->msgBuffer[$evtName] instanceof SplQueue){
$this->msgBuffer[$evtName] = new SplQueue();
}
//update counters and add it to publish queue
$reqArray['params']['__COMET_LIB__'] = $this->msgId++;
$this->publishQueue->enqueue(array('evtName'=>$evtName,'evtData'=>$reqArray['params']));
//update msgBuffer
// $this->msgBuffer[$evtName]->enqueue(array('evtName'=>$reqArray['params']['evtName'],'evtData'=>$reqArray['params']));
if(sizeof($this->msgBuffer[$evtName]) == 5 ){
$this->msgBuffer[$evtName]->dequeue();
}
}
private function injectDomain($str){
return "<script>document.domain ='".$str."'</script>";
}
private function logMemUsage(){
$mem_usage = round(memory_get_usage()/(1024*1024),2) ;
$total = ini_get('memory_limit');
echo "\r\n --MEMORY_USAGE:$mem_usage MB $total -- \r\n";
}
public function addResponseHandler(ACometResponse $obj){
$this->dispatchQueue[get_class($obj)] = $obj;
}
public function renderMessage($msgArray,$token){
$evtName = $msgArray['evtName'];
$params = $msgArray['evtData'];
array_walk_recursive($params,function(&$value){
$value=utf8_encode($value);
});
$jsonData = json_encode($params);
$script ='<script>';
$script.="window.parent.COMET_LIB.serverEvent('{$token}','{$evtName}',{$jsonData});";
//4 test performance tweak
$params['TEST_CLONE']=true;
$jsonData = json_encode($params);
$script.="window.parent.COMET_LIB.serverEvent('{$token}','{$evtName}',{$jsonData});";
$script.='</script>';
return $script;
}
public function runEventDispatcherQueue(){
foreach($this->dispatchQueue as $eventName => $eventObj){
foreach($this->clientList[$eventName] as $socketArray){
if($socketArray != null){
if( $returnValue = $this->renderScript($eventObj,$socketArray) ){
socket_write($socketArray['sock'],$this->addChunk($returnValue));
}
}
}
}
}
public function bind($obj){
$rfObj = new ReflectionObject($obj);
foreach( array_keys($this->eventHandlers) as $evtName ){
if( $rfObj->hasMethod($evtName) ){
$this->eventHandlers[$evtName][]=array(&$obj,$evtName);
}
}
if($this->isDebug){
echo "observer class listening:".$rfObj->getName();
}
}
private function trigger($evtName,Array $params){
foreach($this->eventHandlers[$evtName] as $callback){
call_user_func($callback,$params);
}
}
public function publishToSocket($socketArray,array $jsonArray){
$jsonArray = array('evtName'=>$jsonArray['evtName'],'evtData'=>$jsonArray);
$jsonArray['evtData']['__COMET_LIB__'] = $this->msgId++;
socket_write($socketArray[0],$this->addChunk($this->renderMessage($jsonArray,$socketArray[1])));
}
public function processPublishRequests(){
while(sizeof($this->publishQueue) > 0){
$evt = $this->publishQueue->dequeue();
foreach($this->clientList[$evt['evtName']] as $socketArray){
socket_write($socketArray['sock'],$this->addChunk($this->renderMessage($evt,$socketArray['params']['token'])));
}
}
}
public function readRequest($socket){
$raw_request = socket_read($socket,2048,PHP_BINARY_READ);
if(strlen($raw_request) == 0)return false;
$ip = socket_getpeername($socket , $address = null , $port = null);
$request = explode("\r\n",$raw_request);
if(empty($request[0])){$request = explode("\n",$raw_request);}
// process GET /channel1/channel2/?token=26236&else=boo HTTP/1.1
$initReqLine = explode(' ',$request[0]);
$reqParts = explode('?',$initReqLine[1]);
$onlyEvtPath = array_shift($reqParts);
$queryData = $reqParts;
// headers
$headers = array();
foreach(array_slice($request,1) as $header){
list($k,$v) = explode(':',$header);
$headers[$k]= implode('.',array_slice(explode('.',trim($v)),-2));
}
if($onlyEvtPath == '/favicon.ico')return false;
// channel names
$evtName = explode('/',substr($onlyEvtPath,1));
if($evtName[-1] == '') array_pop($evtName);
if(empty($evtName)){
$cType= null;
} else {
$cType = ($evtName[0] == 'publish') ? 'publish' : 'subscribe';
}
//query params ?...
$evtQueryBuffer = array();
foreach($queryData as $eQuery){
$params = explode('&',$eQuery);
$buff = array();
foreach($params as $val){
list($k,$v) = explode('=',$val);
$buff[$k] = $v;
}
$evtQueryBuffer[]=$buff;
}
$returnValue = array();
foreach($evtQueryBuffer as $event){
$returnValue[] = array('cType'=>$cType,'evtName'=>$evtName,'params'=> $event,'headers'=>$headers,'ip'=>$ip.':'.$port);
}
return $returnValue;
}
public function startChunked(){
$response = "HTTP/1.1 200 OK\r\n";
$response .= "Date: ".gmdate('D, d M Y H:i:s \G\M\T',time())."\r\n";
$response .= "Connection: Keep-Alive\r\n";
$response .= "Transfer-Encoding: chunked\r\n";
$response .= "Access-Control-Allow-Origin: *\r\n";
$response .= "Keep-Alive:timeout=5, max=100\r\n";
$response .= "Content-Type: text/html\r\n\r\n";
return $response;
}
public function pubResponse($body){
$body='ok';
$body_len=strlen($body);
$response = "HTTP/1.1 200 OK\r\n";
$response .= "Host: comet.example.com\r\n";
$response .= "Date: ".gmdate('D, d M Y H:i:s \G\M\T',time())."\r\n";
$response .= "Connection: Close\r\n";
$response .= "Content-Type: text/html\r\n";
$response .= "Content-Length: {$body_len}\r\n\r\n";
return $response.$body;
}
public function addChunk($chunks){
$response = '';
if(is_string($chunks)){
$chunks = array($chunks);
}
foreach($chunks as $value){
$size=dechex(strlen($value));
$response .= "{$size}\r\n";
$response .= "{$value}\r\n";
}
return $response;
}
public function endChunked(){
$response = "0\r\n";
$response .= "\r\n";
return $response;
}
}
?>
Raw
initCometServer.php
<?php
/**
* @author giulio
* @copyright 2010
*/
error_reporting(E_ERROR);
//set_include_path(get_include_path().PATH_SEPARATOR.dirname(dirname(__FILE__)).DIRECTORY_SEPARATOR.'ajax_chat'.DIRECTORY_SEPARATOR.'beta2');
include_once('CometServer.php');
include dirname(__FILE__).DIRECTORY_SEPARATOR.'/extra/clientCountHandler.php';
$server = CometServer::getInstance(array('debug'=>false,'ip'=>'127.0.0.1','port'=>8080,'pubSub'=>true,'timeout'=>10*60));
$handlerObject = new clientCountHandler($server);
$server->bind($handlerObject);
$server->runMain();
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment