Доброго времени суток. Стал недавно разбираться с сокетами и понял что я чего-то не понимаю. Начну я с пред истории.
Захотел я реализовать на сайте обмен сообщениями как ВКонтакте, чтобы было все мгновенно. Выбрал для этого сокеты, ведь все таки на дворе 2016 и HTML5 нам позволяет создавать WebSoket.
Начал реализовывать. Создал демона который обрабатывает все подключения (сервер). Задача его постоянно смотреть есть ли у него те пользователи которые должны получить сообщения если есть то он им отправляет. Создав демоно понял что он работает немного не так. Он отвечает только когда ему отправляют запрос на вход, выход, или отправки сообщения. Сам же он не может отвечать. Как сделать так чтобы если есть серверу что сказать то говорил?

Вот код демона:

include('config.php');

$path = SERVER_PATH."/../../class/sql.php";
if(file_exists($path))
	require_once($path);
require_once(SERVER_PATH."/socket_server.php");
//pid-file
//Если в PID файле хранится PID процесса и он активен, то не запускаем копию
if (socket_server::getstatusfwithconsole($server['pidfile'])) {
	exit;//1 уже запущен
}

//system config
error_reporting(E_ALL); //Выводим все ошибки и предупреждения
set_time_limit(0);		//Время выполнения скрипта безгранично
ob_implicit_flush();	//Включаем вывод без буферизации
ignore_user_abort(true);//Выключаем зависимость от пользователя
//system config

ini_set('error_log', $server['wserrorslogfile']);


fclose(STDOUT);
fclose(STDERR);
fclose(STDIN);
$STDIN = @fopen('/dev/null', 'r');
$STDOUT = fopen($server['wsconsolelogfile'], 'ab');
$STDERR = fopen($server['wsconsoleerrfile'], 'ab');

$config = array(
	'pidfile' => $server['pidfile'],
	'offfile' => $server['offfile'],
	'max_connects_from_ip' => $server['max_user'],
	'host' => $server['ip_server'],
    'port' => $server['port']
);
$websocketserver = new socket_server($config);
$websocketserver->start(function($t, $connect, $data){	#message
		
		$uid = array_search($connect, $t->connects);
        $timestart = microtime(true);

		$f = $t->decode($data); 
		if($f['payload'] == "" || $f['payload'] == " ") return;
		$cook = $t::getCookie($t->users[$uid]['Cookie']);
		fwrite($connect,$t->encode($cook['PHPSESSID']));

});

Вот класс сокет сервер

class socket_server{
	
	public $config; 
	public $starttime;
	public $server;
	public $connects;
	public $users;
	public $ips; //Массив IP адресов, запрещаем больше 3х подключений с одного IP
	public $id; //счётчик id подключений
	public $online;
	
	
	public function __construct($config) {
        $this->config = $config;
		$this->connects = array();
		$this->users = array();
		$this->ips = array();
		$this->id = 1; //Начинаем с 1го номера
		$this->online = 0; 
    }
	
	public function start($onMessage=null, $onOpen = null, $onClose = null, $onAll = null) {
		$pidfile = $this->config['pidfile'];
		$offfile = $this->config['offfile'];
		$this->starttime = round(microtime(true),2);
		file_put_contents($pidfile, getmypid());//СОХРАНЯЕМ PID в файле
		$this->server = stream_socket_server("tcp://".$this->config['host'].":".$this->config['port'], $errno, $errstr);
		if (!$this->server) { //Если сокеты не работают
			unlink($pidfile);
			die($errstr. "(" .$errno. ")\n");
		}
		$msgtosend = "";
		while (true) {
			//формируем массив прослушиваемых сокетов:
			$read = $this->connects;
			$read[]= $this->server;
			$write = $except = null;
			if (!stream_select($read, $write, $except, null)) {//ожидаем сокеты доступные для чтения (без таймаута)
				break;
			}
			if (in_array($this->server, $read)) {//есть новое соединение то обязательно делаем handshake
				//принимаем новое соединение и производим рукопожатие:
				if (($connect = stream_socket_accept($this->server, -1)) && $info = $this->handshake($connect)) {
					if(!isset($this->ips[$info['ip']])){
						$this->ips[$info['ip']] = 1; // Одно подключение
					}else{
						$this->ips[$info['ip']]++;
						if($this->ips[$info['ip']]>$this->config['max_connects_from_ip']){
							continue;
						}
					}
					$this->connects[] = $connect;//добавляем его в список необходимых для обработки
					$info['id'] = $this->id++;
					if($this->id > 10000) $this->id = 1; 
					$this->online ++;
					$this->users[] = $info;
					if(is_callable($onOpen)) $onOpen();
				}
				unset($read[ array_search($this->server, $read) ]);
			}
			foreach($read as $connect) {//обрабатываем все соединения
				$data = fread($connect, 100000);
				if (!$data) {
					$uid = array_search($connect, $this->connects);//определяем uid закрытого соединения
					
					if($this->ips[$this->users[$uid]['ip']]==1) unset($this->ips[$this->users[$uid]['ip']]); //Удаляем IP адрес из списка коннектов
					else $this->ips[$this->users[$uid]['ip']]--;

					$this->online --;

					unset($this->users[ array_search($connect, $this->connects) ]); //Удаляем информацию о пользователе соединения
					unset($this->connects[ array_search($connect, $this->connects) ]); //Удаляем ресурс подключения

					fclose($connect);//Закрываем соединение
					if(is_callable($onClose)) $onClose();
					continue;
				}
				if(is_callable($onMessage)) $onMessage($this, $connect, $data);
			}
			if(is_callable($onAll)) $onAll($this);
		}
	}
	
