<commit_before><?php

/*
 * This file is part of the Predis\Async package.
 *
 * (c) <NAME> <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Async\Connection;

use Predis\Command\CommandInterface;
use Predis\Connection\ParametersInterface;
use Predis\Response\Status as StatusResponse;
use Predis\Response\Error as ErrorResponse;
use Clue\Redis\Protocol\Model\StatusReply;
use Clue\Redis\Protocol\Model\ErrorReply;
use Clue\Redis\Protocol\Parser\ResponseParser;
use Clue\Redis\Protocol\Serializer\RecursiveSerializer;
use React\EventLoop\LoopInterface;

class StreamConnection extends AbstractConnection
{
    protected $reader;
    protected $serializer;

    /**
     * {@inheritdoc}
     */
    public function __construct(LoopInterface $loop, ParametersInterface $parameters)
    {
        parent::__construct($loop, $parameters);

        $this->initializeSerializer();
        $this->initializeParser();
    }

    /**
     * Initializes the protocol serializer instance.
     */
    protected function initializeSerializer()
    {
        $this->serializer = new RecursiveSerializer();
    }

    /**
     * Initializes the protocol parser instance.
     */
    protected function initializeParser()
    {
        $this->reader = new ResponseParser();
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $buffer = stream_socket_recvfrom($this->getResource(), 4096);

        if ($buffer === false || $buffer === '') {
            return $this->onError(new ConnectionException($this, 'Error while reading bytes from the server'));
        }

        if ($responses = $this->reader->pushIncoming($buffer)) {
            foreach ($responses as $response) {
                $value = $response->getValueNative();

                if ($response instanceof StatusReply) {
                    $response = StatusResponse::get($value);
                } elseif ($response instanceof ErrorReply) {
                    $response = new ErrorResponse($value);
                } else {<commit_msg>Rename "reader" to "parser" in StreamConnection class.<commit_after><?php

/*
 * This file is part of the Predis\Async package.
 *
 * (c) <NAME> <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Async\Connection;

use Predis\Command\CommandInterface;
use Predis\Connection\ParametersInterface;
use Predis\Response\Status as StatusResponse;
use Predis\Response\Error as ErrorResponse;
use Clue\Redis\Protocol\Model\StatusReply;
use Clue\Redis\Protocol\Model\ErrorReply;
use Clue\Redis\Protocol\Parser\ResponseParser;
use Clue\Redis\Protocol\Serializer\RecursiveSerializer;
use React\EventLoop\LoopInterface;

class StreamConnection extends AbstractConnection
{
    protected $parser;
    protected $serializer;

    /**
     * {@inheritdoc}
     */
    public function __construct(LoopInterface $loop, ParametersInterface $parameters)
    {
        parent::__construct($loop, $parameters);

        $this->initializeResponseParser();
        $this->initializeRequestSerializer();
    }

    /**
     * Initializes the response parser instance.
     */
    protected function initializeResponseParser()
    {
        $this->parser = new ResponseParser();
    }

    /**
     * Initializes the request serializer instance.
     */
    protected function initializeRequestSerializer()
    {
        $this->serializer = new RecursiveSerializer();
    }

    /**
     * {@inheritdoc}
     */
    public function read()
    {
        $buffer = stream_socket_recvfrom($this->getResource(), 4096);

        if ($buffer === false || $buffer === '') {
            return $this->onError(new ConnectionException($this, 'Error while reading bytes from the server'));
        }

        if ($responses = $this->parser->pushIncoming($buffer)) {
            foreach ($responses as $response) {
                $value = $response->getValueNative();

                if ($response instanceof StatusReply) {
                    $response = StatusResponse::get($value);
                } elseif ($response instanceof ErrorReply) {
                    $response = new ErrorResponse($value);
                } else {
{'fraction_non_alphanumeric': 0.11359447004608295, 'fraction_numerical': 0.0018433179723502304, 'mean_word_length': 2.807894736842105, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 27, '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': '5611062', 'n_tokens_mistral': 1129, 'n_tokens_neox': 1098, 'n_words': 286}