Added Provider to ImportConnector.

Author: Bilge

.github/workflows/CI.yml

@@ -9,11 +9,13 @@ jobs:
     runs-on: ubuntu-latest
 
     strategy:
+      fail-fast: false
       matrix:
         php:
           - 7.3
           - 7.4
           - 8.0
+          - 8.1
         dependencies:
           - hi
           - lo

composer.json

@@ -20,11 +20,11 @@
   },
   "require-dev": {
     "amphp/phpunit-util": "^1.1",
-    "infection/infection": ">=0.18,<1",
+    "infection/infection": ">=0.18,<0.26",
     "justinrainbow/json-schema": "^5.2.10",
     "mockery/mockery": "^1.4.2",
     "phpunit/php-code-coverage": "^9.2.5",
-    "phpunit/phpunit": "^9.5",
+    "phpunit/phpunit": "^9.5.13",
     "thecodingmachine/safe": "^1.3.3"
   },
   "suggest" : {
@@ -46,6 +46,9 @@
     "mutation": "infection --configuration=test/infection.json"
   },
   "config": {
-    "sort-packages": true
+    "sort-packages": true,
+    "allow-plugins": {
+      "infection/extension-installer": false
+    }
   }
 }

src/Connector/ImportConnector.php

@@ -9,6 +9,8 @@
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableException;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
 use ScriptFUSION\Porter\Connector\Recoverable\StatelessRecoverableExceptionHandler;
+use ScriptFUSION\Porter\Provider\AsyncProvider;
+use ScriptFUSION\Porter\Provider\Provider;
 use function Amp\call;
 use function Amp\Promise\all;
 use function ScriptFUSION\Retry\retry;
