pheanstalk / pheanstalk

src/Socket/SocketSocket.php

<?php


namespace Pheanstalk\Socket;

use Pheanstalk\Contract\SocketInterface;
use Pheanstalk\Exception\ConnectionException;
use Pheanstalk\Exception\SocketException;

/**
 * A Socket implementation using the Sockets extension
 */
class SocketSocket implements SocketInterface
{
    /** @var resource */
    private $socket;

    public function __construct(
        string $host,
        int $port,
        int $connectTimeout
    ) {
        if (!extension_loaded('sockets')) {
            throw new \Exception('Sockets extension not found');
        }

        $this->socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
        if ($this->socket === false) {
            $this->throwException();
        }

        $timeout = [
            'sec' => 0,
            'usec' => 100
        ];

        socket_set_option($this->socket, SOL_TCP, SO_KEEPALIVE, 1);
        socket_set_option($this->socket, SOL_SOCKET,SO_SNDTIMEO, $timeout);
        socket_set_option($this->socket, SOL_SOCKET,SO_RCVTIMEO, $timeout);

        $timeout = microtime(true) + $connectTimeout;
        $address = gethostbyname($host);

        while (microtime(true) < $timeout) {
            socket_clear_error($this->socket);
            if (@socket_connect($this->socket, $address, $port)) {
                return;
            };
            $error = socket_last_error($this->socket);
            if (!($error === SOCKET_EALREADY || $error === SOCKET_EINPROGRESS)) {
                break;
            }
        }

        throw new ConnectionException($error, socket_strerror($error));

The variable $error does not seem to be defined for all execution paths leading up to this point.

    }

    /**
     * Writes data to the socket.
     *
     * @param string $data
     *
     * @return void
     */
    public function write(string $data): void
    {
        $this->checkClosed();
        while (!empty($data)) {
            $written = socket_write($this->socket, $data);
            if ($written === false) {
                $this->throwException();
            }
            $data = substr($data, $written);
        }
    }

    private function throwException()
    {
        $error = socket_last_error($this->socket);
        throw new SocketException(socket_strerror($error), $error);
    }

    private function checkClosed()
    {
        if (!isset($this->socket)) {
            throw new SocketException('The connection was closed');
        }
    }

    /**
     * Reads up to $length bytes from the socket.
     *
     * @return string
     */
    public function read(int $length): string
    {
        $this->checkClosed();

        $buffer = '';
        while (mb_strlen($buffer, '8BIT') < $length) {
            $result = socket_read($this->socket, $length - mb_strlen($buffer, '8BIT'));
            if ($result === false) {
                $this->throwException();
            }
            $buffer .= $result;
        }

        return $buffer;
    }

    public function getLine(): string
    {
        $this->checkClosed();

        $buffer = '';
        // Reading stops at \r or \n. In case it stopped at \r we must continue reading.
        while(substr($buffer, -1, 1) !== "\n") {
            $result = socket_read($this->socket, 1024, PHP_NORMAL_READ);
            if ($result === false) {
                $this->throwException();
            }
            $buffer .= $result;
        }



        return rtrim($buffer);
    }

    /**
     * Disconnect the socket; subsequent usage of the socket will fail.
     */
    public function disconnect(): void
    {
        $this->checkClosed();
        socket_close($this->socket);
        unset($this->socket);
    }
}