как выбрать подключенного пользователя и отправить сообщение php socket

Я общаюсь.

Я пишу базовый сокет в php для репликации сообщений, отправленных пользователями чата, пользователи регистрируются в базе данных, когда пользователь отправляет сообщение, это видят все подключенные пользователи, у меня есть форма в административной панели, чтобы отправлять сообщения определенным пользователям, то есть из выбора HTML, затем я выбираю там пользователя и отправляю сообщение, что теперь происходит? сообщение видят все подключенные пользователи во всех комнатах, как его отфильтровать? так что только выбранный пользователь может видеть это? как это реализовать?

<?php
//error_reporting(1);
set_time_limit(0);
//
//
define('CON_IP', '127.0.0.1');
define('CON_PORT', 4000);
//
//
//INICIO KEEP ALIVE
$KTimer   = (20);
$LastTime = time();
//END KEEP ALIVE
//
$null     = NULL; //null var
//
$socket   = socket_create(AF_INET, SOCK_STREAM, SOL_TCP); // CREAR STREAM SOCKET TCP
socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1); //REUTILIZAR PUERTO
socket_bind($socket, CON_IP, CON_PORT); //VINCULA EL SOCKET AL HOST ESPECIFICADO
socket_listen($socket); //ESCUCHAR PUERTO
//
//CREAR Y AÑADIR SOCKETS A LA LISTA
$clients = array(
$socket
);
echo ('running...');
//
//INICIAR CICLO PARA QUE EL SCRIPT NO SE DETENGA
//
while (true) {
//
//MANEJAR MULTIPLES CONEXIONES
$changed = $clients;
//RETORNA LOS RECURSOS DE SOCKET EN EL ARRAY ($changed)
socket_select($changed, $null, $null, 0, 10);
//
//BUSCA UN NUEVO SOCKET
if (in_array($socket, $changed)) {
$socket_new = socket_accept($socket); //ACEPTAR NUEVO SOCKET
$clients[]  = $socket_new; //AÑADIR SOCKET CLIENTE A ARRAY()
//
$header     = socket_read($socket_new, 1024); //LEER DATOS ENVIADOS POR EL SOCKET
//
//HEADER FIX
if (substr($header, 0, 3) == 'GET' || substr($header, 0, 4) == 'POST' || substr($header, 0, 4) == 'OPTI') {
//
//HTML HEADER INCLUDE
socket_write($socket_new, ("HTTP/1.1 200 OK\r\n"));
socket_write($socket_new, ("Server: NStream/5.8.425.10413\r\n"));
socket_write($socket_new, ("Content-Type: text/html; charset=utf-8;\r\n"));
socket_write($socket_new, ("Cache-Control: private\r\n\r\n"));
//
//
socket_write($socket_new, $html);
}
//END HEADER FIX
//
socket_getpeername($socket_new, $RemoteAddr, $RemotePort); //OBTENER DIRECCION IP Y PUERTO DE LOS SOCKETS CONECTADOS
MsgSend('[' . $RemoteAddr . ':' . $RemotePort . ']=>>IC', 'ILC'); //NOTIFICAR NUEVAS CONEXIONES
//
//CREAR UN NUEVO CANAL PARA SOCKETS NUEVOS
$found_socket = array_search($socket, $changed);
unset($changed[$found_socket]);
//CMD PARSE
//si sesion coincide con usuario, mostrar packet
if (!empty($header)) {
$GedCmdId = (explode(':', @explode('"', $header)[1]));
//CMD SY
if ($GedCmdId[0] == 'sy') {
$ParsePaket = (explode('|', preg_replace(array(
"/'/",
"/,/"), array(
NULL,
'|'
), $GedCmdId[1])));
$header     = ("parent.ParsePacket(\"sy:'" . $ParsePaket[0] . "','" . $ParsePaket[1] . "','" . $ParsePaket[2] . "','" . $ParsePaket[3] . "','" . $ParsePaket[4] . "','" . $ParsePaket[5] . "'," . $ParsePaket[6] . "\");");
}
}
//}
//END CMD PARSE
}
//VOID
//
//KA - FIX
if (time() - $LastTime > $KTimer) {
MsgSend('[' . $LastTime . '==>' . time() . ']=>>KA(' . (time() - $LastTime) . ')', 'ILC');
MsgSend("k();");
$LastTime = time();
continue 1;
}
//END KA - FIX
//
//LOOP A TRAVEZ DE TODOS LOS SOCKETS CONECTADOS
foreach ($changed as $changed_socket) {
//check for any incomming data
if (!empty($header)) {
//PREPARAR INFO PARA SER ENVIADA A SOCKET
//prepare data to be sent to client
MsgSend('[' . $RemoteAddr . ':' . $RemotePort . ']=>>MSG', 'ILC');
MsgSend($header); //send data
}
$buf = @socket_read($changed_socket, 1024, PHP_NORMAL_READ);
if ($buf === false) { // VERIFICAR CLIENTES DESCONECTADOS
//
// REMOVER CLIENTE DE ARRAY ($clients)
$found_socket = array_search($changed_socket, $clients);
socket_getpeername($changed_socket, $RemoteAddr);
unset($clients[$found_socket]); //NOTIFICAR NUEVAS DESCONEXIONES
MsgSend('[' . $RemoteAddr . ':' . $RemotePort . ']<<=ID', 'ILC');
}
}
}
//
//CIERRA LA ESCUCHA EN EL SOCKET
socket_close($socket);
function MsgSend($msg, $msgt = 'sock')
{
global $clients;
if ($msgt == 'ILC') {
echo ($msg . "\r\n");
} else {
if (!empty($msg)) {
foreach ($clients as $changed_socket) {
@socket_write($changed_socket, '<script>' . $msg . "</script>\r\n");
}
}
}
return true;
}
//
?>