<commit_before><?php

namespace Rvdv\Nntp\Connection;

use Rvdv\Nntp\Command\CommandInterface;
use Rvdv\Nntp\Response\MultiLineResponse;
use Rvdv\Nntp\Response\Response;
use Rvdv\Nntp\Response\ResponseInterface;

class Connection implements ConnectionInterface
{
    private $socket;

    public function connect($host, $port, $secure = false, $timeout = 15)
    {
        $address = gethostbyname($host);
        $url = $this->getSocketUrl($address, $port);

        if (!$this->socket = stream_socket_client($url, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT)) {
            new \RuntimeException(sprintf('Connection to %s:%d failed: %s', $address, $port, $errstr), $errno);
        }

        if ($secure) {
            stream_socket_enable_crypto($this->socket, true, STREAM_CRYPTO_METHOD_TLS_CLIENT);
        }

        stream_set_blocking($this->socket, 1);

        return $this->getSingleLineResponse();
    }

    public function disconnect()
    {
        if (is_resource($this->socket)) {
            stream_socket_shutdown($this->socket, STREAM_SHUT_RDWR);
            fclose($this->socket);
        }
    }

    public function sendCommand(CommandInterface $command)
    {
        $commandString = $command->execute();

        // NNTP/RFC977 only allows command up to 512 (-2 \r\n) chars.
        if (!strlen($commandString) > 510) {
            return \InvalidArgumentException('Failed to write to socket: command exceeded 510 characters');
        }

        if (!@fwrite($this->socket, $commandString."\r\n")) {
            new \RuntimeException('Failed to write to socket');
        }

        $response = $this->getSingleLineResponse();
        $responseHandlers = $command->getResponseHandlers();

        // Check if we received a response expected by the command.
        if (!isset($responseHandlers[$response->getStatusCode()])) {
            throw new \RuntimeException(sprintf(
                'Unexpected response received: [%d] %s',
                $response->getStatusCode(),
                $response->getMessage()
            ));
        }

        $responseHandler = $responseHandlers[$response->getStatusCode()];

        if (!is_callable(array($command, $responseHandler))) {
            throw new \RuntimeException(sprintf('Response handler (%s) is not callable method on given command object', $responseHandler));
        }

        if ($command->isMultiLine()) {
            $response = $this->getMultiLineResponse($response);
        }

        $command->setResponse($response);
        $command->$responseHandler($response);

        return $command;
    }

    protected function getSingleLineResponse()
    {
        if (!$response = @fgets($this->socket, 256)) {
            new \RuntimeException('Failed to read from socket');
        }

        return Response::createFromString($response);
    }

    protected function getMultiLineResponse(ResponseInterface $response)
    {
        $buffer = "";
        while(!feof($this->socket)) {
            $buffer .= @fread($this->socket, 256);

            if (substr($buffer, -3) === ".\r\n") {
                break;
            }
        }

        $lines = explode("\r\n", trim($buffer));
        if (end($lines) === ".") {
            array_pop($lines);
        }

        return new MultiLineResponse($response, $lines);
    }

    protected function getSocketUrl($host, $port)
    {
        if (strpos($host, ':') !== false) {
            // enclose IPv6 addresses in square brackets before appending port
            $host = '[' . $host . ']';
        }

        return sprintf('tcp://%s:%s', $host, $port);
    }
}<commit_msg>Single- and multi-line response are now fetched by a single function.<commit_after><?php

namespace Rvdv\Nntp\Connection;

use Rvdv\Nntp\Command\CommandInterface;
use Rvdv\Nntp\Response\MultiLineResponse;
use Rvdv\Nntp\Response\Response;
use Rvdv\Nntp\Response\ResponseInterface;

class Connection implements ConnectionInterface
{
    private $socket;

    public function connect($host, $port, $secure = false, $timeout = 15)
    {
        $address = gethostbyname($host);
        $url = $this->getSocketUrl($address, $port);

        if (!$this->socket = stream_socket_client($url, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT)) {
            new \RuntimeException(sprintf('Connection to %s:%d failed: %s', $address, $port, $errstr), $errno);
        }

        if ($secure) {
            stream_socket_enable_crypto($this->socket, true, STREAM_CRYPTO_METHOD_TLS_CLIENT);
        }

        stream_set_blocking($this->socket, 0);

        return $this->getResponse();
    }

    public function disconnect()
    {
        if (is_resource($this->socket)) {
            stream_socket_shutdown($this->socket, STREAM_SHUT_RDWR);
            return fclose($this->socket);
        }

        return false;
    }

    public function sendCommand(CommandInterface $command)
    {
        $commandString = $command->execute();

        // NNTP/RFC977 only allows command up to 512 (-2 \r\n) chars.
        if (!strlen($commandString) > 510) {
            return \InvalidArgumentException('Failed to write to socket: command exceeded 510 characters');
        }

        if (!@fwrite($this->socket, $commandString."\r\n")) {
            new \RuntimeException('Failed to write to socket');
        }

        $response = $this->getResponse($command->isMultiLine());
        $responseHandlers = $command->getResponseHandlers();

        // Check if we received a response expected by the command.
        if (!isset($responseHandlers[$response->getStatusCode()])) {
            throw new \RuntimeException(sprintf(
                'Unexpected response received: [%d] %s',
                $response->getStatusCode(),
                $response->getMessage()
            ));
        }

        $responseHandler = $responseHandlers[$response->getStatusCode()];

        if (!is_callable(array($command, $responseHandler))) {
            throw new \RuntimeException(sprintf('Response handler (%s) is not callable method on given command object', $responseHandler));
        }

        $command->setResponse($response);
        $command->$responseHandler($response);

        return $command;
    }

    protected function getResponse($multiLine = false)
    {
        $buffer = "";
        $response = null;

        while(!feof($this->socket)) {
            $buffer .= @fread($this->socket, 256);

            if (!$response && substr($buffer, -2) === "\r\n") {
                $response = Response::createFromString($buffer);

                $lines = explode("\r\n", trim($buffer));
                if (count($lines) > 1) {
                    $buffer = implode("\r\n", array_slice($lines, 1));
                } else {
                    $buffer = "";
                }

                if (!$multiLine) {
                    return $response;
                }
            }

            if (false !== (bool) preg_match("/\r\n\.(\r\n)?$/", $buffer)) {
                $lines = explode("\r\n", trim($buffer));
                if (end($lines) === ".") {
                    array_pop($lines);
                }

                return new MultiLineResponse($response, $lines);
            }
        }
    }

    protected function getSocketUrl($host, $port)
    {
        if (strpos($host, ':') !== false) {
            // enclose IPv6 addresses in square brackets before appending port
            $host = '[' . $host . ']';
        }

        return sprintf('tcp://%s:%s', $host, $port);
    }
}
{'fraction_non_alphanumeric': 0.14095288372706877, 'fraction_numerical': 0.006202982710835423, 'mean_word_length': 2.252360515021459, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3419132', 'n_tokens_mistral': 2168, 'n_tokens_neox': 2073, 'n_words': 453}