Set read buffer to 0

composer.json

@@ -5,15 +5,15 @@
     "license": "MIT",
     "require": {
         "php": ">=5.4.0",
-        "evenement/evenement": "~2.0"
+        "evenement/evenement": "^2.0"
     },
     "require-dev": {
-        "react/event-loop": "0.4.*",
-        "react/promise": "~2.0"
+        "react/event-loop": "^0.4",
+        "react/promise": "^2.0"
     },
     "suggest": {
-        "react/event-loop": "0.4.*",
-        "react/promise": "~2.0"
+        "react/event-loop": "^0.4",
+        "react/promise": "^2.0"
     },
     "autoload": {
         "psr-4": {

src/Stream.php

@@ -24,6 +24,7 @@ public function __construct($stream, LoopInterface $loop)
         }
 
         stream_set_blocking($this->stream, 0);
+        stream_set_read_buffer($this->stream, 0);
 
         $this->loop = $loop;
         $this->buffer = new Buffer($this->stream, $this->loop);

tests/StreamIntegrationTest.php

@@ -0,0 +1,58 @@
+<?php
+
+namespace React\Tests\Stream;
+
+use React\Stream\Stream;
+use React\EventLoop as rel;
+
+class StreamIntegrationTest extends TestCase
+{
+    public function loopProvider()
+    {
+        return [
+            [function() { return true; }, function() { return new rel\StreamSelectLoop; }],
+            [function() { return function_exists('event_base_new'); }, function() { return new rel\LibEventLoop; }],
+            [function() { return class_exists('libev\EventLoop'); }, function() { return new rel\LibEvLoop; }],
+            [function() { return class_exists('EventBase'); }, function() { return new rel\ExtEventLoop; }]
+        ];
+    }
+
+    /**
+     * @dataProvider loopProvider
+     */
+    public function testBufferReadsLargeChunks($condition, $loopFactory)
+    {
+        if (true !== $condition()) {
+            return $this->markTestSkipped('Loop implementation not available');
+        }
+
+        $loop = $loopFactory();
+
+        list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+
+        $streamA = new Stream($sockA, $loop);
+        $streamB = new Stream($sockB, $loop);
+
+        $bufferSize = 4096;
+        $streamA->bufferSize = $bufferSize;
+        $streamB->bufferSize = $bufferSize;
+
+        $testString = str_repeat("*", $streamA->bufferSize + 1);
+
+        $buffer = "";
+        $streamB->on('data', function ($data, $streamB) use (&$buffer, &$testString) {
+            $buffer .= $data;
+        });
+
+        $streamA->write($testString);
+
+        $loop->tick();
+        $loop->tick();
+        $loop->tick();
+
+        $streamA->close();
+        $streamB->close();
+
+        $this->assertEquals($testString, $buffer);
+    }
+}