<commit_before><?php declare(strict_types = 1);

namespace Aerys;

use Amp\{ Internal\Producer, Observable, Observer, Postponed };

/**
 * An API allowing responders to buffer or stream request entity bodies
 *
 * Applications are invoked as soon as headers are received and before
 * entity body data is parsed. The $request->body instance allows
 * applications to await receipt of the full body (buffer) or stream
 * it in chunks as it arrives.
 *
 * Buffering Example:
 *
 *     $responder = function(Request $request, Response $response) {
 *          $bufferedBody = yield $request->getBody();
 *          $response->send("Echoing back the request body: {$bufferedBody}");
 *     };
 *
 * Streaming Example:
 *
 *     $responder = function(Request $request, Response $response) {
 *          $payload = "";
 *          $body = $request->getBody()
 *          while (yield $body->next()) {
 *              $payload .= $body->getCurrent();
 *          }
 *          $response->send("Echoing back the request body: {$payload}");
 *     };
 */
class Body extends Observer implements Observable {
    use Producer;

    public function __construct(Observable $observable) {
        $observable->subscribe(function ($data) {
            return $this->emit($data);
        });
        
        parent::__construct($observable); // DO NOT MOVE - preserve order in which things happen
        
        $observable->when(function($e, $result) {
            if ($e) {
                $this->fail($e);
                return;
            }
            
            $when = static function ($e, $bool) use (&$continue) {
                $continue = $bool;
            };
            $string = "";
            $this->next()->when($when);
            while ($continue) {
                $string .= $this->getCurrent();
                $this->next()->when($when);
            }
            
            // way to restart, so that even after the success, the next() / getCurrent() API will still work
            $postponed = new Postponed;
            parent::__construct($postponed->getObservable());
            $postponed->emit($string);
            if ($e) {
                $postponed->fail($e);
            } else {
                $postponed->resolve($result);
            }
            
            $this->resolve($string);
        });
    }<commit_msg>Drain body stream using coroutine<commit_after><?php declare(strict_types = 1);

namespace Aerys;

use Amp\{ Coroutine, Internal\Producer, Observable, Observer, Postponed };
use Interop\Async\Awaitable;

/**
 * An API allowing responders to buffer or stream request entity bodies
 *
 * Applications are invoked as soon as headers are received and before
 * entity body data is parsed. The $request->body instance allows
 * applications to await receipt of the full body (buffer) or stream
 * it in chunks as it arrives.
 *
 * Buffering Example:
 *
 *     $responder = function(Request $request, Response $response) {
 *          $bufferedBody = yield $request->getBody();
 *          $response->send("Echoing back the request body: {$bufferedBody}");
 *     };
 *
 * Streaming Example:
 *
 *     $responder = function(Request $request, Response $response) {
 *          $payload = "";
 *          $body = $request->getBody()
 *          while (yield $body->next()) {
 *              $payload .= $body->getCurrent();
 *          }
 *          $response->send("Echoing back the request body: {$payload}");
 *     };
 */
class Body extends Observer implements Observable {
    use Producer;
    
    public function __construct(Observable $observable) {
        $observable->subscribe(function ($data) {
            return $this->emit($data);
        });
        
        parent::__construct($observable); // DO NOT MOVE - preserve order in which things happen
        
        $observable->when(function($e) {
            if ($e) {
                $this->fail($e);
                return;
            }
            
            $awaitable = new Coroutine($this->drain());
            
            $awaitable->when(function ($e, $string) {
                // way to restart, so that even after the success, the next() / getCurrent() API will still work
                $postponed = new Postponed;
                parent::__construct($postponed->getObservable());
                
                if ($e) {
                    $postponed->fail($e);
                    return;
                }
                
                $postponed->emit($string);
                $postponed->resolve();
            });
            
            $this->resolve($awaitable);
        });
    }
    
    private function drain(): \Generator {
        $string = "";
        while (yield $this->next()) {
            $string .= $this->getCurrent();
        }
        return $string;
    }
{'fraction_non_alphanumeric': 0.12927993359618178, 'fraction_numerical': 0.0004150238638721727, 'mean_word_length': 1.8436578171091444, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 44, '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': '3892224', 'n_tokens_mistral': 1262, 'n_tokens_neox': 1232, 'n_words': 350}