It looks like there is currently a bug in PHP:
https://bugs.php.net/bug.php?id=72333

As a temporary solution we could try to limit fwrite data length to 4096 of 8192 bytes, like suggested here:
http://stackoverflow.com/questions/14695247/in-php-openssl-error-messages-error-1409f07f-ssl-routines-ssl3-write-pendin

The name close doesn't imply that it would lose any data. Would not a better name be something like cancel(), abort() or shutdown()?

For example, POSIX function close() flushes all unsaved data before closing a file descriptor.

In Buffer->handleWrite() the result of fwrite is compared to false to check for errors.
In my case it returns 0 (and fills lastError with fwrite(): send of 49 bytes failed with errno=10054 An existing connection was forcibly closed by the remote host.).

The check for (0 === $sent && feof($this->stream)) fails as well because feof($this->stream) returns false.
Removing && feof($this->stream) fixes the problem.

I use the socket server and socket client. When I kill the server process and the client tries to write something, it loops forever with 100% cpu.

For further reference: http://php.net/manual/en/function.fwrite.php#96951

ReactPHP v3 is going to happen! 🎉

We're committed to work on the next major version of ReactPHP. We've started working on this a while ago and believe it's time to finally make this public and give people a chance to see what we're up to and to contribute. In order to show the ongoing development towards ReactPHP v3, we're using dedicated roadmap tickets (like this one) for each component.

Our plans towards Stream v3

• ✅ Create 3.x branch #174
  Once this is created, we can start working on the new v3.0.0 milestone. The default branch will be 3.x and the previous 1.x branch still stay in place at least until v3.0.0 is released.

• ✅ Require PHP 7.1+ and recommend PHP 8.1+ #175
  Time to upgrade! The new v3 will make use of newer language features, so we will upgrade to PHP 7.1+ as a minimum requirement (to run absolutely anywhere) and recommend PHP 8.1+ looking forward (Fibers).

• Type-Safe APIs with PHPStan on max level
  All our public APIs should use native type declarations to avoid invalid API usage and guide IDEs and static analysis tools.

• Remove all optional loop parameters
  In order to introduce Fiber-based APIs and make our APIs easier to use, we're going to make the event loop entirely opaque and remove all optional loop parameters.

• Upgrade to require EventLoop v3
  All upcoming v3 components should require any other ReactPHP components from the same major version or above, so we can deprecate legacy versions in the future. May require upgrade to dev version prior to v3 release (reactphp/event-loop#271).

• Prepare any remaining v1 releases
  Before tagging a v3 release, we should make a clear cut to ease upgrading. We should finish any v1 releases already prepared until v3 is ready.

• Release reactphp/stream v3.0.0
  Planned around mid-2024, approaching our 12th birthday? To ensure a smooth release, we will coordinate the process across all our components.

How you can help

We're optimistic to get the above things done in the near future, so this ticket aims to serve as a basic overview and is subject to change as we progress. For more details, see also the milestone links and any referenced tickets. If you have any additional input for ReactPHP v3, we invite you to join our discussion about the roadmap for the next major version.

Working on the next major version involves a lot of work and we're always looking for sponsors to allow us spending more time on ReactPHP. Check out ReactPHP's sponsors profile and consider supporting the ongoing development ❤️

We'll do our best to keep this ticket updated as we make progress. To keep things organized, let's try to limit the discussions in here and please use new tickets and discussions for input. Help us spread the word! We are excited to move forward together! 🚀

Is there any way to read from the buffer a set number of bytes? Something like http://php.net/manual/en/eventbuffer.read.php .

Or would I have to implement that buffer system myself?

React PHP has support for both bidirectional duplex streams and unidirectional half-duplex streams.

• A unidirectional readable stream should close once it can no longer be read.
• A unidirectional writable stream should close once it can no longer be written to.
• A bidirectional duplex stream currently closes once if can either no longer be read or no longer be written to.

We should add an option to allow bidirectional duplex streams to be half-open. In this mode the readable and writable state would be checked independently.

This is particularly useful for some TCP/IP-based protocols where the client sends a FIN message to the server indicating that it will no longer write any data but is still willing to accept incoming data.

I would like to achieve a data export from a fast-growing database table (assume more than 1m entries). Additionally, I would like to have different CSV files with different data. Imagine orders.csv and order-items.csv. I don't want to pull all the data into memory so I was thinking about streaming the data into CSV files and stream those into a ZIP archive that should be streamed as well. I can't find any documentation about how I can achieve that and I wasn't able to find out by try and error. I completely miss the possibility to stream the CSV file streams into a ZIP file stream.

Is there anything you can point me to to achieve my goal or is it simply not possible? I guess I'm too focused on something that can't work to be able to come up with other ideas...

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

// streamX should be replaced with somthing I can stream into a ZIP stream.
$outputStream1 = new \React\Stream\WritableResourceStream(stream1, $loop);
$csvStream1 = new \Clue\React\Csv\Encoder($outputStream1);

$outputStream2 = new \React\Stream\WritableResourceStream(stream2, $loop);
$csvStream2 = new \Clue\React\Csv\Encoder($outputStream2);

// This is what I'm looking for. I also looked into maennchen/zipstream-php but it requires to have a tmpfile
$zip->addFromStream('file1.csv', $outputStream1);
$zip->addFromStream('file2.csv', $outputStream2);

$loop->run();

foreach ($rows as $row) {
    $csvStream1->write($row);
    $csvStream2->write($row);
}

The documentation currently suggests the following behavior:

• An error event will be emitted when the stream/connection encounters an error (and will then continue to close the connection)
• A close event will be emitted when a stream/connection closes (regardless of whether an error occurred or the stream simply ended)
• An end event will be emitted when the stream/connection reached EOF (end of file or other side closes the connection)

However, the actual stream implementations do not accurately follow this behavior:

• Stream::close() should only emit a close event, no end event.
• Stream::end() is not to be confused with the end event. It should simply call close() once the outgoing buffer has been flushed
• When an error occurs, the Stream should only emit an error and then a close event.
• The Stream should only emit an end and then a close event if the sending side closes (i.e. an empty read event occurs)
• WritableStream does not implement ReadableStreamInterface, so it can never emit a data event and hence no end event
• Possibly many others throughout React's ecosystem

I think we should emphasize these event semantics. This is actually documented already, however we should probably be a bit more verbose and explicit about this.

I'm currently looking into actually fixing all of the above occurrences. This will result in a BC break for anybody who relies on the current buggy behavior - this will not be a BC break for anybody following the documentation.

This means we should probably target a v0.5.0 release here and provide some upgrade instructions. Many library will simply need to upgrade their version requirements, however some may also have to check their event semantics.

I have read the description of WritableStreamInterface here: https://github.com/reactphp/stream#write and it looks very unclear. I don't really understand what is the meaning of return value (true/false) for write() and how I should use it to make sure that all data are written, so I am not writing into closed or broken stream.

Let me quote the docs:

A successful write MUST be confirmed with a boolean true, which means that either the data was written (flushed) immediately or is buffered and scheduled for a future write.
...
If the internal buffer is full after adding $data, then write() SHOULD return false, indicating that the caller should stop sending data until the buffer drains.

This sentences are contradicting: the first one says that write() must return true if the data were saved into a buffer but the second one says that write() can return false in this case. Should I check write()'s return value? Should I consider returning false an error?

Another thing I didn't understand is why write() to a non-writable or closed stream (which is programmer's mistake) is silently ignored instead of throwing an exception. Doesn't this make finding such mistake harder?

