<commit_before><?php
namespace Legionth\React\Http;

use Evenement\EventEmitter;
use React\Socket\ServerInterface;
use React\Stream\Stream;
use React\Socket\Connection;
use RingCentral\Psr7;
use RingCentral\Psr7\Response;
use RingCentral\Psr7\Request;
use React\Socket\ConnectionInterface;
use RingCentral;
use React\Stream\ReadableStream;
use React\Promise\Promise;

class HttpServer extends EventEmitter
{

    private $socket;
    private $callback;

    /**
     *
     * @param ServerInterface $socket - the server runs on this socket (ip address and port)
     * @param callable $callback - callback function which returns a RingCentral\Psr7\Response
     */
    public function __construct(ServerInterface $socket, $callback)
    {
        if (!is_callable($callback)) {
            throw new \Exception('The given parametr is not callable');
        }

        $this->socket = $socket;
        $this->callback = $callback;
        $this->socket->on('connection', array(
            $this,
            'handleConnection'
        ));
    }

    /**
     * Handles the requests of a client
     *
     * @param ConnectionInterface $connection - client-server connection stream
     */
    public function handleConnection(ConnectionInterface $connection)
    {
        $bodyBuffer = '';
        $headerCompleted = false;
        $bodyCompleted = false;
        $that = $this;

        $chunkStream = new ReadableStream();
        $chunkedDecoder = new ChunkedDecoder($chunkStream);

        $headerStream = new ReadableStream();
        $headerDecoder = new HeaderDecoder($headerStream);

        $connection->on('data', function ($data) use ($connection, &$headerCompleted, &$bodyBuffer, $that, &$chunkedDecoder, &$headerDecoder, $chunkStream, $headerStream) {
            if (!$headerCompleted) {
                $headerDecoder->on('data', function ($header) use (&$request, &$data, &$headerCompleted) {
                    $request = RingCentral\Psr7\parse_request($header);
                    $data = substr($data, strlen($header));
                    $headerCompleted = true;
                });
                $headerStream->emit('data', array($data));
            }

            if (isset($request)) {
                if ($that->isChunkedEncodingActive($request)) {
                    $chunkedDecoder->on('data', function ($chunk) use (&$bodyBuffer, &$request, $that, $connection) {
                        $bodyBuffer .= $chunk;
                        if (strlen($chunk) == 0) {
                            $that->sendBody($bodyBuffer, $connection, $request);
                        }
                    });
                    $chunkStream->emit('data', array($data));
                } else {
                    $bodyBuffer .= $data;
                    $contentLengthArray = $request->getHeader('Content-Length');

                    if (!empty($contentLengthArray) && strlen($bodyBuffer) == $contentLengthArray[0]) {
                        $that->sendBody($bodyBuffer, $connection, $request);
                    } else if (empty($contentLengthArray) || $contentLengthArray[0] == 0) {
                        $that->handleRequest($connection, $request);
                    }
                }
            }
        });
    }

