clrghouz/app/Classes/Sock/SocketClient.php

496 lines
12 KiB
PHP

<?php
namespace App\Classes\Sock;
use Illuminate\Support\Arr;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Str;
/**
* Class SocketClient
*
* @package App\Classes\Sock
* @property int cps
* @property int speed
*/
final class SocketClient {
private const LOGKEY = 'SC-';
// For deep debugging
private bool $DEBUG = FALSE;
private \Socket $connection;
private string $address_local = '';
private int $port_local = 0;
private string $address_remote = '';
private int $port_remote = 0;
// Our session state
private array $session = [];
private const OK = 0;
private const TIMEOUT = -2;
private const ERROR = -5;
/** @var string Size of our TX buffer */
private const TX_BUF_SIZE = 0xFFFF;
/** @var string Maximum amount of data to send at a time */
private const TX_SIZE = 0xFFFF;
/** @var string Data in the TX buffer */
private string $tx_buf = '';
/** @var string Size of our RX buffer */
private const RX_BUF_SIZE = 0xFFFF;
/** @var string Maximum amount of data to received at a time */
private const RX_SIZE = 0xFFFF;
/** @var string Data in the RX buffer */
private string $rx_buf = '';
public function __construct (\Socket $connection) {
$this->connection = $connection;
if ($this->type === SOCK_STREAM) {
socket_getsockname($connection,$this->address_local,$this->port_local);
socket_getpeername($connection,$this->address_remote,$this->port_remote);
Log::info(sprintf('%s:+ Connection host [%s] on port [%d] (%s)',self::LOGKEY,$this->address_remote,$this->port_remote,$this->type));
}
}
public function __get($key) {
switch ($key) {
case 'address_remote':
case 'port_remote':
return $this->{$key};
case 'cps':
case 'speed':
return Arr::get($this->session,$key);
case 'rx_free':
return self::RX_BUF_SIZE-$this->rx_left;
case 'rx_left':
return strlen($this->rx_buf);
case 'tx_free':
return self::TX_BUF_SIZE-strlen($this->tx_buf);
case 'type':
return socket_get_option($this->connection,SOL_SOCKET,SO_TYPE);
default:
throw new \Exception(sprintf('%s:! Unknown key [%s]:',self::LOGKEY,$key));
}
}
public function __set($key,$value) {
switch ($key) {
case 'cps':
case 'speed':
return $this->session[$key] = $value;
default:
throw new \Exception(sprintf('%s:! Unknown key [%s]:',self::LOGKEY,$key));
}
}
/**
* Create a client socket
*
* @param string $address
* @param int $port
* @return static
* @throws SocketException
*/
public static function create(string $address,int $port): self
{
Log::info(sprintf('%s:+ Creating connection to [%s:%d]',self::LOGKEY,$address,$port));
$sort = collect(['AAAA','A']);
// We only look at AAAA/A records
$resolved = collect(dns_get_record($address,DNS_AAAA|DNS_A))
->filter(function($item) use ($sort) { return $sort->search(Arr::get($item,'type')) !== FALSE; })
->sort(function($item) use ($sort) { return $sort->search(Arr::get($item,'type')); });
if (! $resolved->count())
throw new SocketException(SocketException::CANT_CONNECT,sprintf('%s doesnt resolved to an IPv4/IPv6 address',$address));
$result = FALSE;
foreach ($resolved as $address) {
try {
$try = Arr::get($address,Arr::get($address,'type') === 'AAAA' ? 'ipv6' : 'ip');
if (! $try)
continue;
Log::info(sprintf('%s:- Trying [%s:%d]',self::LOGKEY,$try,$port));
/* Create a TCP/IP socket. */
$socket = socket_create(Arr::get($address,'type') === 'AAAA' ? AF_INET6 : AF_INET,SOCK_STREAM,SOL_TCP);
if ($socket === FALSE)
throw new SocketException(SocketException::CANT_CREATE_SOCKET,socket_strerror(socket_last_error($socket)));
$result = socket_connect($socket,$try,$port);
break;
} catch (\ErrorException $e) {
// If 'Cannot assign requested address'
if (socket_last_error($socket) === 99)
continue;
throw new SocketException(SocketException::CANT_CONNECT,socket_strerror(socket_last_error($socket)));
}
}
if ($result === FALSE)
throw new SocketException(SocketException::CANT_CONNECT,socket_strerror(socket_last_error($socket)));
return new self($socket);
}
/**
* We'll add to our transmit buffer and if doesnt have space, we'll empty it first
*
* @param string $data
* @return void
* @throws \Exception
*/
public function buffer_add(string $data): void
{
$ptr = 0;
$num_bytes = strlen($data);
while ($num_bytes) {
if ($this->DEBUG)
Log::debug(sprintf('%s:- To add [%d] to the TX buffer',self::LOGKEY,$num_bytes));
if ($num_bytes > $this->tx_free) {
if ($this->DEBUG)
Log::debug(sprintf('%s:- TX buffer will be too full, draining...',self::LOGKEY));
do {
$this->buffer_flush(5);
$n = min($this->tx_free,$num_bytes);
$this->tx_buf = substr($data,$ptr,$n);
$num_bytes -= $n;
$ptr += $n;
} while ($num_bytes);
} else {
$this->tx_buf .= substr($data,$ptr,$num_bytes);
$num_bytes = 0;
}
}
if ($this->DEBUG)
Log::debug(sprintf('%s:= TX buffer has [%d] space left',self::LOGKEY,$this->tx_free));
}
/**
* Empty our TX buffer
*
* @param int $timeout
* @return int
* @throws \Exception
*/
public function buffer_flush(int $timeout): int
{
if ($this->DEBUG)
Log::debug(sprintf('%s:+ Emptying TX buffer with [%d] chars, and timeout [%d]',self::LOGKEY,strlen($this->tx_buf),$timeout));
$tm = $this->timer_set($timeout);
$rc = self::OK;
while (strlen($this->tx_buf)) {
$tv = $this->timer_rest($tm);
if (($rc=$this->canSend($tv)) > 0) {
if ($this->DEBUG)
Log::debug(sprintf('%s:- Chars to send [%d]',self::LOGKEY,strlen($this->tx_buf)));
$sent = $this->send(substr($this->tx_buf,0,self::TX_SIZE),0);
if ($this->DEBUG)
Log::debug(sprintf('%s:- Sent [%d] chars [%s]',self::LOGKEY,$rc,Str::limit($this->tx_buf,15)));
$this->tx_buf = substr($this->tx_buf,$sent);
} else {
return $rc;
}
// @todo Enable a delay for slow clients
//sleep(1);
if ($this->timer_expired($tm))
return self::ERROR;
}
$this->tx_purge();
return $rc;
}
/**
* @param int $timeout
* @return int
* @throws \Exception
*/
public function canSend(int $timeout): int
{
$write = [$this->connection];
return $this->socketSelect(NULL,$write,NULL,$timeout);
}
/**
* Close the connection with the client
*/
public function close(): void
{
try {
socket_shutdown($this->connection);
} catch (\ErrorException $e) {
Log::error(sprintf('%s:! Shutting down socket [%s]',self::LOGKEY,$e->getMessage()));
}
try {
socket_close($this->connection);
} catch (\ErrorException $e) {
Log::error(sprintf('%s:! Closing socket [%s]',self::LOGKEY,$e->getMessage()));
}
Log::info(sprintf('%s:= Connection closed with [%s]',self::LOGKEY,$this->address_remote));
}
/**
* We have data in the buffer or on the socket
*
* @param int $timeout
* @return int
* @throws \Exception
*/
public function hasData(int $timeout): int
{
$read = [$this->connection];
return $this->rx_left ?: $this->socketSelect($read,NULL,NULL,$timeout);
}
/**
* Read data, emptying from the RX buffer first, then checking the socket.
*
* @param int $timeout How long to wait for data
* @param int $len The amount of data we want
* @return string
* @throws SocketException
*/
public function read(int $timeout,int $len=1024): string
{
// We have data in our buffer
if ($this->rx_left >= $len) {
if ($this->DEBUG)
Log::debug(sprintf('%s:- Returning [%d] chars from the RX buffer',self::LOGKEY,$len));
$result = substr($this->rx_buf,0,$len);
$this->rx_buf = substr($this->rx_buf,strlen($result));
return $result;
}
if ($timeout AND ($this->hasData($timeout) === 0))
return '';
$buf = '';
try {
switch ($this->type) {
case SOCK_STREAM:
$recv = socket_recv($this->connection,$buf,self::RX_SIZE,MSG_DONTWAIT);
break;
case SOCK_DGRAM:
$recv = socket_recvfrom($this->connection,$buf,self::RX_SIZE,MSG_DONTWAIT,$this->address_remote,$this->port_remote);
break;
default:
throw new SocketException(SocketException::SOCKET_ERROR,sprintf('Unhandled socket type: %s',$this->type));
}
} catch (\Exception $e) {
Log::error(sprintf('%s: - socket_recv Exception [%s]',self::LOGKEY,$e->getMessage()));
throw new SocketException($x=socket_last_error($this->connection),socket_strerror($x));
}
// If we got no data, we'll send whatever is left in the buffer
if ($recv === FALSE) {
// If we have something in the buffer, we'll send it
if ($this->rx_left) {
if ($this->DEBUG)
Log::debug(sprintf('%s:- Network read return an error, returning final [%d] chars from the RX buffer',self::LOGKEY,strlen($this->rx_buf)));
$result = $this->rx_buf;
$this->rx_buf = '';
return $result;
}
Log::error(sprintf('%s:! Request to read [%d] chars resulted in no data',self::LOGKEY,$len));
throw new SocketException($x=socket_last_error($this->connection),socket_strerror($x));
}
// If our buffer is null, see if we have any out of band data.
// @todo We throw an errorexception when the socket is closed by the remote I think.
if (($recv === 0) && is_null($buf) && ($this->hasData(0) > 0) && $this->type === SOCK_STREAM) {
try {
socket_recv($this->connection,$buf,$len,MSG_OOB);
} catch (\Exception $e) {
throw new SocketException($x=socket_last_error($this->connection),socket_strerror($x));
}
}
$this->rx_buf .= $buf;
if ($this->DEBUG)
Log::debug(sprintf('%s:- Added [%d] chars to the RX buffer',self::LOGKEY,strlen($buf)));
// Loop again and return the data, now that it is in the RX buffer
return $this->read($timeout,$len);
}
/**
* Read a character from the remote.
* We'll buffer everything received
*
* @param int $timeout
* @return int
* @throws \Exception
*/
public function read_ch(int $timeout): int
{
if ($this->hasData($timeout) > 0) {
$ch = $this->read($timeout,1);
} else {
return self::TIMEOUT;
}
return ord($ch);
}
public function rx_purge(): void
{
if ($this->DEBUG)
Log::debug(sprintf('%s:+ Discarding [%d] chars from the RX buffer',self::LOGKEY,strlen($this->tx_buf)));
$this->rx_buf = '';
}
/**
* Clear our TX buffer
*/
public function tx_purge(): void
{
if ($this->DEBUG)
Log::debug(sprintf('%s:+ Discarding [%d] chars from the TX buffer',self::LOGKEY,strlen($this->tx_buf)));
$this->tx_buf = '';
}
/**
* Send data to the client
*
* @param $message
* @param int $timeout
* @param null $length
* @return false|int
* @throws \Exception
*/
public function send($message,int $timeout)
{
if ($timeout AND (! $rc=$this->canSend($timeout)))
return $rc;
if ($this->DEBUG)
Log::debug(sprintf('%s:- Sending [%d] chars [%s]',self::LOGKEY,strlen($message),Str::limit($message,15)));
switch ($this->type) {
case SOCK_STREAM:
return socket_write($this->connection,$message,strlen($message));
case SOCK_DGRAM:
return socket_sendto($this->connection,$message,strlen($message),0,$this->address_remote,$this->port_remote);
default:
throw new SocketException(SocketException::SOCKET_ERROR,sprintf('Unhandled socket type: %s',$this->type));
}
}
/**
* Wait for data on a socket
*
* @param array|null $read
* @param array|null $write
* @param array|null $except
* @param int $timeout
* @return int
* @throws \Exception
*/
private function socketSelect(?array $read,?array $write,?array $except,int $timeout): int
{
$rc = socket_select($read,$write,$except,$timeout);
if ($rc === FALSE)
throw new \Exception('Socket Error: '.socket_strerror(socket_last_error()));
if ($this->DEBUG)
Log::debug(sprintf('%s:= Socket select returned [%d] with timeout (%d)',self::LOGKEY,$rc,$timeout),['read'=>$read,'write'=>$write,'except'=>$except]);
return $rc;
}
public function timer_expired(int $timer): int
{
return (time() >= $timer);
}
public function timer_rest(int $timer): int
{
return $timer-time();
}
public function timer_set(int $expire): int
{
return time()+$expire;
}
/**
* See if there is data waiting to collect, or if we can send
*
* @param bool $read
* @param bool $write
* @param int $timeout
* @return int
* @throws \Exception
* @deprecated use canSend or hasData
*/
public function ttySelect(bool $read,bool $write, int $timeout): int
{
if ($this->rx_left) {
if ($this->DEBUG)
Log::debug(sprintf('%s:= We still have [%d] chars in the RX buffer.',self::LOGKEY,$this->rx_left));
return 1;
}
$read = $read ? [$this->connection] : NULL;
$write = $write ? [$this->connection] : NULL;
return $this->socketSelect($read,$write,NULL,$timeout);
}
}