Am I taking crazy pills? How to yield streaming response from http / browser class interface instead of echo?

[gidzr]

Hey

I've built my owner socket_streaming_client php client in PHP that can yield back to the parent caller, enabling a parent or initiating function to be arbiter of how the live streaming data is handled.. So I know it's gotta be possible with reactPHP.. at least it should.

So I created a client class with the reactPHP methods, which I am calling like this:

///USAGE
$reactHttpClient = new ReactHttpClient(.....);
$output = $reactHttpClient->streamData(); //expecting a generator
foreach($output as $bit){ console($bit);} // ingesting generator live streaming

I thought reactPHP would give me more rigour and options so i tried this, which btw, works using ECHO per reactPHP documentation here: https://reactphp.org/promise-stream/#unwrapreadable

public function streamData() {
     //setup 
      $loop = Factory::create();
      $connector = new Connector($loop);
      $browser = new Browser($loop, $connector)->requestStreaming('POST', $url, $headers, $data)

     //run
      $stream = React\Promise\Stream\unwrapReadable(
         $browser->then(function (Psr\Http\Message\ResponseInterface $response) {
            return $response->getBody();
         })->otherwise(function (Exception $e) {
            console("Error: " . $e->getMessage());
         })
      );
      $stream->on('error', function (Exception $e) {
         echo 'Error: ' . $e->getMessage();
      });

      $stream->on('data', function ($data) {
         echo $data;   //<--- this works.. it is live echo the stream as it comes from the server .. live
      });
    
     $loop->run()

but when i tried to convert it to a generator approach.. like this..

      $stream->on('data', function ($data) {
         yield $data;
      });

it doesn't work..

so I then went down the path of...

   private function streamResponse(ReadableStreamInterface $stream): Generator {
      foreach ($stream as $data) {
         yield $data;
      }
   }

public function streamData() {
 
     .... stuff as above..
    
      return streamResponse($stream);

}

But no - it's doesn't work either...

and i've been getting a hand from copilot and gemini, which have come up with all kinds of crazy, adding promises, promiseInterface, etc.. mixing and matching in all kinds of unsuccessful ways.

The issue is that I cannot see ANY references to YIELD in the reactPHP documentation or examples, nor can I find any examples of people discussing this online using regular searches..

So am I crazy? is no-one else encountering this? and is reactPHP constrained to simply echo results back directly and unable to bubble up a live streaming response via yield to the calling / parent function?

Please help

[clue]

Hey @gidzr, welcome to @reactphp! This sounds like a wild ride, so let me try to clear things up a bit:

I'm not sure what you're trying to achieve exactly and without having a better understanding of the motivation, it sounds like you want to turn a ReadableStreaminterface into a Generator. This is very much possible, but does require some work because you'd have to take the thing into account that is so obvious we often forget about it: time.

ReactPHP is all about event-driven programming, so your program flow would be controlled by events happening as time passes by and consequently allows you to run any concurrent routines asynchronously. In your case, it sounds like you want to have more control over this flow and may want to "wait" (or "block") between receiving any data events. This could probably be done with the following (untested!) snippet:

use React\Stream\ReadableStreamInterface;
use function React\Async\await;
use function React\Promise\Stream\first;

function gen(ReadableStreamInterface $stream): Generator
{
    while ($stream->isReadable()) {
        // simplistic approach to await data
        // doesn't take buffering or close/error events into account
        yield await(first($stream, 'data'));
    }
}

foreach (gen($stream) as $data) {
    var_dump($data);
}

Note this snippet doesn't take buffering or close/error events into account to keep things simple, but I hope it can be helpful as a starting point. Anything beyond this would probably need a much better understanding of what you're trying to achieve exactly (and why) and might be out of scope here.

On top of this, there are a number of oversights in your snippet, so it might also make sense to address these:

       //setup 
-      $loop = Factory::create();
-      $connector = new Connector($loop);
-      $browser = new Browser($loop, $connector)->requestStreaming('POST', $url, $headers, $data)
+      $browser = new Browser();
+      $promise = $browser->requestStreaming('POST', $url, $headers, $data);

The Factory has been deprecated in 2021 (https://github.com/reactphp/event-loop/releases/tag/v1.2.0) and the Browser API has been simplified in 2021 (https://github.com/reactphp/http/releases/tag/v1.5.0). It looks like this could be simplified and cleaned up.

but when i tried to convert it to a generator approach.. like this..

      $stream->on('data', function ($data) {
         yield $data;
      });

I can see that this looks tempting, however the Stream documentation explicitly mentions that return values from an event callback should be avoided and have no effect (https://github.com/reactphp/stream#readablestreaminterface). This invalid usage would be much easier to detect with PHPStan in ReactPHP v3 (reactphp/stream#180).

I hope this helps! 👍 If so, consider supporting this project, for example by becoming a sponsor ❤️

[gidzr]

Hey Christian (@clue)

Thanks heaps for your immaculately fast response.

Fyi on context: I'm building a class that I can call anywhere in my project. It's a client for a custom microservice (on separate secure VPS server) which it connects using TLS/SSL. That's why I'm using the reactPHP connector class to pass the certificates.

eg. Usage for context explanation

////USAGE - as a stream of response
$clientCall = new reactServerClass(IP, Port, ServerRoute, etc)
$streamresults = $clientCall->stream()
foreach($streamresults as $bit){echo "$bit/n";}

////USAGE - as batch return of response
$batchresults = $clientCall->return()
echo "$batchresults/n";

Based on a class I'm building

class reactServerClass(){

  public constuct .. stuff here...

  public stream(){

      $loop = Factory::create();
      $connector = new Connector($loop);
      $browser = new Browser($loop, $connector)->requestStreaming('POST', $url, $headers, $data)

     //run
      $stream = React\Promise\Stream\unwrapReadable(
         $browser->then(function (Psr\Http\Message\ResponseInterface $response) {
            return $response->getBody();
         })
      );
      $stream->on('data', function ($data) {
         echo $data;   //<--- this works.. it is live echo the stream as it comes from the server .. live
      });
      $loop->run()

  }
}

TESTING

Just tried out your suggested approach with this update... AND... IT WORKS!!!!... THANK YOU SO MUCH!

Fyi, I do have a more extensive support framework around this call for errors, timeouts, etc.

   private function gen(ReadableStreamInterface $stream): Generator
   {
       while ($stream->isReadable()) {
         $bit = await(first($stream, 'data'));
         if($bit=="<--END_OF_FILE-->"){break;}else{yield $bit;}
       }
   }
   public function streamData() {
      $this->streaming = 1;      
      $browser = $this->execute_react_client();    
      $stream = React\Promise\Stream\unwrapReadable(
         $browser->then(function (Psr\Http\Message\ResponseInterface $response) {
            return $response->getBody();
         })
      );
      return $this->gen($stream);