Adding support for analysers

Closes #3512930

src/Plugin/DatasetDestination/ElasticSearchDestination.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace Drupal\data_pipelines_elasticsearch\Plugin\DatasetDestination;
 
+use Drupal\Component\Utility\NestedArray;
 use Drupal\Component\Utility\UrlHelper;
 use Drupal\Core\Form\FormStateInterface;
 use Drupal\Core\Plugin\ContainerFactoryPluginInterface;
@@ -264,31 +265,29 @@ class ElasticSearchDestination extends DatasetDestinationPluginBase implements C
    *
    * @param string $index_id
    *   The index ID.
-   * @param array $mappings
-   *   The mapping information of the index.
+   * @param array $settings
+   *   The destination settings for the index.
    *
    * @throws \Drupal\data_pipelines_elasticsearch\Exception\CouldNotCreateIndexException
    *   When the index can't be created.
    */
-  protected function createIndex(string $index_id, array $mappings): void {
-    $index_params = [];
-    $index_params_base = [];
-    $index_params_base['index'] = $index_id;
-    $index_params['body'] = [
-      'settings' => [
-        'number_of_shards' => self::INDEX_SHARDS,
-        'number_of_replicas' => self::INDEX_REPLICAS,
-        'codec' => self::INDEX_CODEC,
+  protected function createIndex(string $index_id, array $settings): void {
+    $index_params = [
+      'index' => $index_id,
+      'body' => [
+        'settings' => [
+          'number_of_shards' => self::INDEX_SHARDS,
+          'number_of_replicas' => self::INDEX_REPLICAS,
+          'codec' => self::INDEX_CODEC,
+        ],
       ],
     ];
-    if (!empty($mappings)) {
-      $index_params['body']['mappings'] = $mappings;
-    }
+    $index_params = NestedArray::mergeDeep($index_params, ['body' => $settings]);
     try {
-      if ($this->getClient()->indices()->exists($index_params_base)->asBool()) {
+      if ($this->getClient()->indices()->exists(['index' => $index_id])->asBool()) {
         return;
       }
-      $this->getClient()->indices()->create($index_params_base + $index_params);
+      $this->getClient()->indices()->create($index_params);
     }
     catch (\Exception $e) {
       Error::logException($this->logger, $e);
@@ -338,10 +337,10 @@ class ElasticSearchDestination extends DatasetDestinationPluginBase implements C
    */
   public function beginProcessing(DatasetInterface $dataset): bool {
     parent::beginProcessing($dataset);
-    $mappings = $this->configuration['mappings'] ?? [];
     $index_id = $this->configuration['prefix'] . $dataset->getMachineName();
+    $settings = $dataset->getPipeline()->getDestinationSettings($this->getPluginId());
     try {
-      $this->createIndex($index_id, $mappings);
+      $this->createIndex($index_id, $settings);
       return TRUE;
     }
     catch (\Exception $e) {