Add optional $writeChunkSize parameter for max number of bytes to write

README.md

@@ -941,6 +941,21 @@ This value SHOULD NOT be changed unless you know what you're doing.
 $stream = new WritableResourceStream(STDOUT, $loop, 8192);
 ```
 
+This class takes an optional `int|null $writeChunkSize` parameter that controls
+this maximum buffer size in bytes to write at once to the stream.
+You can use a `null` value here in order to apply its default value.
+This value SHOULD NOT be changed unless you know what you're doing.
+This can be a positive number which means that up to X bytes will be written
+at once to the underlying stream resource. Note that the actual number
+of bytes written may be lower if the stream resource has less than X bytes
+currently available.
+This can be `-1` which means "write everything available" to the
+underlying stream resource.
+
+```php
+$stream = new WritableResourceStream(STDOUT, $loop, null, 8192);
+```
+
 See also [`write()`](#write) for more details.
 
 ### DuplexResourceStream

src/WritableResourceStream.php

@@ -10,13 +10,14 @@ final class WritableResourceStream extends EventEmitter implements WritableStrea
     private $stream;
     private $loop;
     private $softLimit;
+    private $writeChunkSize;
 
     private $listening = false;
     private $writable = true;
     private $closed = false;
     private $data = '';
 
-    public function __construct($stream, LoopInterface $loop, $writeBufferSoftLimit = null)
+    public function __construct($stream, LoopInterface $loop, $writeBufferSoftLimit = null, $writeChunkSize = null)
     {
         if (!is_resource($stream) || get_resource_type($stream) !== "stream") {
             throw new \InvalidArgumentException('First parameter must be a valid stream resource');
@@ -36,6 +37,7 @@ public function __construct($stream, LoopInterface $loop, $writeBufferSoftLimit
         $this->stream = $stream;
         $this->loop = $loop;
         $this->softLimit = ($writeBufferSoftLimit === null) ? 65536 : (int)$writeBufferSoftLimit;
+        $this->writeChunkSize = ($writeChunkSize === null) ? -1 : (int)$writeChunkSize;
     }
 
     public function isWritable()
@@ -107,7 +109,11 @@ public function handleWrite()
             );
         });
 
-        $sent = fwrite($this->stream, $this->data);
+        if ($this->writeChunkSize === -1) {
+            $sent = fwrite($this->stream, $this->data);
+        } else {
+            $sent = fwrite($this->stream, $this->data, $this->writeChunkSize);
+        }
 
         restore_error_handler();
 

tests/FunctionalInternetTest.php

@@ -0,0 +1,106 @@
+<?php
+
+namespace React\Tests\Stream;
+
+use React\Stream\DuplexResourceStream;
+use React\EventLoop\Factory;
+use React\Stream\WritableResourceStream;
+
+/**
+ * @group internet
+ */
+class FunctionalInternetTest extends TestCase
+{
+    public function testUploadKilobytePlain()
+    {
+        $size = 1000;
+        $stream = stream_socket_client('tcp://httpbin.org:80');
+
+        $loop = Factory::create();
+        $stream = new DuplexResourceStream($stream, $loop);
+
+        $buffer = '';
+        $stream->on('data', function ($chunk) use (&$buffer) {
+            $buffer .= $chunk;
+        });
+
+        $stream->on('error', $this->expectCallableNever());
+
+        $stream->write("POST /post HTTP/1.0\r\nHost: httpbin.org\r\nContent-Length: $size\r\n\r\n" . str_repeat('.', $size));
+
+        $loop->run();
+
+        $this->assertNotEquals('', $buffer);
+    }
+
+    public function testUploadBiggerBlockPlain()
+    {
+        $size = 1000 * 30;
+        $stream = stream_socket_client('tcp://httpbin.org:80');
+
+        $loop = Factory::create();
+        $stream = new DuplexResourceStream($stream, $loop);
+
+        $buffer = '';
+        $stream->on('data', function ($chunk) use (&$buffer) {
+            $buffer .= $chunk;
+        });
+
+        $stream->on('error', $this->expectCallableNever());
+
+        $stream->write("POST /post HTTP/1.0\r\nHost: httpbin.org\r\nContent-Length: $size\r\n\r\n" . str_repeat('.', $size));
+
+        $loop->run();
+
+        $this->assertNotEquals('', $buffer);
+    }
+
+    public function testUploadKilobyteSecure()
+    {
+        $size = 1000;
+        $stream = stream_socket_client('tls://httpbin.org:443');
+
+        $loop = Factory::create();
+        $stream = new DuplexResourceStream($stream, $loop);
+
+        $buffer = '';
+        $stream->on('data', function ($chunk) use (&$buffer) {
+            $buffer .= $chunk;
+        });
+
+        $stream->on('error', $this->expectCallableNever());
+
+        $stream->write("POST /post HTTP/1.0\r\nHost: httpbin.org\r\nContent-Length: $size\r\n\r\n" . str_repeat('.', $size));
+
+        $loop->run();
+
+        $this->assertNotEquals('', $buffer);
+    }
+
+    public function testUploadBiggerBlockSecureRequiresSmallerChunkSize()
+    {
+        $size = 1000 * 30000;
+        $stream = stream_socket_client('tls://httpbin.org:443');
+
+        $loop = Factory::create();
+        $stream = new DuplexResourceStream(
+            $stream,
+            $loop,
+            null,
+            new WritableResourceStream($stream, $loop, null, 8192)
+        );
+
+        $buffer = '';
+        $stream->on('data', function ($chunk) use (&$buffer) {
+            $buffer .= $chunk;
+        });
+
+        $stream->on('error', $this->expectCallableNever());
+
+        $stream->write("POST /post HTTP/1.0\r\nHost: httpbin.org\r\nContent-Length: $size\r\n\r\n" . str_repeat('.', $size));
+
+        $loop->run();
+
+        $this->assertNotEquals('', $buffer);
+    }
+}