From 9a9ce99364fcf7d77d81b5300530fac2ee435f37 Mon Sep 17 00:00:00 2001
From: kimchy <kimchy@gmail.com>
Date: Mon, 15 Feb 2010 10:23:53 +0200
Subject: [PATCH] Create Mapping API: Automatically create indices. Closes #12.

---
 .../create/TransportCreateMappingAction.java  | 42 ++++++++++++++++++-
 .../action/index/TransportIndexAction.java    |  2 +-
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/mapping/create/TransportCreateMappingAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/mapping/create/TransportCreateMappingAction.java
index efbed765596..33565b78cbe 100644
--- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/mapping/create/TransportCreateMappingAction.java
+++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/mapping/create/TransportCreateMappingAction.java
@@ -20,18 +20,25 @@
 package org.elasticsearch.action.admin.indices.mapping.create;
 
 import com.google.inject.Inject;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.Actions;
 import org.elasticsearch.action.TransportActions;
+import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.create.TransportCreateIndexAction;
 import org.elasticsearch.action.support.BaseAction;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.metadata.MetaDataService;
+import org.elasticsearch.indices.IndexAlreadyExistsException;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.*;
 import org.elasticsearch.util.io.VoidStreamable;
 import org.elasticsearch.util.settings.Settings;
 
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author kimchy (Shay Banon)
@@ -44,21 +51,54 @@ public class TransportCreateMappingAction extends BaseAction<CreateMappingReques
 
     private final MetaDataService metaDataService;
 
+    private final TransportCreateIndexAction createIndexAction;
+
     private final ThreadPool threadPool;
 
+    private final boolean autoCreateIndex;
+
     @Inject public TransportCreateMappingAction(Settings settings, TransportService transportService, ClusterService clusterService,
-                                                ThreadPool threadPool, MetaDataService metaDataService) {
+                                                ThreadPool threadPool, MetaDataService metaDataService, TransportCreateIndexAction createIndexAction) {
         super(settings);
         this.transportService = transportService;
         this.clusterService = clusterService;
         this.threadPool = threadPool;
         this.metaDataService = metaDataService;
+        this.createIndexAction = createIndexAction;
+        this.autoCreateIndex = settings.getAsBoolean("action.autoCreateIndex", true);
 
         transportService.registerHandler(TransportActions.Admin.Indices.Mapping.CREATE, new TransportHandler());
     }
 
     @Override protected void doExecute(final CreateMappingRequest request, final ActionListener<CreateMappingResponse> listener) {
         final String[] indices = Actions.processIndices(clusterService.state(), request.indices());
+        if (autoCreateIndex) {
+            final CountDownLatch latch = new CountDownLatch(indices.length);
+            for (String index : indices) {
+                if (!clusterService.state().metaData().hasIndex(index)) {
+                    createIndexAction.execute(new CreateIndexRequest(index), new ActionListener<CreateIndexResponse>() {
+                        @Override public void onResponse(CreateIndexResponse result) {
+                            latch.countDown();
+                        }
+
+                        @Override public void onFailure(Throwable e) {
+                            if (ExceptionsHelper.unwrapCause(e) instanceof IndexAlreadyExistsException) {
+                                latch.countDown();
+                            } else {
+                                listener.onFailure(e);
+                            }
+                        }
+                    });
+                } else {
+                    latch.countDown();
+                }
+            }
+            try {
+                latch.await(10, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                // ignore
+            }
+        }
         if (clusterService.state().nodes().localNodeMaster()) {
             threadPool.execute(new Runnable() {
                 @Override public void run() {
diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java
index 496d9d3a48e..cb59cec52f0 100644
--- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java
+++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java
@@ -53,7 +53,7 @@ public class TransportIndexAction extends TransportShardReplicationOperationActi
                                         TransportCreateIndexAction createIndexAction) {
         super(settings, transportService, clusterService, indicesService, threadPool, shardStateAction);
         this.createIndexAction = createIndexAction;
-        this.autoCreateIndex = componentSettings.getAsBoolean("autoCreateIndex", true);
+        this.autoCreateIndex = settings.getAsBoolean("action.autoCreateIndex", true);
         this.allowIdGeneration = componentSettings.getAsBoolean("allowIdGeneration", true);
     }