Also, I didn't understand how the code is separated between Stream and Buffer classes. They both hold a reference to the underlying stream, they both emit events, they both have writable property... would not it be easier to make it a single class?

Also, the isWritable() method doesn't know anything about the underlying stream:

$loop = Factory::create();
$fd = fopen('./file.txt', 'r');
$stream = new Stream($fd, $loop);
var_dump($stream->isWritable()); // prints true though file is opened only for read

As I understand, isWritable() only checks whether the stream has not been closed so maybe a better name would be isOpened()?

When I try to run this code:

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

    $source = new React\Stream\Stream(fopen('omg.txt', 'r'), $loop);
    $dest = new React\Stream\Stream(fopen('wtf.txt', 'w'), $loop);

    $source->pipe($dest);

    $loop->run();

I get the following error:

[warn] Epoll ADD(1) on fd 6 failed.  Old events were 0; read change was 1 (add); write change was 0 (none): Operation not permitted
[warn] Epoll ADD(1) on fd 7 failed.  Old events were 0; read change was 1 (add); write change was 0 (none): Operation not permitted

The Stream class assumes all stream resources are duplex (r/w) streams, while some may only be readable (e.g. STDIN) or only be writable (e.g. STDOUT).

I'm not considering this to be a bug because this happens to be documented behavior, albeit probably unexpected for many newcomers.

The following gist highlights this problem:

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

// by default, this will open a duplex stream (r/w)
$out = new \React\Stream\Stream(STDOUT, $loop);

// this line is completely optional - only here to make this problem more obvious
// the same issue happens without writing any data
$out->write("hello\n");

$loop->run();

The average user would probably assume this to write out some data (if at all) and then exit.

However, this example does not terminate ever. The Stream class assumes the STDOUT stream is a duplex stream resource and thus keeps waiting for incoming data infinitely.

In the above example, this problem can be worked around by pausing the stream anywhere like this:

$out->pause();

I'm going to argue that this is counter-intuitive and confusing (what exactly is being paused here?).

Note that a similar problem happens with stream resources that are read-only (e.g. STDIN) though it's less apparent.

This ticket aims to serve as a reminder for this hardly documented behavior, for the reference and in order to discuss possible alternatives.

In the context of React v0.4.1 in the Ratchet Websocket library (0.3.1) I encountered the following situation. When the buffer code somehow tries to write to an invalid stream (Buffer.php L83) it will emit the error "tried to write to invalid stream".

Buffer.php