	private function handshake($connect) { //Функция рукопожатия
		$info = array();

		$line = fgets($connect);
		
		$header = explode(' ', $line);
		$info['method'] = $header[0];
		$info['uri'] = $header[1];
		
		//считываем заголовки из соединения
		while ($line = rtrim(fgets($connect))) {
			if (preg_match('/\A(\S+): (.*)\z/', $line, $matches)) {
				$info[$matches[1]] = $matches[2];
			} else {
				break;
			}
		}

		$address = explode(':', stream_socket_get_name($connect, true)); //получаем адрес клиента
		$info['ip'] = $address[0];
		$info['port'] = $address[1];

		if (empty($info['Sec-WebSocket-Key'])) {
			return false;
		}

		//отправляем заголовок согласно протоколу вебсокета
		$SecWebSocketAccept = base64_encode(pack('H*', sha1($info['Sec-WebSocket-Key'] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11')));
		$upgrade = "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" .
			"Upgrade: websocket\r\n" .
			"Connection: Upgrade\r\n" .
			"Sec-WebSocket-Accept:".$SecWebSocketAccept."\r\n\r\n";
		fwrite($connect, $upgrade);
		
		return $info;
	}
	
	public function encode($payload, $type = 'text', $masked = false) {
		$frameHead = array();
		$payloadLength = strlen($payload);

		switch ($type) {
			case 'text':
				// first byte indicates FIN, Text-Frame (10000001):
				$frameHead[0] = 129;
				break;

			case 'close':
				// first byte indicates FIN, Close Frame(10001000):
				$frameHead[0] = 136;
				break;

			case 'ping':
				// first byte indicates FIN, Ping frame (10001001):
				$frameHead[0] = 137;
				break;

			case 'pong':
				// first byte indicates FIN, Pong frame (10001010):
				$frameHead[0] = 138;
				break;
		}

		// set mask and payload length (using 1, 3 or 9 bytes)
		if ($payloadLength > 65535) {
			$payloadLengthBin = str_split(sprintf('%064b', $payloadLength), 8);
			$frameHead[1] = ($masked === true) ? 255 : 127;
			for ($i = 0; $i < 8; $i++) {
				$frameHead[$i + 2] = bindec($payloadLengthBin[$i]);
			}
			// most significant bit MUST be 0
			if ($frameHead[2] > 127) {
				return array('type' => '', 'payload' => '', 'error' => 'frame too large (1004)');
			}
		} elseif ($payloadLength > 125) {
			$payloadLengthBin = str_split(sprintf('%016b', $payloadLength), 8);
			$frameHead[1] = ($masked === true) ? 254 : 126;
			$frameHead[2] = bindec($payloadLengthBin[0]);
			$frameHead[3] = bindec($payloadLengthBin[1]);
		} else {
			$frameHead[1] = ($masked === true) ? $payloadLength + 128 : $payloadLength;
		}

		// convert frame-head to string:
		foreach (array_keys($frameHead) as $i) {
			$frameHead[$i] = chr($frameHead[$i]);
		}
		if ($masked === true) {
			// generate a random mask:
			$mask = array();
			for ($i = 0; $i < 4; $i++) {
				$mask[$i] = chr(rand(0, 255));
			}

			$frameHead = array_merge($frameHead, $mask);
		}
		$frame = implode('', $frameHead);

		// append payload to frame:
		for ($i = 0; $i < $payloadLength; $i++) {
			$frame .= ($masked === true) ? $payload[$i] ^ $mask[$i % 4] : $payload[$i];
		}

		return $frame;
	}

	//-------------------------------------------------------------------------------------------------- ------------

	public function decode($data){
		$unmaskedPayload = '';
		$decodedData = array();

		// estimate frame type:
		$firstByteBinary = sprintf('%08b', ord($data[0]));
		$secondByteBinary = sprintf('%08b', ord($data[1]));
		$opcode = bindec(substr($firstByteBinary, 4, 4));
		$isMasked = ($secondByteBinary[0] == '1') ? true : false;
		$payloadLength = ord($data[1]) & 127;

		// unmasked frame is received:
		if (!$isMasked) {
			return array('type' => '', 'payload' => '', 'error' => 'protocol error (1002)');
		}

		switch ($opcode) {
			// text frame:
			case 1:
				$decodedData['type'] = 'text';
				break;

			case 2:
				$decodedData['type'] = 'binary';
				break;

			// connection close frame:
			case 8:
				$decodedData['type'] = 'close';
				break;

			// ping frame:
			case 9:
				$decodedData['type'] = 'ping';
				break;

			// pong frame:
			case 10:
				$decodedData['type'] = 'pong';
				break;

			default:
				return array('type' => '', 'payload' => '', 'error' => 'unknown opcode (1003)');
		}

		if ($payloadLength === 126) {
			$mask = substr($data, 4, 4);
			$payloadOffset = 8;
			$dataLength = bindec(sprintf('%08b', ord($data[2])) . sprintf('%08b', ord($data[3]))) + $payloadOffset;
		} elseif ($payloadLength === 127) {
			$mask = substr($data, 10, 4);
			$payloadOffset = 14;
			$tmp = '';
			for ($i = 0; $i < 8; $i++) {
				$tmp .= sprintf('%08b', ord($data[$i + 2]));
			}
			$dataLength = bindec($tmp) + $payloadOffset;
			unset($tmp);
		} else {
			$mask = substr($data, 2, 4);
			$payloadOffset = 6;
			$dataLength = $payloadLength + $payloadOffset;
		}

		if (strlen($data) < $dataLength) {
			return false;
		}

		if ($isMasked) {
			for ($i = $payloadOffset; $i < $dataLength; $i++) {
				$j = $i - $payloadOffset;
				if (isset($data[$i])) {
					$unmaskedPayload .= $data[$i] ^ $mask[$j % 4];
				}
			}
			$decodedData['payload'] = $unmaskedPayload;
		} else {
			$payloadOffset = $payloadOffset - 4;
			$decodedData['payload'] = substr($data, $payloadOffset);
		}

		return $decodedData;
	}
	
	
	//-----------------------static function-----------------------
	public static function getstatusfwithconsole($pidfile) {
	  if( file_exists($pidfile) ) {
		$pid = file_get_contents($pidfile);
		//получаем статус процесса
		$status = self::getstatusp($pid);
		if($status['run']) { 
		  //демон уже запущен
		  return true;
		}else{
		  //pid-файл есть, но процесса нет
		  if(!unlink($pidfile)) {
			//не могу уничтожить pid-файл. ошибка
			exit(-1);
		  }
		}
	  }
	  return false;
	}
	public static function getstatusp($pid){
		$result = array ('run'=>false);
		$output = null;

		if (strtoupper(substr(PHP_OS,0,3)) === 'WIN'){ 
			exec("tasklist /fi \"pid eq ".$pid."\"", $output);

			if(count($output)>3){//Если в результате выполнения больше одной строки то процесс есть! т.к. первая строка это заголовок, а третья уже процесс
				$result['run'] = true;
				$result['info'] = $output[3];//строка с информацией о процессе
			} 
		} else {//Если *nix
			exec("ps -aux -p ".$pid, $output);

			if(count($output)>1){//Если в результате выполнения больше одной строки то процесс есть! т.к. первая строка это заголовок, а вторая уже процесс
				$result['run'] = true;
				$result['info'] = $output[1];//строка с информацией о процессе
			}
		}

		return $result;
	}
	public static function getstatusf($pidfile) {

	  if( file_exists($pidfile) ) {

		$pid = file_get_contents($pidfile);
		$output = null;

		if (strtoupper(substr(PHP_OS,0,3)) === 'WIN'){ 
			exec("tasklist /fi \"pid eq ".$pid."\"", $output);

			if(count($output)>3){//Если в результате выполнения больше одной строки то процесс есть! т.к. первая строка это заголовок, а третья уже процесс
				return $pid;
			} else {
			  //pid-файл есть, но процесса нет
			   return -2;
			}
		} else {//Если *nix
			exec("ps -aux -p ".$pid, $output);
			
			if(count($output)>1){//Если в результате выполнения больше одной строки то процесс есть! т.к. первая строка это заголовок, а вторая уже процесс
				return $pid;
			} else {
			  //pid-файл есть, но процесса нет
			   return -2;
			}
		}
	  }
	  return -1;//файла и процесса нет

	}
	public static function getCookie($cookie){
		$result = array();
		$cookie = explode("; ",$cookie);
		if(count($cookie)){
			foreach($cookie as $v){
				$d = explode("=",$v);
				if(count($d)){
					$result[$d[0]] = $d[1];
				}
			}
		}
		return $result;
	}
}