Merge pull request #40 from WyriHaximus-labs/copy-fix

Reworked file copying to use pipe

Author: WyriHaximus

src/Node/File.php

@@ -206,32 +206,21 @@ protected function copyToFile(FileInterface $node)
     {
         $stream = new ObjectStream();
 
-        $this->open('r')->then(function (ReadableStreamInterface $readStream) use ($node, $stream) {
+        $this->open('r')->then(function (ReadableStreamInterface $readStream) use ($node) {
             $readStream->pause();
-            return $node->open('ctw')->then(function (WritableStreamInterface $writeStream) use ($readStream, $node, $stream) {
-                $deferred = new Deferred();
-                $writePromises = [];
-                $readStream->on('end', function () use ($deferred, $writeStream, &$writePromises) {
-                    \React\Promise\all($writePromises)->then(function ()use ($deferred, $writeStream) {
-                        $writeStream->end();
-                        $deferred->resolve();
-                    });
-                });
-                $readStream->on('data', function ($data) use ($writeStream, &$writePromises) {
-                    $writePromises[] = $writeStream->write($data);
-                });
-                $readStream->resume();
-                return $deferred->promise();
-            })->always(function () use ($node) {
-                $node->close();
+
+            return \React\Promise\all([
+                'read' => $readStream,
+                'write' => $node->open('ctw'),
+            ]);
+        })->then(function (array $streams) use ($stream, $node) {
+            $streams['read']->pipe($streams['write']);
+            $streams['read']->on('close', function () use ($streams, $stream, $node) {
+                $streams['write']->close();
+                $stream->end($node);
             });
-        })->then(function () {
-            return $this->close();
-        }, function () {
-            return $this->close();
-        })->then(function () use ($stream, $node) {
-            $stream->end($node);
-        });
+            $streams['read']->resume();
+        })->done();
 
         return $stream;
     }

tests/Adapters/FileTest.php

@@ -187,15 +187,7 @@ public function testCopyToDirectory(LoopInterface $loop, FilesystemInterface $fi
         } while (!file_exists($tempFileSource) && !file_exists($tempFileDestination));
         $this->assertFileExists($tempFileSource);
         $this->assertSame($contents, file_get_contents($tempFileSource));
-        $promise = $filesystem->file($tempFileSource)->copy($filesystem->dir($tempFileDestination));
-        $timer = $loop->addTimer(self::TIMEOUT, function () use ($loop) {
-            $loop->stop();
-            $this->fail('Event loop timeout');
-        });
-        $promise->always(function () use ($loop, $timer) {
-            $loop->cancelTimer($timer);
-        });
-        $this->await($promise, $loop);
+        $this->await($filesystem->file($tempFileSource)->copy($filesystem->dir($tempFileDestination)), $loop);
         do {
             usleep(500);
             $this->checkIfTimedOut();