public function handleWrite()
{
   if (!is_resource($this->stream)) {
       $this->emit('error', array(new \RuntimeException('Tried to write to invalid stream.'), $this));
       return;
   }
   ...

The stream will receive this error & emits the error also. Somehow this triggers writing to the buffer again, which then emits the "tried to write to invalid stream" error again, thus creating an infinite loop.

Stream.php

public function __construct($stream, LoopInterface $loop)
{
   ...

   $this->buffer->on('error', function ($error) {
        $this->emit('error', array($error, $this));
        $this->close();
   });

   ...
}

Removing the " $this->emit('error', array($error, $this));" from the Stream __construct fixes the infinite loop, but it seems to be that not the real problem is fixed in this way. I have constructed a piece of code that enables me to reproduce this problem (via websockets). I will try to create a unit test that fails for this piece of code. But I hope this report helps as a start.

While working on #37, I've tested the following simple and quite common code to check possible combinations of readable/writable streams and closing/ending either side:

$stdout = new Stream(STDOUT, $loop);
$stdout->pause();

$stdin = new Stream(STDIN, $loop);
$stdin->pipe($stdout);

Here's what works as expected:

$ php test.php #interactive STDIO
$ echo test | test.php # STDIN ends once receiving input
$ true | php test.php # STDIN ends immediately
$ php test.php < /dev/null # STDIN ends immediately
$ php test.php < /dev/zero # endless STDIN stream
$ php test.php < /dev/urandom #endless STDIN stream
$ php test.php > /dev/null # STDOUT closes when trying to write, thus pausing STDIN

Here's what does not work (on most of my systems, more below):

$ php test.php <&- # no STDIN should error and/or close
$ php test.php >&- # no STDOUT should error and/or close when trying to write

It turns out explicitly passing no handle actually results in a default stream for all STDIO streams (STDIN, STDOUT and STERR). Arguably, this may be a sane default behavior if this default stream would follow expected behavior for closed streams, but unfortunately it does not.

Unfortunately, this default stream behaves just like an endless stream of random data. In fact, it actually is /dev/urandom. This can be verified by checking /proc/{PID}/fd. And it turns out this is not in fact related to ReactPHP at all and can be reproduced with the most simple PHP scripts.

Here's a very simple gist to reproduce this output:

$ LANG=en php -r 'var_dump(!!fstat(STDIN));passthru("ls -o /proc/".getmypid()."/fd");' <&-

This SHOULD show false (because STDIN is not a valid stream) and SHOULD not include file descriptor 0 in the listing.

Note that I could reproduce this in multiple setups, but have seen other setups where this does not occur. As such, I'm opening this ticket to gather more information on why this happens and how this could possibly be avoided.

Unless I'm missing something obvious, it looks like this may boil down to an issue in PHP itself, but I'd rather collect more evidence first.

This bug (#20) keeps killing me - I inadvertently reenable LibEventLoop and lost a few more hours to this bug.

I've pretty complex application which has stable crashes. I couldn't provide any exact code or use case because crashes happens few times in several hours and I still can't catch problematic situation (if it exists at all). But crashes are:

exception 'ErrorException' with message 'fread(): SSL: Connection reset by peer' in /home/projects/Scraper/vendor/react/stream/src/Stream.php:121
Stack trace:
#0 [internal function]: Illuminate\Exception\Handler->handleError(2, 'fread(): SSL: C...', '/home/projects/...', 121, Array)
#1 /home/projects/Scraper/vendor/react/stream/src/Stream.php(121): fread(Resource id #374608, 4096)
#2 [internal function]: React\Stream\Stream->handleData(Resource id #374608, Object(React\EventLoop\LibEventLoop))
#3 /home/projects/Scraper/vendor/react/event-loop/LibEventLoop.php(335): call_user_func(Array, Resource id #374608, Object(React\EventLoop\LibEventLoop))
#4 [internal function]: React\EventLoop\LibEventLoop->React\EventLoop\{closure}(Resource id #374608, 2, NULL)
#5 /home/projects/Scraper/vendor/react/event-loop/LibEventLoop.php(211): event_base_loop(Resource id #113, 1)
#6 /home/projects/Scraper/library/DomainScrapper/Worker/Main.php(146): React\EventLoop\LibEventLoop->run()
#7 /home/projects/Scraper/library/DomainScrapper/Worker/Manager.php(933): DomainScrapper\Worker\Main->run(Array)
#8 /home/projects/Scraper/app/commands/WorkerRun.php(44): DomainScrapper\Worker\Manager::run('Crawler-2', '40/0/0/0')
#9 /home/projects/Scraper/vendor/laravel/framework/src/Illuminate/Console/Command.php(112): WorkerRun->fire()
#10 /home/projects/Scraper/vendor/symfony/console/Symfony/Component/Console/Command/Command.php(253): Illuminate\Console\Command->execute(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#11 /home/projects/Scraper/vendor/laravel/framework/src/Illuminate/Console/Command.php(100): Symfony\Component\Console\Command\Command->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#12 /home/projects/Scraper/vendor/symfony/console/Symfony/Component/Console/Application.php(889): Illuminate\Console\Command->run(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#13 /home/projects/Scraper/vendor/symfony/console/Symfony/Component/Console/Application.php(193): Symfony\Component\Console\Application->doRunCommand(Object(WorkerRun), Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#14 /home/projects/Scraper/vendor/symfony/console/Symfony/Component/Console/Application.php(124): Symfony\Component\Console\Application->doRun(Object(Symfony\Component\Console\Input\ArgvInput), Object(Symfony\Component\Console\Output\ConsoleOutput))
#15 /home/projects/Scraper/crawler(20): Symfony\Component\Console\Application->run()
#16 {main} [] []

Probably question is "Who is reponsible for correct processing this exception, stream itself or library that uses stream"?

The existing possibilities of how a Stream transformer can be implemented are infinite. One of these is the ThroughStream, a simple class that takes what it receives from one side and writes it on the other site. You can add here a filter (or maybe better a transformer?) action between this pass through but is limited to this. To just make the 1 => 1 transformation.

I say filter because as I can see in the implementation, there's no way to really filter and discard a chunk. You can throw an exception but this closes the stream and emits an error, which is not really the point.

The reason for this issue is that, for example, I'd like to implement a transformation for 1 => n. Let's say I receive an array of items and I want to pipe each item isolated. Using this ThroughStream will not be possible for that for the 1 => 1 limitation, which makes me think that I should implement my own Stream transformer.

Building a stream is not easy. Basically, because you have to take in consideration many parts of the architecture (events logic...) and at the end, most of the transformers implemented will be a ThroughStream somehow, each one with its own particular business logic, so extending the ThroughStream would really make sense, having the opportunity of having the default (tested and working) though logic implemented, and having to overwrite only the write method.

That would need as well to make some variables protected as well, like the $writable and some flags.

What do you think?

Hello,

I ran into an interesting issue with this library when attempting to use Devristo/phpws and its builtin websocket client. When data is read from a stream using the Stream class' handleData function, reads seem to be blocking even though stream_set_blocking is called in __construct. I found that replacing line 121's fread with a stream_get_contents resolved my blocking reads issue. I feel like this class should be using the stream_ functions anyways - but it seems like a mix of f* and stream_ functions are used.

Is there a possibility a fix for this can be issued?

The only way I know to reproduce this involves a lot of steps and a specific product (Slack) and protocol (Hybi Websockets). I have tested that this issue exists on multiple boxes using PHP 5.5. I'll outline the steps to reproduce..

1. Sign up for a free Slack team here: https://slack.com/
2. Use the slack API doc site to 'test' the rtm.start call. This will give you a Websocket URL to connect to (must be used within 30 seconds, FYI). It's at the very bottom of the output. https://api.slack.com/methods/rtm.start/test
3. Connect using Devristo/phpws using the URL you got in step 2.

require_once("vendor/autoload.php");                // Composer autoloader

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

$logger = new \Zend\Log\Logger();
$writer = new Zend\Log\Writer\Stream("php://output");
$logger->addWriter($writer);

$client = new \Devristo\Phpws\Client\WebSocket("wss_url_here", $loop, $logger);

$client->on("request", function($headers) use ($logger){
    $logger->notice("Request object created!");
});

$client->on("handshake", function() use ($logger) {
    $logger->notice("Handshake received!");
});

$client->on("connect", function($headers) use ($logger, $client){
    $logger->notice("Connected!");
});

$client->on("message", function($message) use ($client, $logger){
    $logger->notice("Got message: ".$message->getData());
});

$client->open();
$loop->run();

4. Type messages on the Slack web client from one or more users. Note that your PHP app will always be one event behind because it it blocking until the next read comes in.

Again, I know this is kind of a weird example, but it's the only thing I've been using this lib for and I don't have another websocket service I can test. It may be related to how the websocket clients end their frames/lines and how fread expects them to end... I'm not completely sure.

Let me know if you need more information on this. I will be happy to provide what I can.

I am not sure if this is actually an issue, but if you pipe from a stream that has already reached EOF (or closed some other way) pipe will cause the loop to never exit because no events are sent - or will ever be sent - to the destination stream.

It seems that it would be better for pipe to close the destination stream if isReadable is false.

Reproduce:

<?php

require_once __DIR__ . '/vendor/autoload.php';

system('echo Testing > somefile.txt');

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

$readStream = new \React\Stream\Stream(fopen("somefile.txt", "r"), $loop);

$loop->addTimer(1, function () use ($loop, $readStream) {
    $output = new \React\Stream\Stream(STDOUT, $loop);
    $readStream->pipe($output);
});

$loop->run();

So I'm creating a server (via reactphp/socket) and an SSL secureConnection(via reactphp/socket-client)

$creation = 0;

function create($port, $callback)
{
    global $creation;
    if ($creation == 0)
    {
        $loop = React\EventLoop\Factory::create();
        $server = new React\Socket\Server($loop);
        $server->on('connection', function ($conn)
        {
            connectFarside(function($socket) use ($conn)
            {
                $socket->pipe($conn);
                $conn->pipe($socket);
                echo "success pipe";
            });
        });

        $server->listen($port);
        $creation++;
        $callback();

        $loop->run();
    } else
    {
        $creation++;
        $callback();
    }
}

function connectFarside($callback)
{
    $loop = React\EventLoop\Factory::create();
    $tcpConnector = new React\SocketClient\TcpConnector($loop);
    $dnsResolverFactory = new React\Dns\Resolver\Factory();
    $dns = $dnsResolverFactory->createCached('8.8.8.8', $loop);


    $dnsConnector = new React\SocketClient\DnsConnector($tcpConnector, $dns);
    $secureConnector = new React\SocketClient\SecureConnector($dnsConnector, $loop, array(
        'capath' => <CAPATH>
        'local_pk' => '<PK>'
        'local_cert' => '<CERT>'
    ));

    try
    {
        $secureConnector->create('<HOST>',<PORT>)->then(function(React\Stream\Stream $stream) use ($callback)
        {
            $stream->on("error", function()
            {
                echo "SOCKET ERROR";
            });
            $callback($stream);
        }, function($err)
        {
            echo "error connect farside: ". $err;
        });

        $loop->run();
    } catch (Exception $ex)
    {
        die("lala" . $ex);
    }
    //create secure connector
}

create(60000, function()
{
});

The server creation and secure connection did perform successfully.
In my console I also can see "success pipe" because I'm echoing after piping the server and secure connection.

What I'm trying to do is to send an ldapquery to localhost 60000. But It seems it doesnt work.
Nothing happened.

I manage to have a nodejs code, which logically should the same here https://jsfiddle.net/L8w82rto/ . The LDAP Request did return the data for me..

I suppose there was something with the pipe?

This is the continue of problem #48

I updated my gist here https://gist.github.com/tokidoki11/668a3a6f0888c05995b5f45bb2439322

I tried to use ldap_bind, but the process seems to be stuck. The On Connection event is not triggering at all.

What I have done:

1. Try to ldapsearch via another terminal, it works.. (this only for proof of concept, I need to do this at the same terminal)
2. Try to make class of thread, and then run the thread on the create callback

create(60000,function()
{
    $thread = new LDAPSearchThread(<ID>);
    $thread->run();
});
class LDAPSearchThread extends Thread
{

    private $searchID;
    private $searchData;

    public function __construct($searchID)
    {
        $this->searchID = $searchID;
    }

    public function run()
    {
        echo "running thread";
        //creaete ldap coonection and bind //stuck also
    }

}

It's weird though, if I invoke fsockopen, the connection event will trigger.
If I invoke fsockopen then ldap_bind, the connection event will not trigger

Hi everyone,

I have this error with my websocket :

Unable to write to stream: fwrite(): send of 243 bytes failed with errno=32 Broken pipe

this appends a lot and I don't know how to solve it...I googled it but find nothing...so I really need your help.

Here my version :

ratchet/pawl   v0.3.1
ratchet/rfc6455  v0.2.3
react/cache  v0.4.2
react/dns v0.4.13
react/event-loop  v0.4.3
react/promise v2.5.1
react/promise-timer  v1.2.1
react/socket  v0.8.10
react/stream  v0.7.7 

PHP 7.1.11 (cli) (built: Oct 27 2017 14:45:33) ( NTS )
Copyright (c) 1997-2017 The PHP Group
Zend Engine v3.1.0, Copyright (c) 1998-2017 Zend Technologies

Hello,
Can you add an option for the bindto inside streams/sockets ?

This is really useful when you have multiple ips and you want to use other one from the list.
.

Hi, I'm doing a program with childprocess and stream. and at some point I want to clean the console to load new data in the console, I will try doing a simple command with childprocess but I think there must be another method within stream to clean the console.

Probably related to: #133

I had to make this change for the code to also work on Windows: phpstan/phpstan-src@8a7ae4f

Otherwise it silently failed and the loop stopped without writing that file.

1. Is there anything I could change in the original code with pipe() that would make this work?
2. Is there an edge case that would make the previous or current logic not work as expected? I followed some examples but I'm not sure about the code, for me it seems there should be some call like "wait for this stream to end before continuing", but maybe it's fine :)

Thank you!

My original test case:

<?php
require __DIR__ . '/vendor/autoload.php';
$stdoutLogFile = 'test.log';
/*
This works fine - it's writing to my log file
*/
$loop = new \React\EventLoop\StreamSelectLoop();
/*
Nothing gets written to the log file
Also, when starting this up, I get a notice & warning:
PHP Notice:  Event::add(): Epoll ADD(1) on fd 9 failed.  Old events were 0; read change was 1 (add); write change was 0 (none): Operation not permitted in vendor/react/event-loop/ExtEventLoop.php on line 258
PHP Warning:  Event::add(): Failed adding event in vendor/react/event-loop/ExtEventLoop.php on line 258
My Actual script doesn't have this warning, as it pipes from a Process stdout to the Stream. It still has the same problem though (not writing to the file)
*/
$loop = new \React\EventLoop\ExtEventLoop();
$stdout = new React\Stream\Stream(fopen($stdoutLogFile, 'a'), $loop);
$stdout->on(
    'error',
    function ($error) {
        var_dump($error);
    }
);
$loop->addPeriodicTimer(
    1,
    function (React\EventLoop\Timer\TimerInterface $timer) use ($stdout) {
        $stdout->write('Starting...');
    }
);
$loop->run();

I tried the simplier demo code provided in README.md, and that didn't write either (same warnings generated).

Works fine when using 'StreamSelectLoop'.

Test System details:

• CentOS 6.x
• PHP 5.6.15 (Remi RPMs)

event phpinfo output

• Event support => enabled
• Sockets support => enabled
• Debug support => disabled
• Extra functionality support including HTTP, DNS, and RPC => enabled
• OpenSSL support => enabled
• Thread safety support => disabled
• Extension version => 1.11.1
• libevent2 headers version => 2.0.21-stable

Also tried it on a Vagrant box (scotch/box) by doing pecl install event

11:19:34 ERROR [websocket] Connection error occurred Unable to write to stream: fwrite(): send of 8192 bytes failed with errno=32 Broken pipe in /var/www/awex.com.ua/vendor/react/stream/src/Buffer.php line 121 ["connection_id" => 10862,"session_id" => "456846955a323dee318b1236559306","client" => "anon-456846955a323dee318b1236559306"] []

"require": {
        "php": ">=7.1",
        "ext-intl" : "*",
        "doctrine/doctrine-bundle": "^1.6",
        "doctrine/doctrine-fixtures-bundle": "^2.3",
        "doctrine/doctrine-migrations-bundle": "^1.2",
        "doctrine/orm": "^2.5",
        "gos/web-socket-bundle": "^1.8",
        "gregwar/captcha-bundle": "^2.0",
        "h4cc/wkhtmltopdf-amd64": "^0.12.3",
        "h4cc/wkhtmltopdf-i386": "^0.12.3",
        "incenteev/composer-parameter-handler": "^2.0",
        "knplabs/knp-paginator-bundle": "^2.6",
        "knplabs/knp-snappy-bundle": "^1.5",
        "kwn/number-to-words": "^1.3",
        "misd/phone-number-bundle": "^1.2",
        "phpoffice/phpexcel": "^1.8",
        "sensio/distribution-bundle": "^5.0.19",
        "sensio/framework-extra-bundle": "^4.0",
        "sonata-project/admin-bundle": "^3.23",
        "sonata-project/doctrine-orm-admin-bundle": "^3.1",
        "stof/doctrine-extensions-bundle": "^1.2",
        "symfony/monolog-bundle": "^3.1.0",
        "symfony/polyfill-apcu": "^1.0",
        "symfony/swiftmailer-bundle": "^2.3.10",
        "symfony/symfony": "^3.4",
        "twig/extensions": "^1.5",
        "twig/twig": "^1.0||^2.0",
        "yadelivery/name-case-lib": "^1.1"
    },

PHP 7.1.12-1 (cli) (built: Nov 29 2017 10:01:08) ( NTS ) Copyright (c) 1997-2017 The PHP Group Zend Engine v3.1.0, Copyright (c) 1998-2017 Zend Technologies with Zend OPcache v7.1.12-1, Copyright (c) 1999-2017, by Zend Technologies

"name": "react/stream",
"version": "v0.4.6",

"name": "react/socket",
"version": "v0.4.6",

Up until v0.3 this stream component used to have a StreamInterface class that could be used to typehint against all of the existing stream implementations such as Stream and CompositeStream.

This interface was removed with commit 52d6e57 (which was released as part of v0.4).

We still have a ReadableStreamInterface and a WritableStreamInterface which we can individually typehint against. However, PHP does not support typehinting against multiple interfaces. So now that the StreamInterface is gone, there is no way to typehint a stream that is both readable and writable.

It's worth noting that typehinting against Stream still works and is probably what most users do. However, this means that passing a CompositeStream or ThroughStream (or similar) will not work.

Arguably, I personally consider this a pretty common case for many (OSI layer 7) client implementations (such as client implementations for HTTP/redis etc.). Now that the interface is gone, what is the recommended way to pass any stream instance to a client implementation?

I was working on Thruway and created a client that did non-trivial blocking I/O in response to an invocation message. While it was doing this work, more invocation messages came in and buffered on the read side of the stream. The first task completed and "wrote" to the stream. Because there is no attempt to actually write at that moment (even though it would not have blocked), the next command is read and the process begins work on that - leaving data in the write buffer until that is done (or possibly until more messages and responses are completed, depending on read-side message processing).

I was able to fix the problem by calling $this->handleWrite() immediately after $this->loop->addWriteStream($this->stream, [$this, 'handleWrite']); in Buffer.php.

I do suppose this could cause issues with handleWrite because it is possible that the write would block and get an EWOULDBLOCK error (I think), which it currently does not appear to handle.

This change also causes the unit testing to fail.

1) React\Tests\Stream\BufferTest::testWriteReturnsFalseWhenBufferIsFull
Failed asserting that true is false.

