Ref #31 Add bulk query string parameters support

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- New `setBulkRequestParams` on the Indexer allowing all the Bulk query params.
 - This changelog file.
 
 ## [1.0.2] - 2020-07-31

README.md

@@ -71,6 +71,12 @@ $dto->foo = 'Hops from Alsace, France';
 $indexer->scheduleIndex('beers', new Document('123', $dto));
 $indexer->flush();
 
+// Set parameters on the Bulk
+$indexer->setBulkRequestParams([
+    'pipeline' => 'covfefe',
+    'refresh' => 'wait_for'
+]);
+
 // Force index refresh if needed
 $indexer->refresh('beers');
 

src/Indexer.php

@@ -15,14 +15,16 @@ class Indexer
     private $serializer;
     /** @var Bulk|null */
     private $currentBulk = null;
+    private $bulkRequestParams = [];
 
-    public function __construct(Client $client, SerializerInterface $serializer, int $bulkMaxSize = 100)
+    public function __construct(Client $client, SerializerInterface $serializer, int $bulkMaxSize = 100, array $bulkRequestParams = [])
     {
         // TODO: on the destruct, maybe throw an exception for non empty indexer queues?
 
         $this->client = $client;
         $this->bulkMaxSize = $bulkMaxSize ?? 100;
         $this->serializer = $serializer;
+        $this->bulkRequestParams = $bulkRequestParams;
     }
 
     public function scheduleIndex($index, Document $document)
@@ -116,6 +118,7 @@ protected function getCurrentBulk(): Bulk
     {
         if (!($this->currentBulk instanceof Bulk)) {
             $this->currentBulk = new Bulk($this->client);
+            $this->refreshBulkRequestParams();
         }
 
         return $this->currentBulk;
@@ -134,6 +137,29 @@ protected function flushIfNeeded(): void
     {
         if ($this->getQueueSize() >= $this->bulkMaxSize) {
             $this->flush();
+
+        }
+    }
+
+    public function getBulkRequestParams(): array
+    {
+        return $this->bulkRequestParams;
+    }
+
+    public function setBulkRequestParams(array $bulkRequestParams): void
+    {
+        $this->bulkRequestParams = $bulkRequestParams;
+        $this->refreshBulkRequestParams();
+    }
+
+    private function refreshBulkRequestParams()
+    {
+        if (!$this->currentBulk) {
+            return;
+        }
+
+        foreach ($this->bulkRequestParams as $key => $value) {
+            $this->currentBulk->setRequestParam($key, $value);
         }
     }
 }

tests/IndexerTest.php

@@ -140,6 +140,40 @@ public function testIndexingWithError(): void
         $this->assertEquals(0, $indexer->getQueueSize());
     }
 
+    public function testRequestParameters(): void
+    {
+        $indexName = mb_strtolower(__FUNCTION__);
+        $dto = new TestDTO();
+        $dto->bar = 'I like unicorns.';
+        $dto->foo = 'Why is the sky blue?';
+
+        $client = $this->getClient();
+        $client->setConfigValue(Client::CONFIG_BULK_SIZE, 3);
+        $indexer = $client->getIndexer();
+        $indexer->setBulkRequestParams([
+            'refresh' => 'wait_for',
+        ]);
+
+        $indexer->scheduleIndex($indexName, new Document('1', $dto));
+        $response = $indexer->flush();
+
+        $this->assertInstanceOf(ResponseSet::class, $response);
+        $transferInfo = $response->getTransferInfo();
+        $this->assertStringContainsString('_bulk?refresh=wait_for', $transferInfo['url']);
+
+        // Test the same with an invalid pipeline
+        $indexer->setBulkRequestParams([
+            'pipeline' => 'covfefe',
+        ]);
+
+        $indexer->scheduleIndex($indexName, new Document('1', $dto));
+
+        $this->expectException(ResponseException::class);
+        $this->expectExceptionMessageRegExp('/pipeline with id \[covfefe\] does not exist/');
+
+        $indexer->flush();
+    }
+
     public function testIndexJsonString(): void
     {
         $indexName = mb_strtolower(__FUNCTION__);