Fix ResourceInputStream not properly nulling the resource on close

This causes the stream to be reported as open by amphp/socket, which in
turn causes InvalidWatcherErrors in amphp/http-client, because the
socket is reported as still open and thus referenced.

lib/ResourceInputStream.php

@@ -61,15 +61,16 @@ public function __construct($stream, int $chunkSize = self::DEFAULT_CHUNK_SIZE)
         \stream_set_blocking($stream, false);
         \stream_set_read_buffer($stream, 0);
 
-        $this->resource = $stream;
+        $this->resource = &$stream;
         $this->chunkSize = &$chunkSize;
 
         $deferred = &$this->deferred;
         $readable = &$this->readable;
 
-        $this->watcher = Loop::onReadable($this->resource, static function ($watcher, $stream) use (
+        $this->watcher = Loop::onReadable($this->resource, static function ($watcher) use (
             &$deferred,
             &$readable,
+            &$stream,
             &$chunkSize,
             $useSingleRead
         ) {
@@ -86,12 +87,13 @@ public function __construct($stream, int $chunkSize = self::DEFAULT_CHUNK_SIZE)
             // See https://github.com/amphp/byte-stream/issues/32
             if ($data === '' && @\feof($stream)) {
                 $readable = false;
-                Loop::cancel($watcher);
+                $stream = null;
                 $data = null; // Stream closed, resolve read with null.
+                Loop::cancel($watcher);
+            } else {
+                Loop::disable($watcher);
             }
 
-            Loop::disable($watcher);
-
             $temp = $deferred;
             $deferred = null;
 
@@ -136,8 +138,10 @@ public function read(): Promise
             // See https://github.com/amphp/byte-stream/issues/32
             if (@\feof($this->resource)) {
                 $this->readable = false;
+                $this->resource = null;
                 Loop::cancel($this->watcher);
-                $data = null; // Stream closed, resolve read with null.
+
+                return new Success; // Stream closed, resolve read with null.
             } else {
                 $this->deferred = new Deferred;
                 Loop::enable($this->watcher);
@@ -170,7 +174,6 @@ public function close()
             } else {
                 @\fclose($this->resource);
             }
-            $this->resource = null;
         }
 
         $this->free();
@@ -182,6 +185,7 @@ public function close()
     private function free()
     {
         $this->readable = false;
+        $this->resource = null;
 
         if ($this->deferred !== null) {
             $deferred = $this->deferred;