/Applications/MAMP/htdocs/thruway_website/vendor/react/stream/tests/BufferTest.php:54

2) React\Tests\Stream\BufferTest::testWriteDetectsWhenOtherSideIsClosed
React\Tests\Stream\CallableStub::__invoke(RuntimeException Object (...), React\Stream\Buffer Object (...)) was not expected to be called more than once.

I have tried this with both StreamSelectLoop and LibEventLoop and get the same behavior. (This buffering buffers and the fix above fixes it.)

While working on react/child-process, I noticed occasional, but reproducible SEGFAULTs for really old PHP versions (< 5.4). I've managed to work around this in reactphp/child-process#29, but this actually boils down to an issue in this component when instantiating a new Stream with a pipe stream. Here's the gist of the broken code:

$s = popen('echo test', 'r');

stream_set_blocking($s, 0);
stream_set_read_buffer($s, 0);

$data = stream_get_contents($s, 8000);
var_dump($data);

Or the same issue when using the STDIN pipe:

$s = STDIN;

stream_set_blocking($s, 0);
stream_set_read_buffer($s, 0);

$data = stream_get_contents($s, 8000);
var_dump($data);

Apparently, this reads beyond the resource limits and returns a corrupted string value where strlen($data) === -1.

I am using
-windows 7 64 bit
-php 7.1.9

