React and Guzzle experiment

gistfile1.php

<?php
require 'vendor/autoload.php';

use GuzzleHttp\Stream\BufferStream;
use GuzzleHttp\Stream\AsyncReadStream;
use GuzzleHttp\Ring\Future;

$loop = React\EventLoop\Factory::create();
$dnsResolverFactory = new React\Dns\Resolver\Factory();
$dnsResolver = $dnsResolverFactory->createCached('8.8.8.8', $loop);
$factory = new React\HttpClient\Factory();
$client = $factory->create($loop, $dnsResolver);

function complete_response(
    $request,
    $reactRequest,
    $reactResponse,
    &$ringResponse,
    $loop
) {
    $expectedPump = 0;

    // Note: the user should be able to provide a custom buffer with the
    // save_to client option.
    $buffer = new BufferStream();

    // Get our decorated buffer (for react) and async stream (for consumer)
    list($buffer, $async) = AsyncReadStream::create([
        'buffer' => $buffer,
        'drain'  => function () use ($reactRequest) {
            // Resume the request, though I'm not sure how to "pause" it.
            $reactRequest->handleDrain();
        },
        'pump'  => function ($amount) use ($loop, $buffer, &$expectedPump) {
            // Block until at least the required amount of data has been
            // read.
            echo "Pumping $amount\n";
            $expectedPump = $amount;
            while ($expectedPump > 0) {
                $loop->tick();
            }
        },
        'write' => function ($buffer, $data) use (&$expectedPump, $loop) {
            echo "Writing data\n";
            // handle the pump
            if ($expectedPump) {
                $expectedPump -= strlen($data);
                $expectedPump = max(0, $expectedPump);
                if ($expectedPump == 0) {
                    echo "Amount has been pumped\n";
                }
            }
        }
    ]);

    // todo: Actually return an AsyncStream with a pump and drain function
    // that helps to regulate how fast react reads from the socket.
    $reactResponse->on('data', function ($data) use ($buffer) {
        return $buffer->write($data);
    });

    $ringResponse = [
        'headers' => $reactResponse->getHeaders(),
        'status'  => $reactResponse->getCode(),
        'version' => $reactResponse->getProtocol(),
        'reason'  => $reactResponse->getReasonPhrase(),
        'body'    => $async
    ];

    if (isset($request['then'])) {
        $then = $request['then'];
        $ringResponse = $then($ringResponse) ?: $ringResponse;
    }
}

$adapter = function (array $request) use ($loop, $client) {
    // This is the value we will track in the callbacks and future
    $result = null;

    // todo: Implement more advanced request creation and support for all of
    // the request and request client options.
    $reactRequest = $client->request($request['http_method'], $request['url']);

    // When the response is ready, update the result by reference.
    $reactRequest->on(
        'response',
        function ($reactResponse) use ($request, $reactRequest, &$result, $loop) {
            complete_response($request, $reactRequest, $reactResponse, $result, $loop);
        }
    );

    $reactRequest->end();

    // The future is returned immediately. When accessed or wait() is called,
    // it will block until the response has finished, but the data has not
    // not yet been read.
    return new Future(function() use ($reactRequest, $loop, &$result) {

        if (!$result) {
            // Block until it's done
            do {
                $loop->tick();
            } while (!$result);
        }

        return $result;
    });
};

$response = $adapter([
    'http_method' => 'GET',
    'url'         => 'http://www.google.com',
    'then'        => function (array $response) {
        echo "I'm done!\n";
    }
]);

echo $response['status'] . "\n";
echo $response['reason'] . "\n";

// Do some reads. Because there is a pump function, these are "blocking" (but still done in the event loop).
echo $response['body']->read(1024);
echo $response['body']->read(1024);
echo $response['body']->read(1024);
echo $response['body']->read(1024);

Been thinking how to make this work in an async environment but the then callback already makes it possible. So as long as the user doesn't try to dereference the future $loop->tick() isn't forced and requests aren't waiting on each other 👍 . This contract would work IMHO, for both sync as async environments.

This function will block until the number of outstanding responses are completed.

Blocking or better, putting requests on a queue isn't a bad thing. As long as you don't block the loop 😄 . Did a upload to S3 thing with react a while back. And one of the key things in there was to limit the max number of ongoing requests to S3. (Heck I have no idea what the max is so I put it at 25.)

As for the response completion closure, was going to suggest something similar 👍 .