Added ExponentialAsyncDelayRecoverableExceptionHandler.

Changed default exception handler for AsyncImportSpecification.
Added base Specification class for both sync and async specifications.

Author: Bilge

composer.json

@@ -12,7 +12,7 @@
     "php": "^7.1",
     "scriptfusion/static-class": "^1",
     "scriptfusion/retry": "^2.1",
-    "scriptfusion/retry-exception-handlers": "^1.1",
+    "scriptfusion/retry-exception-handlers": "^1.2",
     "amphp/amp": "^2",
     "psr/container": "^1",
     "psr/cache": "^1"

src/Connector/ImportConnectorFactory.php

@@ -3,7 +3,7 @@
 
 namespace ScriptFUSION\Porter\Connector;
 
-use ScriptFUSION\Porter\Specification\ImportSpecification;
+use ScriptFUSION\Porter\Specification\Specification;
 use ScriptFUSION\StaticClass;
 
 final class ImportConnectorFactory
@@ -12,11 +12,11 @@ final class ImportConnectorFactory
 
     /**
      * @param Connector|AsyncConnector $connector
-     * @param ImportSpecification $specification
+     * @param Specification $specification
      *
      * @return ImportConnector
      */
-    public static function create($connector, ImportSpecification $specification): ImportConnector
+    public static function create($connector, Specification $specification): ImportConnector
     {
         return new ImportConnector(
             $connector,

src/Connector/Recoverable/ExponentialAsyncDelayRecoverableExceptionHandler.php

@@ -0,0 +1,41 @@
+<?php
+declare(strict_types=1);
+
+namespace ScriptFUSION\Porter\Connector\Recoverable;
+
+use Amp\Promise;
+use ScriptFUSION\Retry\ExceptionHandler\AsyncExponentialBackoffExceptionHandler;
+
+/**
+ * Delays async execution for an exponentially increasing series of delays.
+ */
+class ExponentialAsyncDelayRecoverableExceptionHandler implements RecoverableExceptionHandler
+{
+    private $initialDelay;
+
+    /**
+     * @var AsyncExponentialBackoffExceptionHandler
+     */
+    private $handler;
+
+    /**
+     * Initializes this instance with the specified initial delay. The initial delay will be used when the first
+     * exception is handled; subsequent exceptions will cause longer delays.
+     *
+     * @param int $initialDelay Initial delay in milliseconds.
+     */
+    public function __construct(int $initialDelay = AsyncExponentialBackoffExceptionHandler::DEFAULT_COEFFICIENT)
+    {
+        $this->initialDelay = $initialDelay;
+    }
+
+    public function initialize(): void
+    {
+        $this->handler = new AsyncExponentialBackoffExceptionHandler($this->initialDelay);
+    }
+
+    public function __invoke(RecoverableException $exception): ?Promise
+    {
+        return ($this->handler)();
+    }
+}

src/Connector/Recoverable/ExponentialSleepRecoverableExceptionHandler.php

@@ -7,19 +7,22 @@
 use ScriptFUSION\Retry\ExceptionHandler\ExponentialBackoffExceptionHandler;
 
 /**
- * Sleeps for an exponentially increasing series of delays specified in microseconds.
+ * Sleeps for an exponentially increasing series of delays.
  */
 class ExponentialSleepRecoverableExceptionHandler implements RecoverableExceptionHandler
 {
     private $initialDelay;
 
+    /**
+     * @var ExponentialBackoffExceptionHandler
+     */
     private $handler;
 
     /**
      * Initializes this instance with the specified initial delay. The initial delay will be used when the first
      * exception is handled; subsequent exceptions will cause longer delays.
      *
-     * @param int $initialDelay Initial delay.
+     * @param int $initialDelay Initial delay in microseconds.
      */
     public function __construct(int $initialDelay = ExponentialBackoffExceptionHandler::DEFAULT_COEFFICIENT)
     {
@@ -33,6 +36,8 @@ public function initialize(): void
 
     public function __invoke(RecoverableException $exception): ?Promise
     {
-        return ($this->handler)($exception);
+        ($this->handler)();
+
+        return null;
     }
 }

src/Porter.php

@@ -59,8 +59,6 @@ public function __construct(ContainerInterface $providers)
      * @param ImportSpecification $specification Import specification.
      *
      * @return PorterRecords|CountablePorterRecords
-     *
-     * @throws ImportException Provider failed to return an iterator.
      */
     public function import(ImportSpecification $specification): PorterRecords
     {
@@ -128,6 +126,13 @@ private function fetch(ImportSpecification $specification): \Iterator
         return $resource->fetch(ImportConnectorFactory::create($connector, $specification));
     }
 
+    /**
+     * Imports data asynchronously according to the design of the specified asynchronous import specification.
+     *
+     * @param AsyncImportSpecification $specification Asynchronous import specification.
+     *
+     * @return AsyncPorterRecords|CountableAsyncPorterRecords
+     */
     public function importAsync(AsyncImportSpecification $specification): AsyncRecordCollection
     {
         $specification = clone $specification;
@@ -143,6 +148,13 @@ public function importAsync(AsyncImportSpecification $specification): AsyncRecor
         return $this->createAsyncPorterRecords($records, $specification);
     }
 
+    /**
+     * Imports one record according to the design of the specified asynchronous import specification.
+     *
+     * @param AsyncImportSpecification $specification Asynchronous import specification.
+     *
+     * @return Promise Promise that resolves to a record.
+     */
     public function importOneAsync(AsyncImportSpecification $specification): Promise
     {
         return \Amp\call(function () use ($specification) {
@@ -210,17 +222,19 @@ private function transformRecords(RecordCollection $records, array $transformers
         return $records;
     }
 
+    /**
+     * @param AsyncRecordCollection $records
+     * @param AsyncTransformer[] $transformers
+     * @param mixed $context
+     *
+     * @return AsyncRecordCollection
+     */
     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);
             }

src/Provider/Resource/PseudoBisyncResource.php

@@ -3,37 +3,51 @@
 
 namespace ScriptFUSION\Porter\Specification;
 
+use ScriptFUSION\Porter\Connector\Recoverable\ExponentialAsyncDelayRecoverableExceptionHandler;
+use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
 use ScriptFUSION\Porter\Provider\Resource\AsyncResource;
-use ScriptFUSION\Porter\Provider\Resource\ProviderResource;
-use ScriptFUSION\Porter\Provider\Resource\PseudoBisyncResource;
+use ScriptFUSION\Porter\Transform\AnysyncTransformer;
+use ScriptFUSION\Porter\Transform\AsyncTransformer;
 
 /**
- * TODO: Use trait instead of extending ImportSpecification. Async specs are NOT a type of sync spec.
+ * Specifies which resource to import asynchronously and how the data should be transformed.
  */
-class AsyncImportSpecification extends ImportSpecification
+class AsyncImportSpecification extends Specification
 {
     private $asyncResource;
 
     public function __construct(AsyncResource $resource)
     {
-        if (!$resource instanceof ProviderResource) {
-            $resource = new PseudoBisyncResource($resource);
-        }
-
-        parent::__construct($resource);
-
         $this->asyncResource = $resource;
+
+        parent::__construct();
     }
 
     public function __clone()
     {
-        parent::__clone();
-
         $this->asyncResource = clone $this->asyncResource;
+
+        parent::__clone();
     }
 
     final public function getAsyncResource(): AsyncResource
     {
         return $this->asyncResource;
     }
+
+    final public function addTransformer(AnysyncTransformer $transformer): Specification
+    {
+        if (!$transformer instanceof AsyncTransformer) {
+            throw new IncompatibleTransformerException(
+                'Transformer does not implement interface: ' . AsyncTransformer::class . '.'
+            );
+        }
+
+        return parent::addTransformer($transformer);
+    }
+
+    protected static function createDefaultRecoverableExceptionHandler(): RecoverableExceptionHandler
+    {
+        return new ExponentialAsyncDelayRecoverableExceptionHandler;
+    }
 }