PHP fatal error: uncaugt RuntimeException: unable to set stream resource to non-blocking mode in ...../vendor/react/stream/src/WritableResourceStream.php:34

look here, "stream_set_blocking doesn't work"
https://raw.githubusercontent.com/php/php-src/PHP-7.1/NEWS

please help. Thank you

Hi everyone! Fairly new to ReactPHP, but it's pretty cool from what I've seen thus far. I find myself in a situation where unfortunately my working knowledge of ReactPHP isn't quite enough yet to solve my problem. I'm hoping this is the correct component in which to post this in ... if not, let me know!

NOTE: I know this isn't really an "issue", but more of a question. I hope that is ok.

Our customer is sending us data in "bundles" of data that look sort of like this:

MESSAGE1 ... BLAH BLAH.... FIELDS HERE.... YADA YADA.... ^M
MESSAGE2 ... BLAH BLAH.... FIELDS HERE.... YADA YADA.... ^M
...

*** Yes, those are carriage returns, not linefeeds, so the data all comes in as one big line.

Now, the way we used to do it with a simple socket server over a VPN. They would connect, send us the stuff, and then disconnect. No biggie there.

However, now they want an "always on" socket connection" (don't ask), where they can send us data without the overhead of a connection, etc.

So in other words, the socket connection is always open, and data could come in at any time. We may get some data one minute, and then here may be a pause of some time before we receive another bundle of data.

