Merge branch 'master' into php83-support

composer.json

@@ -13,6 +13,7 @@
         "php": "~7.4.0|~8.0.0|~8.1.0|~8.2.0|~8.3.0",
         "ext-pcntl": "*",
         "ext-json": "*",
+        "ext-mbstring": "*",
         "symfony/dependency-injection": "^4.3",
         "symfony/config": "^4.3",
         "symfony/yaml": "^4.3",
@@ -33,7 +34,7 @@
         "symfony/property-info": "^4.3",
         "doctrine/annotations": "^1.8",
         "ramsey/uuid": "^3.8",
-        "webgriffe/amp-elasticsearch": "dev-php8-support as 2.1",
+        "webgriffe/amp-elasticsearch": "^2.1",
         "pagerfanta/pagerfanta": "^2.4",
         "symfony/deprecation-contracts": "^2.1",
         "amphp/http-server-form-parser": "^1.1"

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());