Add response callbacks for finished requests

[rbro]

I recently found your blog post and project, and it works extremely well for integrating with rabbitmq and dispatching messages to php-fpm, thanks. I'm starting to look into this more, but related to async requests, is there any sort of integration with the react event loop, so that the code doesn't have to block while the requests are running, and a callback is called when the responses come in? Thanks for your help.

[hollodotme]

@rbro Thank you for your feedback. Although I'm not yet familiar with the react event loop, I had a quick look at it and an integration seems to be possible. I'll further look into this in the next days but would appreciate any help, if you have already experience with react.

[rbro]

Thanks, I'm new to react too, but trying to get more familiar with it. I found these 2 projects which I thought could be useful to see examples of how the socket calls need to work:

https://github.com/CrunchPHP/fastcgi
https://github.com/LawnGnome/fastcgi-react-adapter

[hollodotme]

So here is the plan:

• I'll add a method to register response callbacks (callables) to the request classes.
• Registering response callbacks is optional.
• I'll remove the option to make connections persistent in favour of better timeout handling and fpm process managing (php-fpm doesn't spawn new children on persistent connections, regardless the amount of requests sent through this connection)
• I'll remove the option for keep connection alive (will always be enabled)
• I'll add a method waitForResponses() to the client class. This method...
  • loops as long as there are callbacks that weren't called with a response,
  • reads the response from a stream, only if the stream got data (stream_select),
  • closes the consumed socket and removes callbacks from stack

This achieves:

• You can send multiple async requests
• You can wait for all responses, which will be sent to the particular callbacks of the request
• The order of responding depends on the particular processing duration of the called endpoint, not on the order of requests
• Response reading will have proper timeout behaviour, because every request has its own stream resource
• Both current ways of sending requests and receiving responses will remain as is and benefit from the changes.

[rbro]

Thanks - I think this works. My only question is, will the waitForResponses() method use the react event loop, or will be its own custom loop? As I mentioned, I'm still learning about react, but I believe to use react, it will have to be the react event loop. The idea I'm exploring to tie it all together is:

• run a websocket server using ratchet (https://github.com/ratchetphp/Ratchet)
• as requests are made to the websocket, offload them using fast-cgi-client to call a php-fpm process running separately. This would allow the event loop to continue processing new requests while the existing requests are offloaded.

Again, I'm not completely sure what is involved to use fast-cgi-client inside of react, but I believe it means changing all the socket calls to use https://github.com/reactphp/socket-client.

[hollodotme]

@rbro I should have mentioned that my "plan" above is step one of getting to the react integration part. I now realise that I should have opened another issue for the described new features.

I'll rename this issue accordingly to the new features.

So far this won't include any react components.

Anyway, do you have a repo with your setup which I can fork and then try to integrate fast-cgi-client and provide a PR? If so, I would open an issue in your repo, so that I can reference commits and we can discuss it there. Deal?

[hollodotme]

@rbro I think I found a generic way to integrate into a loop:

<?php

# 1. send an async request
$requestId = $client->sendAsyncRequest($request);

# 2. In a loop ask if a response is available (not blocking)
# This loop could be the react loop
while (true)
{
    # This checks, if the socket stream for this request ID got data (stream_select)
    if ($client->hasResponse($requestId))
    {
        # 3. Read the response (blocking for how long it takes to read the response)
        $response = $client->readResponse($requestId);
        break;
    }

    usleep(20000);
}

With the addition of the above mentioned features the client could then be used with or without an external loop. (I think)

What do you think?

[hollodotme]

And if you're sending multiple requests inside the loop, this could be a solution:

<?php

$requestIds = [];

while (true)
{
    # 1. send multiple async requests (not blocking)
    if ($somethingTellsMeToSendNewRequests)
    {
        $requestIds[] = $client->sendAsyncRequest($request1);
        $requestIds[] = $client->sendAsyncRequest($request2);
    }

    # 2. Check for ready request IDs
    if ($readyRequestIds = $client->getReadyRequestsIds())
    {
        # 3. Read the response (blocking for how long it takes to read the response)
        $responses = $client->readResponses(...$readyRequestIds);

        # Remove responded request IDs
        $requestIds = array_diff($requestIds, $readyRequestIds);

        # do something with responses
    }

    # do something while waiting for responses

    usleep(20000);
}

[rbro]

Thanks for your help with this. I never got back to the project with the react event loop, but I just found another use for it, and was curious to get your thoughts:

As mentioned above, I'm using rabbitmq and dispatching messages to php-fpm very similar to your blog post. One problem I've been running into is some of my requests sent to php-fpm are long-running such as 5-10 minutes. While they are running, php-amqplib has no way to send the heartbeat message, so I lose my connection to the rabbitmq server before the request completes.

What I wanted to do was send the heartbeat message while I'm waiting for the response, so I'm doing this now:

$requestId = $client->sendAsyncRequest($request);

while (true)
{
    if ($client->hasResponse($requestId))
    {
        $response = $client->readResponse($requestId);
        break;
    }

    // check heartbeat message here
    $io = $connection->getIO();
    $io->check_heartbeat();

    usleep(20000);
}

Do you think that works, or is there a better way?

On a side note, the check_heartbeat method above is protected, so you can't directly do the above without changing it to be public in Wire/IO/StreamIO.php in the php-amqplib code. I asked about it here - php-amqplib/php-amqplib#507.

Thanks.