For each bundle of data we get, we need to send them a message back acknowledging receipt of the bundle. The good news is that the "bundle" of data does have a footer that I can key off of to denote the end of the bundle.

I'm guessing that a Duplex stream would be the way to go, whereby we both receive and send data on the same stream (in this case using a TCP socket).

However, I'm not quite sure how best to approach this using ReactPHP, as again, I'm still very new to it. Any guidance or suggestions would be greatly appreciated!

Hi, please can you let me know if there is something wrong with how I am using the library?

Example code:

<?php
require("vendor/autoload.php");

$loop = React\EventLoop\Factory::create();
$stream = new React\Stream\Stream(fopen("php://memory", "w+"), $loop);

$stream->on("data", function($data) {
    echo "Got data: $data" . PHP_EOL;
});

$loop->addPeriodicTimer(1, function() use($stream) {
    echo "Tick." . PHP_EOL;
    $stream->write("Hello, world!");
});

$loop->run();

Output:

php sandbox.php 
Tick.
Tick.
Tick.
Tick.
Tick.
Tick.
Tick.
Tick.
...

Notice there is no "Got data" being displayed, ever.

This library currently cannot be used on Windows OS because of exceptions which are thrown, like here. However, if I comment out such throwns in your library, following code works, albeit extremly slowly

