Move stream factory creation to node

src/AdapterInterface.php

@@ -8,11 +8,11 @@
 interface AdapterInterface
 {
     const CREATION_MODE = 'rwxrw----';
-    
+
     /**
      * Checks whether the current installation supports the adapter.
      *
-     * @return boolean
+     * @return bool
      */
     public static function isSupported();
 
@@ -127,7 +127,7 @@ public function touch($path, $mode = self::CREATION_MODE);
      * @param string $path
      * @param string $flags
      * @param $mode
-     * @return PromiseInterface<file descriptor>
+     * @return PromiseInterface
      */
     public function open($path, $flags, $mode = self::CREATION_MODE);
 

src/ChildProcess/Adapter.php

@@ -15,7 +15,6 @@
 use React\Filesystem\TypeDetectorInterface;
 use React\Filesystem\PermissionFlagResolver;
 use React\Filesystem\Node\NodeInterface;
-use React\Filesystem\Stream\StreamFactory;
 use React\Promise\PromiseInterface;
 use WyriHaximus\React\ChildProcess\Messenger\Messages\Factory;
 use WyriHaximus\React\ChildProcess\Messenger\Messages\Payload;
@@ -212,8 +211,8 @@ public function open($path, $flags, $mode = self::CREATION_MODE)
                 'flags' => $flags,
                 'mode' => $mode,
             ]));
-        })->then(function () use ($path, $flags, &$id) {
-            return \React\Promise\resolve(StreamFactory::create($path, $id, $flags, $this));
+        })->then(function () use (&$id) {
+            return $id;
         });
     }
 

src/Eio/Adapter.php

@@ -12,7 +12,6 @@
 use React\Filesystem\ObjectStream;
 use React\Filesystem\OpenFileLimiter;
 use React\Filesystem\PermissionFlagResolver;
-use React\Filesystem\Stream\StreamFactory;
 use React\Filesystem\TypeDetectorInterface;
 use React\Promise\Deferred;
 
@@ -102,7 +101,7 @@ protected function applyConfiguration(array $options)
     }
 
     /**
-     * @return boolean
+     * @return bool
      */
     public static function isSupported()
     {
@@ -267,9 +266,7 @@ public function open($path, $flags, $mode = self::CREATION_MODE)
                 $eioFlags,
                 $mode,
             ]);
-        })->then(function ($fileDescriptor) use ($path, $flags) {
-            return StreamFactory::create($path, $fileDescriptor, $flags, $this);
-        }, function ($error) {
+        })->otherwise(function ($error) {
             $this->openFileLimiter->close();
             return \React\Promise\reject($error);
         });

src/Node/Directory.php

@@ -240,7 +240,7 @@ public function lsRecursiveStreaming()
 
     /**
      * @param $sourceStream
-     * @return Stream
+     * @return ObjectStream
      */
     protected function processLsRecursiveContents($sourceStream)
     {

src/Node/File.php

@@ -7,7 +7,7 @@
 use React\Filesystem\FilesystemInterface;
 use React\Filesystem\ObjectStream;
 use React\Filesystem\ObjectStreamSink;
-use React\Filesystem\Stream\GenericStreamInterface;
+use React\Filesystem\Stream\StreamFactory;
 use React\Promise\Stream;
 use React\Stream\ReadableStreamInterface;
 use React\Stream\WritableStreamInterface;
@@ -118,10 +118,10 @@ public function open($flags, $mode = AdapterInterface::CREATION_MODE)
             return \React\Promise\reject(new Exception('File is already open'));
         }
 
-        return $this->adapter->open($this->path, $flags, $mode)->then(function (GenericStreamInterface $stream) {
+        return $this->adapter->open($this->path, $flags, $mode)->then(function ($fd) use ($flags) {
             $this->open = true;
-            $this->fileDescriptor = $stream->getFiledescriptor();
-            return $stream;
+            $this->fileDescriptor = $fd;
+            return StreamFactory::create($this->path, $fd, $flags, $this->adapter);
         });
     }
 

src/Stream/GenericStreamTrait.php

@@ -48,15 +48,15 @@ public function getFiledescriptor()
     }
 
     /**
-     * @return boolean
+     * @return bool
      */
     public function isClosed()
     {
         return $this->closed;
     }
 
     /**
-     * @param boolean $closed
+     * @param bool $closed
      */
     public function setClosed($closed)
     {

tests/Node/FileTest.php

@@ -235,23 +235,21 @@ public function testOpen()
         $path = 'foo.bar';
         $filesystem = $this->mockAdapter();
 
-        $stream = $this->getMock('React\Filesystem\Stream\GenericStreamInterface', [], ['foo:bar']);
+        $fd = 'foo:bar';
         $flags = 'abc';
 
         $filesystem
             ->expects($this->once())
             ->method('open')
             ->with($path, $flags)
-            ->will($this->returnValue(new FulfilledPromise($stream)))
+            ->will($this->returnValue(new FulfilledPromise($fd)))
         ;
 
-        $callbackFired = false;
-        (new File($path, Filesystem::createFromAdapter($filesystem)))->open($flags)->then(function ($passStream) use (&$callbackFired, $stream) {
-            $this->assertSame($stream, $passStream);
-            $callbackFired = true;
-        });
+        $fs = Filesystem::createFromAdapter($filesystem);
+        $pass = $this->await((new File($path, $fs))->open($flags), $fs->getAdapter()->getLoop());
 
-        $this->assertTrue($callbackFired);
+        $this->assertInstanceOf('\React\Filesystem\Stream\GenericStreamInterface', $pass);
+        $this->assertSame($fd, $pass->getFiledescriptor());
     }
 
 
