Changed AsyncTransformer to work with collections instead of records.

Added collection types for async collections.

Author: Bilge

src/Collection/AsyncFilteredRecords.php

@@ -0,0 +1,23 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Collection;
+
+use Amp\Iterator;
+
+class AsyncFilteredRecords extends AsyncRecordCollection
+{
+    private $filter;
+
+    public function __construct(Iterator $records, AsyncRecordCollection $previousCollection, callable $filter)
+    {
+        parent::__construct($records, $previousCollection);
+
+        $this->filter = $filter;
+    }
+
+    public function getFilter(): callable
+    {
+        return $this->filter;
+    }
+}

src/Collection/AsyncPorterRecords.php

@@ -0,0 +1,24 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Collection;
+
+use Amp\Iterator;
+use ScriptFUSION\Porter\Specification\AsyncImportSpecification;
+
+class AsyncPorterRecords extends AsyncRecordCollection
+{
+    private $specification;
+
+    public function __construct(Iterator $records, AsyncImportSpecification $specification)
+    {
+        parent::__construct($records, $records);
+
+        $this->specification = $specification;
+    }
+
+    public function getSpecification(): AsyncImportSpecification
+    {
+        return $this->specification;
+    }
+}

src/Collection/AsyncProviderRecords.php

@@ -0,0 +1,24 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Collection;
+
+use Amp\Iterator;
+use ScriptFUSION\Porter\Provider\Resource\AsyncResource;
+
+class AsyncProviderRecords extends AsyncRecordCollection
+{
+    private $resource;
+
+    public function __construct(Iterator $records, AsyncResource $resource)
+    {
+        parent::__construct($records);
+
+        $this->resource = $resource;
+    }
+
+    public function getResource(): AsyncResource
+    {
+        return $this->resource;
+    }
+}

src/Collection/AsyncRecordCollection.php

@@ -0,0 +1,44 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Collection;
+
+use Amp\Iterator;
+use Amp\Promise;
+
+abstract class AsyncRecordCollection implements Iterator
+{
+    private $records;
+
+    private $previousCollection;
+
+    public function __construct(Iterator $records, self $previousCollection = null)
+    {
+        $this->records = $records;
+        $this->previousCollection = $previousCollection;
+    }
+
+    public function advance(): Promise
+    {
+        return $this->records->advance();
+    }
+
+    public function getCurrent(): array
+    {
+        return $this->records->getCurrent();
+    }
+
+    public function getPreviousCollection(): ?AsyncRecordCollection
+    {
+        return $this->previousCollection;
+    }
+
+    public function findFirstCollection(): ?AsyncRecordCollection
+    {
+        do {
+            $previous = $nextPrevious ?? $this->getPreviousCollection();
+        } while ($previous && $nextPrevious = $previous->getPreviousCollection());
+
+        return $previous ?: $this;
+    }
+}

src/Collection/CountableAsyncPorterRecords.php

@@ -0,0 +1,18 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Collection;
+
+use ScriptFUSION\Porter\Specification\AsyncImportSpecification;
+
+class CountableAsyncPorterRecords extends AsyncPorterRecords implements \Countable
+{
+    use CountableRecordsTrait;
+
+    public function __construct(AsyncRecordCollection $records, int $count, AsyncImportSpecification $specification)
+    {
+        parent::__construct($records, $specification);
+
+        $this->setCount($count);
+    }
+}

src/Collection/CountableAsyncProviderRecords.php

@@ -0,0 +1,19 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Collection;
+
+use Amp\Iterator;
+use ScriptFUSION\Porter\Provider\Resource\AsyncResource;
+
+class CountableAsyncProviderRecords extends AsyncProviderRecords implements \Countable
+{
+    use CountableRecordsTrait;
+
+    public function __construct(Iterator $records, int $count, AsyncResource $resource)
+    {
+        parent::__construct($records, $resource);
+
+        $this->setCount($count);
+    }
+}

src/Collection/CountablePorterRecords.php

