Refactored Pipeline processor handling (#2218)

Refactored processor handling to more closely resemble what Elasticsearch ingest pipeline endpoint expects. Fixes #1810
ruflin · Jun 26, 2024 · 8b1826d · 8b1826d
1 parent 2fc446e
commit 8b1826d
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 29 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Removed
 
 ### Fixed
+* Fixed Pipeline Processor handling to allow for multiple processors of the same type [#2218](https://github.com/ruflin/Elastica/pull/2218)
 
 ### Security
 

diff --git a/src/Pipeline.php b/src/Pipeline.php
@@ -28,16 +28,10 @@ class Pipeline extends Param
      */
     protected $_client;
 
-    /**
-     * @var AbstractProcessor[]
-     *
-     * @phpstan-var array{processors?: AbstractProcessor[]}
-     */
-    protected $_processors = [];
-
     public function __construct(Client $client)
     {
         $this->_client = $client;
+        $this->setParam('processors', []);
     }
 
     /**
@@ -55,7 +49,7 @@ public function create(): Response
             throw new InvalidException('You should set a valid processor description.');
         }
 
-        if (empty($this->_processors['processors'])) {
+        if (empty($this->_params['processors'])) {
             throw new InvalidException('You should set a valid processor of type Elastica\Processor\AbstractProcessor.');
         }
 
@@ -95,19 +89,14 @@ public function deletePipeline(string $id): Response
      */
     public function setRawProcessors(array $processors): self
     {
-        $this->_processors = $processors;
+        $this->setParam('processors', $processors);
 
         return $this;
     }
 
     public function addProcessor(AbstractProcessor $processor): self
     {
-        if (!$this->_processors) {
-            $this->_processors['processors'] = $processor->toArray();
-            $this->_params['processors'] = [];
-        } else {
-            $this->_processors['processors'] = \array_merge($this->_processors['processors'], $processor->toArray());
-        }
+        $this->setParam('processors', \array_merge($this->getParam('processors'), [$processor->toArray()]));
 
         return $this;
     }
@@ -129,7 +118,9 @@ public function getId(): ?string
      */
     public function setProcessors(array $processors): self
     {
-        $this->setParam('processors', [$processors]);
+        foreach ($processors as $processor) {
+            $this->addProcessor($processor);
+        }
 
         return $this;
     }
@@ -148,8 +139,6 @@ public function setDescription(string $description): self
      */
     public function toArray(): array
     {
-        $this->_params['processors'] = [$this->_processors['processors']];
-
         return $this->getParams();
     }
 

diff --git a/tests/PipelineTest.php b/tests/PipelineTest.php
@@ -36,19 +36,25 @@ public function testProcessor(): void
 
         $expected = [
             'description' => 'this is a new pipeline',
-            'processors' => [[
-                'trim' => [
-                    'field' => 'field1',
+            'processors' => [
+                [
+                    'trim' => [
+                        'field' => 'field1',
+                    ],
                 ],
-                'rename' => [
-                    'field' => 'foo',
-                    'target_field' => 'target.field',
+                [
+                    'rename' => [
+                        'field' => 'foo',
+                        'target_field' => 'target.field',
+                    ],
                 ],
-                'set' => [
-                    'field' => 'field4',
-                    'value' => 324,
+                [
+                    'set' => [
+                        'field' => 'field4',
+                        'value' => 324,
+                    ],
                 ],
-            ]],
+            ],
         ];
 
         $this->assertEquals($expected, $processors->toArray());
@@ -79,7 +85,8 @@ public function testPipelineCreate(): void
         $this->assertSame('pipeline for Set', $result['my_custom_pipeline']['description']);
         $this->assertSame('field4', $result['my_custom_pipeline']['processors'][0]['set']['field']);
         $this->assertSame(333, $result['my_custom_pipeline']['processors'][0]['set']['value']);
-        $this->assertSame('field1', $result['my_custom_pipeline']['processors'][0]['trim']['field']);
+        $this->assertSame('field1', $result['my_custom_pipeline']['processors'][1]['trim']['field']);
+        $this->assertSame('foo', $result['my_custom_pipeline']['processors'][2]['rename']['field']);
     }
 
     /**