@@ -260,14 +258,14 @@ public function testOpenTwice()
         $path = 'foo.bar';
         $filesystem = $this->mockAdapter();
 
-        $stream = $this->getMock('React\Filesystem\Stream\GenericStreamInterface', [], ['foo:bar']);
+        $fd = 'foo:bar';
         $flags = 'abc';
 
         $filesystem
             ->expects($this->once())
             ->method('open')
             ->with($path, $flags)
-            ->will($this->returnValue(new FulfilledPromise($stream)))
+            ->will($this->returnValue(new FulfilledPromise($fd)))
         ;
 
         $file = new File($path, Filesystem::createFromAdapter($filesystem));
@@ -280,14 +278,10 @@ public function testGetContents()
         $path = 'foo.bar';
         $fd = '0123456789abcdef';
 
-        $openPromise = $this->getMock('React\Promise\PromiseInterface', [
-            'then',
-        ]);
-
         $filesystem = $this->mockAdapter();
 
         $filesystem
-            ->expects($this->once())
+            ->expects($this->any())
             ->method('stat')
             ->with($path)
             ->will($this->returnValue(new FulfilledPromise([
@@ -299,14 +293,14 @@ public function testGetContents()
             ->expects($this->once())
             ->method('open')
             ->with($path, 'r')
-            ->will($this->returnValue($openPromise))
+            ->will($this->returnValue(new FulfilledPromise($fd)))
         ;
 
         $filesystem
             ->expects($this->once())
             ->method('read')
             ->with($fd, 1, 0)
-            ->will($this->returnValue(new FulfilledPromise(str_repeat('a', 1))))
+            ->will($this->returnValue(new FulfilledPromise('a')))
         ;
 
         $filesystem
@@ -316,21 +310,6 @@ public function testGetContents()
             ->will($this->returnValue(new FulfilledPromise()))
         ;
 
-        $stream = new ReadableStream(
-            $path,
-            $fd,
-            $filesystem
-        );
-
-        $openPromise
-            ->expects($this->once())
-            ->method('then')
-            ->with($this->isType('callable'))
-            ->will($this->returnCallback(function ($resolveCb) use ($stream) {
-                return new FulfilledPromise($resolveCb($stream));
-            }))
-        ;
-
         $getContentsPromise = (new File($path, Filesystem::createFromAdapter($filesystem)))->getContents();
         $this->assertInstanceOf('React\Promise\PromiseInterface', $getContentsPromise);
     }
@@ -341,30 +320,20 @@ public function testClose()
         $fd = '0123456789abcdef';
         $filesystem = $this->mockAdapter();
 
-        $stream = $this->getMock('React\Filesystem\Stream\GenericStreamInterface', [
-            'getFiledescriptor',
-        ], [
-            'foo:bar',
-        ]);
+        $openPromise = new FulfilledPromise($fd);
 
-        $stream
-            ->expects($this->once())
-            ->method('getFiledescriptor')
-            ->with()
-            ->will($this->returnValue($fd))
+        $filesystem
+            ->method('stat')
+            ->with($path)
+            ->will($this->returnValue(new FulfilledPromise([
+                'size' => 1,
+            ])))
         ;
 
-        $openPromise = $this->getMock('React\Promise\PromiseInterface', [
-            'then',
-        ]);
-
-        $openPromise
-            ->expects($this->once())
-            ->method('then')
-            ->with($this->isType('callable'))
-            ->will($this->returnCallback(function ($resolveCb) use ($stream) {
-                return new FulfilledPromise($resolveCb($stream));
-            }))
+        $filesystem
+            ->method('read')
+            ->with($path)
+            ->will($this->returnValue(new FulfilledPromise('a')))
         ;
 
         $filesystem
@@ -374,24 +343,11 @@ public function testClose()
             ->will($this->returnValue($openPromise))
         ;
 
-        $closePromise = $this->getMock('React\Promise\PromiseInterface', [
-            'then',
-        ]);
-
-        $closePromise
-            ->expects($this->once())
-            ->method('then')
-            ->with($this->isType('callable'))
-            ->will($this->returnCallback(function ($resolveCb) use ($stream) {
-                return \React\Promise\resolve($resolveCb($stream));
-            }))
-        ;
-
         $filesystem
             ->expects($this->once())
             ->method('close')
             ->with($fd)
-            ->will($this->returnValue($closePromise))
+            ->will($this->returnValue(\React\Promise\resolve()))
         ;
 
         $file = new File($path, Filesystem::createFromAdapter($filesystem));