@@ -9,11 +9,6 @@ class CountablePorterRecords extends PorterRecords implements \Countable
 {
     use CountableRecordsTrait;
 
-    /**
-     * @param RecordCollection $records
-     * @param int $count
-     * @param ImportSpecification $specification
-     */
     public function __construct(RecordCollection $records, int $count, ImportSpecification $specification)
     {
         parent::__construct($records, $specification);

src/Collection/RecordCollection.php

@@ -18,6 +18,7 @@ public function __construct(\Iterator $records, self $previousCollection = null)
         $this->previousCollection = $previousCollection;
     }
 
+    // TODO: Consider throwing our own exception type for clarity, instead of relying on PHP's TypeError.
     public function current(): array
     {
         return $this->records->current();

src/Porter.php

@@ -4,9 +4,12 @@
 namespace ScriptFUSION\Porter;
 
 use Amp\Iterator;
-use Amp\Producer;
 use Amp\Promise;
 use Psr\Container\ContainerInterface;
+use ScriptFUSION\Porter\Collection\AsyncPorterRecords;
+use ScriptFUSION\Porter\Collection\AsyncProviderRecords;
+use ScriptFUSION\Porter\Collection\AsyncRecordCollection;
+use ScriptFUSION\Porter\Collection\CountableAsyncPorterRecords;
 use ScriptFUSION\Porter\Collection\CountablePorterRecords;
 use ScriptFUSION\Porter\Collection\CountableProviderRecords;
 use ScriptFUSION\Porter\Collection\PorterRecords;
@@ -125,13 +128,19 @@ private function fetch(ImportSpecification $specification): \Iterator
         return $resource->fetch(ImportConnectorFactory::create($connector, $specification));
     }
 
-    public function importAsync(AsyncImportSpecification $specification): Iterator
+    public function importAsync(AsyncImportSpecification $specification): AsyncRecordCollection
     {
         $specification = clone $specification;
 
         $records = $this->fetchAsync($specification);
 
-        return $this->transformAsync($records, $specification->getTransformers(), $specification->getContext());
+        if (!$records instanceof AsyncProviderRecords) {
+            $records = new AsyncProviderRecords($records, $specification->getAsyncResource());
+        }
+
+        $records = $this->transformAsync($records, $specification->getTransformers(), $specification->getContext());
+
+        return $this->createAsyncPorterRecords($records, $specification);
     }
 
     public function importOneAsync(AsyncImportSpecification $specification): Promise
@@ -201,38 +210,25 @@ private function transformRecords(RecordCollection $records, array $transformers
         return $records;
     }
 
-    private function transformAsync(Iterator $records, array $transformers, $context): Producer
-    {
-        return new Producer(function (\Closure $emit) use ($records, $transformers, $context) {
-            while (yield $records->advance()) {
-                $record = $records->getCurrent();
-
-                foreach ($transformers as $transformer) {
-                    if (!$transformer instanceof AsyncTransformer) {
-                        // TODO: Proper exception or separate async stack.
-                        throw new \RuntimeException('Cannot use sync transformer.');
-                    }
-                    if ($transformer instanceof PorterAware) {
-                        $transformer->setPorter($this);
-                    }
-
-                    $record = yield $transformer->transformAsync($record, $context);
-
-                    if ($record === null) {
-                        // Do not process more transformers.
-                        break;
-                    }
-                }
-
-                if ($record !== null) {
-                    if (!\is_array($record)) {
-                        throw new \RuntimeException('Unexpected type: record must be array or null.');
-                    }
-
-                    yield $emit($record);
-                }
+    private function transformAsync(
+        AsyncRecordCollection $records,
+        array $transformers,
+        $context
+    ): AsyncRecordCollection {
+        foreach ($transformers as $transformer) {
+            if (!$transformer instanceof AsyncTransformer) {
+                // TODO: Proper exception or separate async stack.
+                throw new \RuntimeException('Cannot use sync transformer.');
             }
-        });
+
+            if ($transformer instanceof PorterAware) {
+                $transformer->setPorter($this);
+            }
+
+            $records = $transformer->transformAsync($records, $context);
+        }
+
+        return $records;
     }
 
     private function createProviderRecords(\Iterator $records, ProviderResource $resource): ProviderRecords
@@ -253,6 +249,17 @@ private function createPorterRecords(RecordCollection $records, ImportSpecificat
         return new PorterRecords($records, $specification);
     }
 
+    private function createAsyncPorterRecords(
+        AsyncRecordCollection $records,
+        AsyncImportSpecification $specification
+    ): AsyncPorterRecords {
+        if ($records instanceof \Countable) {
+            return new CountableAsyncPorterRecords($records, \count($records), $specification);
+        }
+
+        return new AsyncPorterRecords($records, $specification);
+    }
+
     /**
      * Gets the provider matching the specified name.
      *

src/Transform/AsyncTransformer.php

@@ -3,9 +3,9 @@
 
 namespace ScriptFUSION\Porter\Transform;
 
-use Amp\Promise;
+use ScriptFUSION\Porter\Collection\AsyncRecordCollection;
 
 interface AsyncTransformer
 {
-    public function transformAsync(array $record, $context): Promise;
+    public function transformAsync(AsyncRecordCollection $records, $context): AsyncRecordCollection;
 }