Allow to configure ElasticSearch index in esb.yml

esb.yml.sample

@@ -30,6 +30,11 @@ services:
 flows:
   sample_flow:                          # The flow "code" and will be the Beanstalkd tube name
     description: Sample Flow            # The flow description
+    es_index:                           # Optional: the create/update ElasticSearch index API body (see https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html). This is useful if you want to control index mapping, settings and aliases.
+        settings:                       # For example you can set the total_fields limit to an higher (or lower) value:
+            index:
+                total_fields:
+                    limit: 2000
     producer:
       service: My\Esb\Producer          # A producer service ID defined above
       batch_size: 1200                  # Jobs are produced in batches of 1200 jobs. Optional: default is 1000

src/FlowConfiguration.php

@@ -24,6 +24,7 @@ public function getConfigTreeBuilder(): TreeBuilder
             ->arrayPrototype()
                 ->children()
                     ->scalarNode('description')->isRequired()->cannotBeEmpty()->end()
+                    ->variableNode('es_index')->defaultNull()->end()
                     ->arrayNode('producer')
                         ->children()
                             ->scalarNode('service')->isRequired()->end()

src/Model/FlowConfig.php

@@ -102,4 +102,12 @@ public function getProducerBatchSize(): int
     {
         return $this->config['producer']['batch_size'];
     }
+
+    /**
+     * @return array<array-key, mixed>|null
+     */
+    public function getElasticSearchIndexCreateOrUpdateBody(): ?array
+    {
+        return $this->config['es_index'] ?? null;
+    }
 }

src/Service/ElasticSearch.php

@@ -91,6 +91,14 @@ public function getClient(): Client
         return $this->client;
     }
 
+    /**
+     * @param array<array-key, mixed>|null $createOrUpdateBody
+     */
+    public function setElasticSearchIndex(string $indexName, array $createOrUpdateBody = null): Amp\Promise
+    {
+        return $this->client->createOrUpdateIndex($indexName, $createOrUpdateBody);
+    }
+
     /**
      * @param JobInterface $job
      * @param string $indexName

src/Service/QueueManager.php

@@ -72,6 +72,19 @@ public function __construct(
     public function boot(): Promise
     {
         return call(function () {
+            if ($this->flowConfig->getElasticSearchIndexCreateOrUpdateBody() !== null) {
+                yield $this->elasticSearch->setElasticSearchIndex(
+                    $this->flowConfig->getTube(),
+                    $this->flowConfig->getElasticSearchIndexCreateOrUpdateBody()
+                );
+                $this->logger->debug(
+                    'Successfully set ElasticSearch index',
+                    [
+                        'index' => $this->flowConfig->getTube(),
+                        'body' => $this->flowConfig->getElasticSearchIndexCreateOrUpdateBody()
+                    ]
+                );
+            }
             //Producer
             yield $this->beanstalkClient->use($this->flowConfig->getTube());