    /**
     * Checks if the 'Transfer-Encoding: chunked' is set anywhere in the header
     * @param Request $request - user request object containing the header
     * @return boolean
     */
    public function isChunkedEncodingActive(Request $request)
    {
        $transferEncodingArray = $request->getHeader('Transfer-Encoding');

        if (!empty($transferEncodingArray)) {
            foreach ($transferEncodingArray as $value) {
                if ($value == 'chunked') {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Processes the request by the given callback functions and writes the responses on the connection stream
     *
     * @param ConnectionInterface $connection - connection between user and server, the response will be written
     *                                          on this connection
     * @param Request $request - User request to be handled by the callback function
     */
    public function handleRequest(ConnectionInterface $connection, Request $request)
    {
        $callback = $this->callback;

        try {
            $response = $callback($request);

            if ($response instanceof Promise) {
                $this->handlePromise($connection, $response);
                return;
            }

            if (!$response instanceof Response) {
                throw new InvalidResponseTypeException('The returned type of callback function is invalid');
            }
        } catch (\Exception $exception) {
            $connection->write(RingCentral\Psr7\str(new Response(500)));
            $connection->end();
            return;
        }

        $connection->write(RingCentral\Psr7\str($response));
        $connection->end();
    }

    /**
     * Handles an promise
     *
     * @param Connection $connection - connection between server and client
     * @param Promise $promise - Promise returned by the callback function of the server
     */
    private function handlePromise(Connection $connection, Promise $promise)
    {
        $promise->then(function ($response) use ($connection, $promise){
            $responseString = RingCentral\Psr7\str(new Response(500));

            if ($response instanceof Response) {
                $responseString = RingCentral\Psr7\str($response);
            }

            $connection->write($responseString);
            $connection->end();
        },
        function () use ($connection) {
            $connection->write(RingCentral\Psr7\str(new Response(500)));
            $connection->end();
        });
    }

    /**
     * Adds the body to the request before handling the request
     *
     * @param string $body - body to be added to the request object
     * @param ConnectionInterface $connection - client-server connection
     * @param Request $request - Adds the body to this request object
     */
    public function sendBody($body, ConnectionInterface $connection, Request $request)
    {
        $request = $request->withBody(RingCentral\Psr7\stream_for($body));
        $this->handleRequest($connection, $request);
    }
}
<commit_msg>Will convert normal Response to promise<commit_after><?php
namespace Legionth\React\Http;

use Evenement\EventEmitter;
use React\Socket\ServerInterface;
use React\Stream\Stream;
use React\Socket\Connection;
use RingCentral\Psr7;
use RingCentral\Psr7\Response;
use RingCentral\Psr7\Request;
use React\Socket\ConnectionInterface;
use RingCentral;
use React\Stream\ReadableStream;
use React\Promise\Promise;

class HttpServer extends EventEmitter
{

    private $socket;
    private $callback;

    /**
     *
     * @param ServerInterface $socket - the server runs on this socket (ip address and port)
     * @param callable $callback - callback function which returns a RingCentral\Psr7\Response
     */
    public function __construct(ServerInterface $socket, $callback)
    {
        if (!is_callable($callback)) {
            throw new \Exception('The given parametr is not callable');
        }

        $this->socket = $socket;
        $this->callback = $callback;
        $this->socket->on('connection', array(
            $this,
            'handleConnection'
        ));
    }

    /**
     * Handles the requests of a client
     *
     * @param ConnectionInterface $connection - client-server connection stream
     */
    public function handleConnection(ConnectionInterface $connection)
    {
        $bodyBuffer = '';
        $headerCompleted = false;
        $bodyCompleted = false;
        $that = $this;

        $chunkStream = new ReadableStream();
        $chunkedDecoder = new ChunkedDecoder($chunkStream);

        $headerStream = new ReadableStream();
        $headerDecoder = new HeaderDecoder($headerStream);

        $connection->on('data', function ($data) use ($connection, &$headerCompleted, &$bodyBuffer, $that, &$chunkedDecoder, &$headerDecoder, $chunkStream, $headerStream) {
            if (!$headerCompleted) {
                $headerDecoder->on('data', function ($header) use (&$request, &$data, &$headerCompleted) {
                    $request = RingCentral\Psr7\parse_request($header);
                    $data = substr($data, strlen($header));
                    $headerCompleted = true;
                });
                $headerStream->emit('data', array($data));
            }

            if (isset($request)) {
                if ($that->isChunkedEncodingActive($request)) {
                    $chunkedDecoder->on('data', function ($chunk) use (&$bodyBuffer, &$request, $that, $connection) {
                        $bodyBuffer .= $chunk;
                        if (strlen($chunk) == 0) {
                            $that->sendBody($bodyBuffer, $connection, $request);
                        }
                    });
                    $chunkStream->emit('data', array($data));
                } else {
                    $bodyBuffer .= $data;
                    $contentLengthArray = $request->getHeader('Content-Length');

                    if (!empty($contentLengthArray) && strlen($bodyBuffer) == $contentLengthArray[0]) {
                        $that->sendBody($bodyBuffer, $connection, $request);
                    } else if (empty($contentLengthArray) || $contentLengthArray[0] == 0) {
                        $that->handleRequest($connection, $request);
                    }
                }
            }
        });
    }

    /**
     * Checks if the 'Transfer-Encoding: chunked' is set anywhere in the header
     * @param Request $request - user request object containing the header
     * @return boolean
     */
    public function isChunkedEncodingActive(Request $request)
    {
        $transferEncodingArray = $request->getHeader('Transfer-Encoding');

        if (!empty($transferEncodingArray)) {
            foreach ($transferEncodingArray as $value) {
                if ($value == 'chunked') {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Processes the request by the given callback functions and writes the responses on the connection stream
     *
     * @param ConnectionInterface $connection - connection between user and server, the response will be written
     *                                          on this connection
     * @param Request $request - User request to be handled by the callback function
     */
    public function handleRequest(ConnectionInterface $connection, Request $request)
    {
        $callback = $this->callback;

        try {
            $response = $callback($request);

            $promise = $response;
            if (!$promise instanceof Promise) {
                $promise = new Promise(function($resolve, $reject) use ($response){
                    return $resolve($response);
                });
            }

            $this->handlePromise($connection, $promise);
        } catch (\Exception $exception) {
            $connection->write(RingCentral\Psr7\str(new Response(500)));
            $connection->end();
        }
    }

    /**
     * Handles an promise
     *
     * @param Connection $connection - connection between server and client
     * @param Promise $promise - Promise returned by the callback function of the server
     */
    private function handlePromise(Connection $connection, Promise $promise)
    {
        $promise->then(
            function ($response) use ($connection, $promise){
                $responseString = RingCentral\Psr7\str(new Response(500));

                if ($response instanceof Response) {
                    $responseString = RingCentral\Psr7\str($response);
                }

                $connection->write($responseString);
                $connection->end();
            },
            function () use ($connection) {
                $connection->write(RingCentral\Psr7\str(new Response(500)));
                $connection->end();
            }
        );
    }

    /**
     * Adds the body to the request before handling the request
     *
     * @param string $body - body to be added to the request object
     * @param ConnectionInterface $connection - client-server connection
     * @param Request $request - Adds the body to this request object
     */
    public function sendBody($body, ConnectionInterface $connection, Request $request)
    {
        $request = $request->withBody(RingCentral\Psr7\stream_for($body));
        $this->handleRequest($connection, $request);
    }
}

{'fraction_non_alphanumeric': 0.11345460247765407, 'fraction_numerical': 0.0036851183942292615, 'mean_word_length': 2.0514354066985647, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5729719', 'n_tokens_mistral': 3200, 'n_tokens_neox': 3047, 'n_words': 814}