Improve sending large chunks over TLS on macOS

See reactphp/stream#150.

.travis.yml

@@ -1,4 +1,4 @@
-sudo: false
+os: linux
 
 language: php
 
@@ -10,9 +10,17 @@ php:
   - 7.4
   - nightly
 
-matrix:
+jobs:
+  include:
+    - name: macOS
+      os: osx
+      language: php
+      before_install:
+        - curl -s http://getcomposer.org/installer | php
+        - mv composer.phar /usr/local/bin/composer
   allow_failures:
     - php: nightly
+    - os: osx
   fast_finish: true
 
 env:

lib/ResourceOutputStream.php

@@ -13,7 +13,9 @@
  */
 final class ResourceOutputStream implements OutputStream
 {
+    /** @deprecated No longer used. */
     const MAX_CONSECUTIVE_EMPTY_WRITES = 3;
+
     const LARGE_CHUNK_SIZE = 128 * 1024;
 
     /** @var resource|null */
@@ -58,8 +60,6 @@ public function __construct($stream, int $chunkSize = null)
         $resource = &$this->resource;
 
         $this->watcher = Loop::onWritable($stream, static function ($watcher, $stream) use ($writes, &$chunkSize, &$writable, &$resource) {
-            static $emptyWrites = 0;
-
             try {
                 while (!$writes->isEmpty()) {
                     /** @var Deferred $deferred */
@@ -75,6 +75,11 @@ public function __construct($stream, int $chunkSize = null)
                         throw new ClosedException("The stream was closed by the peer");
                     }
 
+                    $error = null;
+                    \set_error_handler(static function (int $errno) use (&$error) {
+                        $error = $errno;
+                    });
+
                     // Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full.
                     // Use conditional, because PHP doesn't like getting null passed
                     if ($chunkSize) {
@@ -83,6 +88,8 @@ public function __construct($stream, int $chunkSize = null)
                         $written = @\fwrite($stream, $data);
                     }
 
+                    \restore_error_handler();
+
                     \assert(
                         $written !== false || \PHP_VERSION_ID >= 70400, // PHP 7.4+ returns false on EPIPE.
                         "Trying to write on a previously fclose()'d resource. Do NOT manually fclose() resources the still referenced in the loop."
@@ -98,21 +105,14 @@ public function __construct($stream, int $chunkSize = null)
                     }
 
                     // Broken pipes between processes on macOS/FreeBSD do not detect EOF properly.
-                    if ($written === 0 || $written === false) {
-                        if ($emptyWrites++ > self::MAX_CONSECUTIVE_EMPTY_WRITES) {
-                            $message = "Failed to write to stream after multiple attempts";
-                            if ($error = \error_get_last()) {
-                                $message .= \sprintf("; %s", $error["message"]);
-                            }
-                            throw new StreamException($message);
+                    if (($written === 0 || $written === false) && $error !== null) {
+                        $message = "Failed to write to stream";
+                        if ($error = \error_get_last()) {
+                            $message .= \sprintf("; %s", $error["message"]);
                         }
-
-                        $writes->unshift([$data, $previous, $deferred]);
-                        return;
+                        throw new StreamException($message);
                     }
 
-                    $emptyWrites = 0;
-
                     if ($length > $written) {
                         $data = \substr($data, $written);
                         $writes->unshift([$data, $written + $previous, $deferred]);

test/ResourceOutputStreamTest.php

@@ -2,8 +2,10 @@
 
 namespace Amp\ByteStream\Test;
 
+use Amp\ByteStream\ResourceInputStream;
 use Amp\ByteStream\ResourceOutputStream;
 use Amp\ByteStream\StreamException;
+use Amp\Delayed;
 use Amp\PHPUnit\AsyncTestCase;
 
 class ResourceOutputStreamTest extends AsyncTestCase
@@ -68,6 +70,34 @@ public function testClosedRemoteSocket()
 
         // The first write still succeeds somehow...
         yield $stream->write("foobar");
+
+        // A delay seems required for the OS to realize the socket is indeed closed.
+        yield new Delayed(10);
+
         yield $stream->write("foobar");
     }
+
+    /**
+     * @see https://github.com/reactphp/stream/pull/150
+     */
+    public function testUploadBiggerBlockSecure()
+    {
+        $size = 2 ** 18; // 256kb
+
+        $resource = \stream_socket_client('tls://httpbin.org:443');
+
+        $output = new ResourceOutputStream($resource);
+
+        $body = \str_repeat('.', $size);
+
+        yield $output->write("POST /post HTTP/1.0\r\nHost: httpbin.org\r\nContent-Length: $size\r\n\r\n" . $body);
+
+        $input = new ResourceInputStream($resource);
+        $buffer = '';
+        while (null !== ($chunk = yield $input->read())) {
+            $buffer .= $chunk;
+        }
+
+        $this->assertStringContainsString($body, $buffer);
+    }
 }