[http] Link between Connection layer and Request handler

[quazardous]

Hi,

I'm using this kind of code to handle connection timeouts

$loop = React\EventLoop\Factory::create();

$socket = new React\Socket\Server($loop);
$socket->on('connection', function ($conn) use ($loop) {
    $func = function () use ($conn) {$conn->close();};
    $timer = $loop->addTimer(1, $func);
    $conn->on('data', function ($data) use ($loop, &$timer, $func) {
       $timer->cancel();
       $timer = $loop->addTimer(1, $func);
    });
});

It's working OK. But for some GET request with long results (search) I've to cancel the timer from the request handler (because it timeouts).

But I don't know how to connect the two worlds...

[WyriHaximus]

We do, what you're looking for are connection and request timeouts. We have this feature planned for a future release.

[quazardous]

Nice to know.

That said, it would be nice to be able to pass data along connection to request.

Connection could have attributes. And connection could be injected in the request. Or exposing the 'headers' event that knows the 2 objects.

Edit: I've created some kind of dynamic timeout to handle variable length request. So I'm happy with something no too polished...

[WyriHaximus]

What is your use case for that? You have indirect access to all properties on the connection in the PSR-7 request.

[quazardous]

I've created a pseudo stream around mongodb cursor using the function futurTick() to allow the loop to retake controle.
So it's a get request that stream a long response and gives back the hand to the loop every 1000 rows so the server can handle other incoming request. Of course there is no stream reading on the incoming request body and the on('data') "trick" to cancel the timer does note work. So I'd like to pass the timer object to the request layer so I could cancel it for this type of request with some business logic.

EDIT: I've done it somehow (ugly) with Closure::bind() to access private properties...

            $reader = static function & ($object, $property) {
                $value = & \Closure::bind(function & () use ($object, $property) {
                    return $object->$property;
                }, $object, $object)->__invoke();
                return $value;
            };
            
            /** @var \React\Http\Io\StreamingServer $streamingServer */
            $streamingServer = & $reader($server, 'streamingServer');
            /** @var \React\Http\Io\RequestHeaderParser $parser */
            $parser = & $reader($streamingServer, 'parser');
            $listeners = & $reader($parser, 'listeners');
            
            // adding in first position
            \array_unshift($listeners['headers'], function (ServerRequestInterface &$request, ConnectionInterface $conn) {
                $request = $request->withAttribute('conn', $conn);
            });

And I use a \spl_hash($conn) => $whatever map to do the stuff I need in the request handler...

EDIT 2:

I'm putting the fake readable stream to explain, feel free to point any issue in this approach (I'm very new in non blocking coding)

class GeneratorStream implements ReadableStreamInterface
{
    use EventEmitterTrait;
    
    protected $cursor;
    
    protected $closed = false;
    protected $paused = false;
    protected $ticking = false;
    protected $loop;
    protected $chunkSize;
    public function __construct(\Generator $cursor, int $chunkSize = 1000, LoopInterface $loop = null)
    {
        $this->cursor = $cursor;
        $this->chunkSize = $chunkSize;
        $this->loop = $loop ?: Loop::get();
        $this->loop->futureTick([$this, 'handle']);
    }
    
    public function pause()
    {
        if ($this->closed || $this->paused) return;
        $this->paused = true;
    }
    
    public function resume()
    {
        if (!$this->paused || $this->closed) return;
        $this->paused = false;
        $this->handle();  
    }
    
    public function pipe(WritableStreamInterface $dest, array $options = [])
    {
        return Util::pipe($this, $dest, $options);
    }

    public function isReadable()
    {
        return !$this->closed;
    }

    public function close()
    {
        if ($this->closed) return;
        $this->paused = false;
        $this->closed = true;
        $this->emit('close');
        $this->removeAllListeners();
    }
    
    /**
     * @internal
     */
    public function handle()
    {
        if ($this->closed||$this->paused) return;
        for ($i = 0; $this->cursor->valid() && (($this->chunkSize > 0 && $i < $this->chunkSize) || $this->chunkSize == 0); $this->cursor->next(), ++$i) {
            $this->emit('data', [$this->cursor->current()]);
        }
        if ($this->cursor->valid()) {
            $this->loop->futureTick([$this, 'handle']);
        } else {
            $this->emit('end');
            $this->close();
        }
    }

}

[clue]

@quazardous Thanks for bringing this up. I'm not sure I follow exactly what you're trying to achieve here, can you elaborate and/or provide a gist?

Your last gist implements a ReadableStreamInterface. This can be used as a response body using the existing APIs as described here: https://github.com/reactphp/http#streaming-outgoing-response

#324 discusses adding additional attributes to the ServerRequestInterface object to access underlying TLS/HTTPS properties.

#396 discusses ways to match the underlying TCP/IP connection to a HTTP request object by matching the source and destination addresses.

Do you feel there's anything missing here?

[quazardous]

I'll try to reformulate.

The way the read timeout is working depends on (real) streams that the loop can check on.

client(POST/STREAM) ----> Loop(STREAM is alive) ----> response(BODY/STREAM) : OK

But in case of non stream request (Like GET)

client(GET) --X--> Loop(NO STREAM Left !!!) -----> response(Long response wrapped in timer / pseudo stream) : KO !!!!

What I need is a way to profile the timeouts at runtime because I handle different type of request:

• Request A: (Real) Stream to (Real) Stream -> boring
• Request B: Very short classic GET -> boring
• Request C: Very long response on a GET (wrapped with timers/pseudo stream) -> special care

So it would be nice to add advanced timeout capabilities (not just a general-same-timeout-for-all).

Your last gist implements a ReadableStreamInterface. This can be used as a response body using the existing APIs as described here: https://github.com/reactphp/http#streaming-outgoing-response

In this case the on('data') ... timer trick cannot work because there is no stream left to read on the input side (aka Request C).

I'll try to dispatch my thoughts to the linked issues

Do you feel there's anything missing here?

the "advanced at runtime dynamic super timeout" could be discussed further

[clue]

Your last gist implements a ReadableStreamInterface. This can be used as a response body using the existing APIs as described here: https://github.com/reactphp/http#streaming-outgoing-response

In this case the on('data') ... timer trick cannot work because there is no stream left to read on the input side (aka Request C).

The incoming response is indeed buffered by default to make it compatible with PSR-7. As per https://github.com/reactphp/http#streaming-incoming-request, you can explicitly stream the incoming response as well. In this case, you would have full control over managing the incoming stream and any timeouts yourself.

Other that that, I believe this has been answered, so I'm closing this for now. Please come back with more details if this problem persists and we can always reopen this 👍