@@ -24,6 +26,8 @@
  */
 final class ImportConnector implements ConnectorWrapper
 {
+    private $provider;
+
     private $connector;
 
     /**
@@ -45,6 +49,7 @@ final class ImportConnector implements ConnectorWrapper
     private $throttle;
 
     /**
+     * @param Provider|AsyncProvider $provider Provider.
      * @param Connector|AsyncConnector $connector Wrapped connector.
      * @param RecoverableExceptionHandler $recoverableExceptionHandler User's recoverable exception handler.
      * @param int $maxFetchAttempts Maximum fetch attempts.
@@ -53,6 +58,7 @@ final class ImportConnector implements ConnectorWrapper
      *     for synchronous imports only.
      */
     public function __construct(
+        $provider,
         $connector,
         RecoverableExceptionHandler $recoverableExceptionHandler,
         int $maxFetchAttempts,
@@ -63,6 +69,7 @@ public function __construct(
             throw CacheUnavailableException::createUnsupported();
         }
 
+        $this->provider = $provider;
         $this->connector = clone (
             $connector instanceof CachingConnector && !$mustCache
                 // Bypass cache when not required.
@@ -174,6 +181,16 @@ private static function invokeHandler(
         return $handler($recoverableException);
     }
 
+    /**
+     * Gets the provider owning the resource being imported.
+     *
+     * @return AsyncProvider|Provider
+     */
+    public function getProvider()
+    {
+        return $this->provider;
+    }
+
     /**
      * Gets the wrapped connector.
      *

src/Connector/ImportConnectorFactory.php

@@ -4,6 +4,8 @@
 namespace ScriptFUSION\Porter\Connector;
 
 use ScriptFUSION\Async\Throttle\NullThrottle;
+use ScriptFUSION\Porter\Provider\AsyncProvider;
+use ScriptFUSION\Porter\Provider\Provider;
 use ScriptFUSION\Porter\Specification\AsyncImportSpecification;
 use ScriptFUSION\Porter\Specification\Specification;
 use ScriptFUSION\StaticClass;
@@ -16,12 +18,13 @@ final class ImportConnectorFactory
     use StaticClass;
 
     /**
+     * @param Provider|AsyncProvider $provider
      * @param Connector|AsyncConnector $connector
      * @param Specification $specification
      *
      * @return ImportConnector
      */
-    public static function create($connector, Specification $specification): ImportConnector
+    public static function create($provider, $connector, Specification $specification): ImportConnector
     {
         if ($specification instanceof AsyncImportSpecification) {
             $throttle = $specification->getThrottle();
@@ -32,6 +35,7 @@ public static function create($connector, Specification $specification): ImportC
         }
 
         return new ImportConnector(
+            $provider,
             $connector,
             $specification->getRecoverableExceptionHandler(),
             $specification->getMaxFetchAttempts(),

src/Porter.php

@@ -120,7 +120,9 @@ private function fetch(ImportSpecification $specification): PorterRecords
             ));
         }
 
-        $records = $resource->fetch(ImportConnectorFactory::create($provider->getConnector(), $specification));
+        $records = $resource->fetch(
+            ImportConnectorFactory::create($provider, $provider->getConnector(), $specification)
+        );
 
         if (!$records instanceof ProviderRecords) {
             $records = $this->createProviderRecords($records, $specification->getResource());
@@ -201,7 +203,7 @@ private function fetchAsync(AsyncImportSpecification $specification): AsyncPorte
         }
 
         $records = $resource->fetchAsync(
-            ImportConnectorFactory::create($provider->getAsyncConnector(), $specification)
+            ImportConnectorFactory::create($provider, $provider->getAsyncConnector(), $specification)
         );
 
         if (!$records instanceof AsyncProviderRecords) {

src/Provider/Resource/AsyncResource.php

@@ -19,7 +19,7 @@ interface AsyncResource
     public function getProviderClassName(): string;
 
     /**
-     * Fetches data using the the specified connector and presents its data as an enumerable series.
+     * Fetches data using the specified connector and presents its data as an enumerable series.
      *
      * @param ImportConnector $connector Connector.
      *

src/Provider/Resource/ProviderResource.php

@@ -18,7 +18,7 @@ interface ProviderResource
     public function getProviderClassName(): string;
 
     /**
-     * Fetches data using the the specified connector and presents it as an iterable series.
+     * Fetches data using the specified connector and presents it as an iterable series.
      *
      * @param ImportConnector $connector Connector.
      *

test/FixtureFactory.php

@@ -6,6 +6,7 @@
 use ScriptFUSION\Porter\Connector\Connector;
 use ScriptFUSION\Porter\Connector\ImportConnector;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
+use ScriptFUSION\Porter\Provider\Provider;
 use ScriptFUSION\Porter\Specification\ImportSpecification;
 use ScriptFUSION\StaticClass;
 
@@ -17,9 +18,11 @@ public static function buildImportConnector(
         Connector $connector,
         RecoverableExceptionHandler $recoverableExceptionHandler = null,
         int $maxFetchAttempts = ImportSpecification::DEFAULT_FETCH_ATTEMPTS,
-        bool $mustCache = false
+        bool $mustCache = false,
+        Provider $provider = null
     ): ImportConnector {
         return new ImportConnector(
+            $provider ?? \Mockery::mock(Provider::class),
             $connector,
             $recoverableExceptionHandler ?: \Mockery::spy(RecoverableExceptionHandler::class),
             $maxFetchAttempts,

test/Functional/ThrottlePrecedenceHierarchyTest.php

@@ -50,7 +50,11 @@ public function testNonThrottledConnector(): \Generator
         $this->specificationThrottle->expects('await')->once()->andReturn(new Success());
         $this->connectorThrottle->expects('await')->never();
 
-        $connector = ImportConnectorFactory::create($this->provider->getAsyncConnector(), $this->specification);
+        $connector = ImportConnectorFactory::create(
+            $this->provider,
+            $this->provider->getAsyncConnector(),
+            $this->specification
+        );
 
         yield $connector->fetchAsync(\Mockery::mock(AsyncDataSource::class));
     }
@@ -64,7 +68,11 @@ public function testThrottledConnector(): \Generator
         $this->specificationThrottle->expects('await')->never();
         $this->connectorThrottle->expects('await')->once()->andReturn(new Success());
 
-        $connector = ImportConnectorFactory::create($this->mockThrottledConnector(), $this->specification);
+        $connector = ImportConnectorFactory::create(
+            $this->provider,
+            $this->mockThrottledConnector(),
+            $this->specification
+        );
 
         yield $connector->fetchAsync(\Mockery::mock(AsyncDataSource::class));
     }
@@ -79,7 +87,11 @@ public function testThrottledConnectorOverride(): \Generator
         $this->specificationThrottle->expects('await')->once()->andReturn(new Success());
         $this->connectorThrottle->expects('await')->never();
 
-        $connector = ImportConnectorFactory::create($this->mockThrottledConnector(), $this->specification);
+        $connector = ImportConnectorFactory::create(
+            $this->provider,
+            $this->mockThrottledConnector(),
+            $this->specification
+        );
 
         yield $connector->fetchAsync(\Mockery::mock(AsyncDataSource::class));
     }

test/Integration/PorterAsyncTest.php

@@ -12,7 +12,6 @@
 use ScriptFUSION\Porter\Collection\AsyncRecordCollection;
 use ScriptFUSION\Porter\Collection\CountableAsyncPorterRecords;
 use ScriptFUSION\Porter\Collection\CountableAsyncProviderRecords;
-use ScriptFUSION\Porter\Collection\PorterRecords;
 use ScriptFUSION\Porter\ForeignResourceException;
 use ScriptFUSION\Porter\ImportException;
 use ScriptFUSION\Porter\IncompatibleProviderException;