Why the first() method does not resolve the promise upon "end" event?

[mmenozzi]

Hi,
I'm trying to use rakdar/reactphp-csv inside an amphp/amp application and I would like to yield one CSV row at a time. To do so I use this package to transform data events of rakdar/reactphp-csv to ReactPHP promises which in turns are then converted to AMP promises. So I can yield CSV rows, one at a time. Here follows a simple test script:

<?php

require 'vendor/autoload.php';

\Amp\Loop::defer(function () {
    $csv = fopen('https://raw.githubusercontent.com/Rakdar/reactphp-csv/master/examples/country-codes.csv', 'rb');
    $stream = new \React\Stream\ReadableResourceStream($csv, \Amp\ReactAdapter\ReactAdapter::get());
    $reader = new \Rakdar\React\Csv\Reader($stream);

    while ($row = yield \Interop\Amp\Promise\adapt(\React\Promise\Stream\first($reader, 'data'))) {
        echo $row[10] .PHP_EOL;
    }
});

\Amp\Loop::run();

This code works but when the stream reaches the end it throws a "Stream closed" exception.
If I add the following inside the first function it works without exceptions:

$stream->on('end', function () use ($resolve) {
    $resolve(null);
});

So, Why the first() method does not resolve the promise upon "end" event?
What do you think?

[kelunik]

I haven't looked into the actual problem, but Amp supports ReactPHP's promises out of the box, you can just yield React\Promise\Stream\first(...) without the call to Interop\Amp\Promise\adapt.

[mmenozzi]

You're right @kelunik, thank you. Here the updated example:

<?php

require 'vendor/autoload.php';

\Amp\Loop::defer(function () {
    $csv = fopen('https://raw.githubusercontent.com/Rakdar/reactphp-csv/master/examples/country-codes.csv', 'rb');
    $stream = new \React\Stream\ReadableResourceStream($csv, \Amp\ReactAdapter\ReactAdapter::get());
    $reader = new \Rakdar\React\Csv\Reader($stream);

    while ($row = yield \React\Promise\Stream\first($reader, 'data')) {
        echo $row[10] .PHP_EOL;
    }
});

\Amp\Loop::run();

[clue]

This code works but when the stream reaches the end it throws a "Stream closed" exception.
If I add the following inside the first function it works without exceptions: […]
So, Why the first() method does not resolve the promise upon "end" event?

Hi @mmenozzi! While this is indeed expected behavior as per https://github.com/reactphp/promise-stream#first, I think this is a very good question. The idea is that this function resolves with the first value once. If the stream ends (closes) without ever emitting a first value, this means that there can not be a first value, hence the exception.

The rest of you question seems to be directed at another independent package, so I would suggest reporting this upstream and maybe add a link back to this issue.

I hope this helps 👍

[mmenozzi]

Hi @clue,
thank you anyway IMHO if I’m waiting for the first “data” and the stream successfully ends (“end” is different from “close” from what I understand) I would not expect an exception.

Given the current implementation of promise-stream, to implement the example above I have to catch a generic RuntimeException and then ignore the exception only if its message contains “stream closed”. It would make sense if the promise-stream throws a specific exception in such case?

[kelunik]

I think it could make sense to special-case the data event, as this is a stream-specific package, but on the other hand, first() isn't specific to streams, but accepts any event emitter.

@mmenozzi You can also copy&paste this and use that instead of first(). A direct usage might also be possible if you modify the CSV reader making it implement ReadableStreamInterface. Nevermind, that doesn't really work, because streams are restricted to strings in Amp.

[mmenozzi]

Ok @kelunik thank you anyway.

[kelunik]

@mmenozzi You can directly turn the reader into an iterator, that's even easier then.

[mmenozzi]

Yes this is what I would like to achieve. How can I do that without converting stream events to promises?

[clue]

@mmenozzi @kelunik Afaict this discussion does not appear to be related to this project at all, so I would like to ask you to move this discussion to another place as suggested above, thank you.

@mmenozzi Let's get back to your original question: Can you elaborate on your use case where it makes sense to use first() and apply different logic for the end vs close events?

[mmenozzi]

Maybe you mean that I should replace the rakdar/reactphp-csv with a CSV stream parser which implements the Amp\Iterator interface?

[kelunik]

@mmenozzi Basically similar to the code I linked above. This gist might help, but I haven't tested it. If you have further questions, just ping me in #amphp on Freenode or create an issue @ amphp/amp.

[mmenozzi]

Ok @kelunik thank you.

@clue sorry for the off topic discussion. About the original question I don’t understand what is not clear about it.