<?php

use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\Factory;
use React\Http\Response;
use React\Http\Server;
use React\Stream\ReadableResourceStream;
use React\Stream\WritableResourceStream;

require __DIR__ . '/../vendor/autoload.php';

$loop = Factory::create();

$server = new Server(function (ServerRequestInterface $request) {
    return new Response(
        200,
        array(
            'Content-Type' => 'text/plain'
        ),
        "Hello world\n"
    );
});

$t1 = time();

\React\Stream\Util::pipe(
    new ReadableResourceStream(fopen("C:\Downloads\Rogue One A Star Wars Story 2016 x264 HDTS AAC(Deflickered)-DDR\sample-Rogue One A Star Wars Story 2016 x264 HDTS AAC(Deflickered)-DDR.mkv", 'r'), $loop),
    new WritableResourceStream(fopen('file', 'w'), $loop)
)->on('end', function() use ($t1) {
    echo 'finished in '.(time() - $t1);
});

$socket = new \React\Socket\Server(isset($argv[1]) ? $argv[1] : '0.0.0.0:0', $loop);
$server->listen($socket);

echo 'Listening on ' . str_replace('tcp:', 'http:', $socket->getAddress()) . PHP_EOL;

$loop->run();

I think it's better to allow usage of this library on Windows OS, even if it means it will be very slow. For some use cases speed in kilobytes might be enough.

Let's face it, this project is stable and has been used in production for years

However, we're currently following a v0.X.Y release scheme (http://sentimentalversioning.org/).

We should finally make this explicit and fully adhere to SemVer and release a stable v1.0.0.

To a large extend, a stable v1.0.0 helps making BC breaks more explicit and thus the whole project more reliable from a consumer perspective. This project is actively maintained and has received some major updates in the last weeks and has some major updates planned in the next weeks. Given our current versioning scheme, we'd like to ensure all anticipated BC breaks will be merged before the planned v1.0.0 release.

As such, I've set up a roadmap that enlists only the major changes for each version among with planned release dates towards a stable v1.0.0 release:

v0.4.5 ✅

• Released 2016-11-13
• Major performance improvements
• Infinite read buffer

v0.4.6 ✅

• Released 2017-01-25
• close event semantics
• Standalone Buffer

v0.5.0 ✅

• Released 2017-03-08
• end event semantics
• Strict / explicit event semantics and parameters
• pipe() semantics

v0.6.0 ✅

• Released 2017-03-26
• Read-/Write-only resources
• Enforce non-blocking

v0.7.0 ✅

• Released 2017-05-04
• Reduce public API

v1.0.0

• Planned 2017-?
• No new changes planned, this should merely mark the previous release as "stable"

This ticket aims to serve as a basic overview and does not contain every single change. Please also see the milestone links and the CHANGELOG for more details.

Obviously, this roadmap is subject to change and I'll try to keep it updated as we progress. In order to avoid cluttering this, please keep discussion in this ticket to a minimum and consider reaching out to us through new tickets or Twitter etc.

In Util::pipe() the default behavior is to connect the sources end event with the destination's end event, but it does not forward any potential $data.

        // forward end event from source as $dest->end()
        $end = isset($options['end']) ? $options['end'] : true;
        if ($end) { 
            $source->on('end', $ender = function () use ($dest) { 
                $dest->end();
            });
            $dest->on('close', function () use ($source, $ender) { 
                $source->removeListener('end', $ender);
            });
        } 

I've had to wrap Util::pipe() with my own function that basically mixes this behavior with the forwardEvents util call

    public function pipe(WritableStreamInterface $dest, array $options = array())
    { 
        $options['end'] = FALSE;

        //default pipe behavior throws away $data for end events
        $this->on('end', $ender = function () use($dest) { 
            call_user_func_array([$dest, 'end'], func_get_args());
        });

        $source = $this;
        $dest->on('close', function () use ($source, $ender) { 
            $source->removeListener('end', $ender);
        });

        return Util::pipe($this, $dest, $options);
    } 

Should end() not be used this way? I'm writing a pipe that will collect data from streams and concat into one file, when this "destination" pipe gets end() from its source it will concat all the info into one file and emit its own end with the name of the file as the only value of $data (wrapped in an array). I could rewrite it to emit 'data' one time, but it seems more natural for a pipe component that waits for upstream pipes to send end that it would also send end.

Editing the Util.php file also gives me the result I was expecting:

        // forward end event from source as $dest->end()
        $end = isset($options['end']) ? $options['end'] : true;
        if ($end) { 
            $source->on('end', $ender = function ($data=null) use ($dest) {   //change 1 of 2
                $dest->end($data);                                       //change 2 of 2
            });
            $dest->on('close', function () use ($source, $ender) { 
                $source->removeListener('end', $ender);
            });
        } 

Would be nice isn't it? But I know, I know. It's complicated, because of PHPUnit 8.

But here is something interesting. I tried the following:

diff --git a/composer.json b/composer.json
index f6faa66..fa65403 100644
--- a/composer.json
+++ b/composer.json
@@ -9,8 +9,8 @@
         "evenement/evenement": "^3.0 || ^2.0 || ^1.0"
     },
     "require-dev": {
-        "phpunit/phpunit": "^6.4 || ^5.7 || ^4.8.35",
-        "clue/stream-filter": "~1.2"
+        "clue/stream-filter": "~1.2",
+        "symfony/phpunit-bridge": "^5.0"
     },
     "autoload": {
         "psr-4": {

And then that:

$ composer install
$ vendor/bin/simple-phpunit

# Here is the docker version if you prefer
$ docker run --rm -v $(pwd):/app -w /app composer:latest composer install
$ docker run --rm -v $(pwd):/app -w /app php:7.4 vendor/bin/simple-phpunit

And here is the output. Not bad for a code with no modification!

WARNINGS!
Tests: 176, Assertions: 259, Warnings: 14, Skipped: 33.
$ echo $?
0 # no test suite modification no error with PHP 7.4 ✨

I hope it helps. ✌️

What is the best way to go from React\Stream\ReadableStreamInterface to Psr\Http\Message\StreamInterface\?

I found https://github.com/WyriHaximus/reactphp-psr7-stream-converter, but it is archived.

Thanks for your help!

Hello, i am running loop as a single process, and trying to read STDIN for commands parsing.
$in = new ReadableResourceStream(fopen('php://stdin', 'rb'), $this->loop);
$in->on('data', function ($data) use ($in) {
$commands = explode('.', $data);
$in->close();
});

All things are good, but for me it works for a single time. I am using a periodic timer for wrapping this process. Is it correct ?

In Buffer there's event full-drain but it's not proxified on Stream level.

I'm trying to read from a pipe. If that closes, reopening the pipe should be attempted. I've tried to do this by subsclassing the Stream:

public function handleClose()
{
    parent::handleClose();

    $this->stream = fopen($file, 'r');
    $this->resume();
}

However, the loop terminates. Any hints would be appreciated.

Hello everyone, I recently looked at the source code. i found that there are some Synchronous code, Sync Read, Sync Write

this means that process is blocked when reading datagram from the kernel, event if it is very short;

According to Asynchronous I/O , Reactphp seems to be non-standard.

May be we should replace it with aio_write/aio_read or libeio/IOCP like nodejs.

Not sure which subsection this falls

[2016-09-17 00:40:06] Dramiel.ERROR: exception 'ErrorException' with message 'fwrite(): SSL operation failed with code 1. OpenSSL Error messages: error:140D00CF:SSL routines:SSL_write:protocol is shutdown' in /home/Dramiel/mamba/vendor/react/stream/src/Buffer.php:86
Stack trace:
#0 [internal function]: React\Stream\Buffer->handleWrite(Resource id #177, Object(React\EventLoop\StreamSelectLoop))
#1 /home/Dramiel/mamba/vendor/react/event-loop/src/StreamSelectLoop.php(240): call_user_func(Array, Resource id #177, Object(React\EventLoop\StreamSelectLoop))
#2 /home/Dramiel/mamba/vendor/react/event-loop/src/StreamSelectLoop.php(201): React\EventLoop\StreamSelectLoop->waitForStreamActivity(4862327.0988464)
#3 /home/Dramiel/mamba/vendor/team-reflex/discord-php/src/Discord/Discord.php(1223): React\EventLoop\StreamSelectLoop->run()
#4 /home/Dramiel/mamba/Dramiel.php(211): Discord\Discord->run() #5 {main} [] []

This is a Discord bot (https://github.com/shibdib/Dramiel) using DiscordPHP (https://github.com/teamreflex/DiscordPHP).

This is a common problem amongst everyone using the framework.

I was able to reproduce the following error in several Windows machines: when trying to read a text file using ReadableResourceStream I receive RuntimeException: Unable to set stream resource to non-blocking mode.

Here is an example code:

<?php

use React\EventLoop\Factory;
use React\Stream\ReadableResourceStream;

require_once 'vendor/autoload.php';

$loop = Factory::create();

$arquivo = fopen('existing_file.txt', 'r+');
$stream = new ReadableResourceStream($arquivo, $loop);

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

$loop->run();

And here is the full output:

PHP Fatal error:  Uncaught RuntimeException: Unable to set stream resource to non-blocking mode in C:\Users\carlo\OneDrive\Documentos\reactphp-windows\vendor\react\stream\src\ReadableResourceStream.php:58
Stack trace:
#0 C:\Users\carlo\OneDrive\Documentos\reactphp-windows\index.php(11): React\Stream\ReadableResourceStream->__construct(Resource id #17, Object(React\EventLoop\StreamSelectLoop))
#1 {main}
  thrown in C:\Users\carlo\OneDrive\Documentos\reactphp-windows\vendor\react\stream\src\ReadableResourceStream.php on line 58

Fatal error: Uncaught RuntimeException: Unable to set stream resource to non-blocking mode in C:\Users\carlo\OneDrive\Documentos\reactphp-windows\vendor\react\stream\src\ReadableResourceStream.php on line 58

RuntimeException: Unable to set stream resource to non-blocking mode in C:\Users\carlo\OneDrive\Documentos\reactphp-windows\vendor\react\stream\src\ReadableResourceStream.php on line 58

Call Stack:
    0.0002     393608   1. {main}() C:\Users\carlo\OneDrive\Documentos\reactphp-windows\index.php:0
    0.0046     612240   2. React\Stream\ReadableResourceStream->__construct($stream = resource(17) of type (stream), $loop = class React\EventLoop\StreamSelectLoop { private $futureTickQueue = class React\EventLoop\Tick\FutureTickQueue { private $queue = class SplQueue { ... } }; private $timers = class React\EventLoop\Timer\Timers { private $time = NULL; private $timers = [...]; private $schedule = [...]; private $sorted = TRUE; private $useHighResolution = TRUE }; private $readStreams = []; private $readListeners = []; private $writeStreams = []; private $writeListeners = []; private $running = NULL; private $pcntl = FALSE; private $pcntlPoll = FALSE; private $signals = class React\EventLoop\SignalsHandler { private $signals = [...] } }, $readChunkSize = ???) C:\Users\carlo\OneDrive\Documentos\reactphp-windows\index.php:11

System info:

• Windows 11
• PHP 8.1.5
• react/event-loop version: v1.3.0
• react/stream version: v1.2.0