Merge pull request #18 from spiral/feature/dynamic-pipelines

Adds an ability to connect to a pipeline without configuration.
spiral · Jul 21, 2022 · fc1371c · fc1371c
2 parents 2fce041 + 798d6cd
commit fc1371c
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 17 deletions.
diff --git a/composer.json b/composer.json
@@ -49,7 +49,10 @@
     }
   },
   "config": {
-    "sort-packages": true
+    "sort-packages": true,
+    "allow-plugins": {
+      "spiral/composer-publish-plugin": true
+    }
   },
   "minimum-stability": "dev",
   "prefer-stable": true

diff --git a/src/Queue/RPCPipelineRegistry.php b/src/Queue/RPCPipelineRegistry.php
@@ -44,9 +44,7 @@ public function getPipeline(string $name): QueueInterface
         }
 
         if (! isset($this->pipelines[$name])) {
-            throw new InvalidArgumentException(
-                sprintf('Queue pipeline with given name `%s` is not found.', $name)
-            );
+            return $this->jobs->connect($name);
         }
 
         if (! isset($this->pipelines[$name]['connector'])) {
@@ -63,15 +61,13 @@ public function getPipeline(string $name): QueueInterface
 
         /** @var CreateInfoInterface $connector */
         $connector = $this->pipelines[$name]['connector'];
-        $consume = (bool)($this->pipelines[$name]['consume'] ?? true);
 
         if (! $this->isExists($connector)) {
-            $queue = $this->create($connector, $consume);
-        } else {
-            $queue = $this->connect($connector);
+            $consume = (bool)($this->pipelines[$name]['consume'] ?? true);
+            return $this->create($connector, $consume);
         }
 
-        return $queue;
+        return $this->connect($connector);
     }
 
     /**

diff --git a/tests/src/Queue/RPCPipelineRegistryTest.php b/tests/src/Queue/RPCPipelineRegistryTest.php
@@ -119,20 +119,24 @@ public function testGetsExistsPipelineByAliasShouldReturnQueue(): void
         );
     }
 
-    public function testGetsNonExistsPipelineShouldThrowAnException(): void
+    public function testGetsNonExistsPipelineShouldReturnQueue(): void
     {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectErrorMessage('Queue pipeline with given name `test` is not found.');
+        $this->jobs->shouldReceive('connect')
+            ->once()
+            ->with('test')
+            ->andReturn($queue = m::mock(QueueInterface::class));
 
-        $this->registry->getPipeline('test');
+        $this->assertSame($queue, $this->registry->getPipeline('test'));
     }
 
-    public function testGetsNonExistsAliasPipelineShouldThrowAnException(): void
+    public function testGetsNonExistsAliasPipelineShouldReturnQueue(): void
     {
-        $this->expectException(InvalidArgumentException::class);
-        $this->expectErrorMessage('Queue pipeline with given name `test` is not found.');
+        $this->jobs->shouldReceive('connect')
+            ->once()
+            ->with('test')
+            ->andReturn($queue = m::mock(QueueInterface::class));
 
-        $this->registry->getPipeline('bad-alias');
+        $this->assertSame($queue, $this->registry->getPipeline('bad-alias'));
     }
 
     public function testGetsPipelineWithoutConnectorShouldThrowAnException(): void