Merge pull request #14 from Flowpack/ah/add-option-for-partition-queues

TASK: optional supply of queuePartition when scheduling a pipeline

Author: Pingu501

Classes/PrunnerApiService.php

@@ -8,6 +8,7 @@
 use Flowpack\Prunner\Dto\PipelinesAndJobsResponse;
 use Flowpack\Prunner\ValueObject\JobId;
 use Flowpack\Prunner\ValueObject\PipelineName;
+use Flowpack\Prunner\ValueObject\QueuePartitionName;
 use GuzzleHttp\Client;
 use GuzzleHttp\Exception\GuzzleException;
 use Neos\Flow\Annotations as Flow;
@@ -91,14 +92,21 @@ public function loadJobLogs(JobId $jobId, string $taskName): JobLogs
     }
 
     /**
+     * @param QueuePartitionName|null $queuePartition only needed for queue_strategy "partitioned_replace"
      * @throws \JsonException
      */
-    public function schedulePipeline(PipelineName $pipeline, array $variables): JobId
+    public function schedulePipeline(PipelineName $pipeline, array $variables, ?QueuePartitionName $queuePartition = null): JobId
     {
-        $response = $this->apiCall('POST', 'pipelines/schedule', json_encode([
+        $requestBody = [
             'pipeline' => $pipeline->getName(),
-            'variables' => $variables
-        ], JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT));
+            'variables' => $variables,
+        ];
+
+        if ($queuePartition !== null) {
+            $requestBody['queuePartition'] = $queuePartition->getName();
+        }
+
+        $response = $this->apiCall('POST', 'pipelines/schedule', json_encode($requestBody, JSON_THROW_ON_ERROR | JSON_FORCE_OBJECT));
         if ($response->getStatusCode() !== 202) {
             throw new \RuntimeException('Scheduling a new pipeline run should have returned status code 202, but got: ' . $response->getStatusCode());
         }

Classes/ValueObject/QueuePartitionName.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flowpack\Prunner\ValueObject;
+
+use Neos\Flow\Annotations as Flow;
+
+/**
+ * @Flow\Proxy(false)
+ */
+class QueuePartitionName
+{
+    private string $name;
+
+    private function __construct(string $name)
+    {
+        $this->name = $name;
+    }
+
+    public static function create(string $name)
+    {
+        return new self($name);
+    }
+
+    public function getName(): string
+    {
+        return $this->name;
+    }
+}