package io.georocket.index.elasticsearch;

import java.io.IOException;

import org.apache.commons.exec.CommandLine;
import org.apache.commons.exec.DefaultExecuteResultHandler;
import org.apache.commons.exec.DefaultExecutor;
import org.apache.commons.exec.ExecuteException;
import org.apache.commons.exec.ExecuteWatchdog;
import org.apache.commons.exec.Executor;
import org.apache.commons.exec.ShutdownHookProcessDestroyer;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang.SystemUtils;

import io.georocket.constants.ConfigConstants;
import io.georocket.util.RxUtils;
import io.vertx.core.impl.NoStackTraceThrowable;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.rxjava.core.Vertx;
import rx.Observable;

/**
 * Runs an Elasticsearch instance
 * @author Michel Kraemer
 */
public class ElasticsearchRunner {
  private static Logger log = LoggerFactory.getLogger(ElasticsearchRunner.class);
  
  private final Vertx vertx;
  private Executor executor;
  private boolean stopped;
  
  /**
   * Create the Elasticsearch runner
   * @param vertx the Vert.x instance
   */
  public ElasticsearchRunner(Vertx vertx) {
    this.vertx = vertx;
  }
  
  /**
   * Run Elasticsearch
   * @param host the host Elasticsearch should bind to
   * @param port the port Elasticsearch should listen on
   * @param elasticsearchInstallPath the path where Elasticsearch is installed
   * @return an observable that emits exactly one item when Elasticsearch has started
   */
  public Observable<Void> runElasticsearch(String host, int port,
      String elasticsearchInstallPath) {
    JsonObject config = vertx.getOrCreateContext().config();
    String storage = config.getString(ConfigConstants.STORAGE_FILE_PATH);
    String root = storage + "/index";
    
    return vertx.<Void>rxExecuteBlocking(f -> {
      log.info("Starting Elasticsearch ...");
      
      // get Elasticsearch executable
      String executable = FilenameUtils.separatorsToSystem(
          elasticsearchInstallPath);
      executable = FilenameUtils.concat(executable, "bin");
      if (SystemUtils.IS_OS_WINDOWS) {
        executable = FilenameUtils.concat(executable, "elasticsearch.bat");
      } else {
        executable = FilenameUtils.concat(executable, "elasticsearch");
      }
      
      // start Elasticsearch
      CommandLine cmdl = new CommandLine(executable);
      cmdl.addArgument("-Ecluster.name=georocket-cluster");
      cmdl.addArgument("-Enode.name=georocket-node");
      cmdl.addArgument("-Enetwork.host=" + host);
      cmdl.addArgument("-Ehttp.port=" + port);
      cmdl.addArgument("-Edefault.path.home=" + elasticsearchInstallPath);
      cmdl.addArgument("-Epath.data=" + root + "/data");
      
      executor = new DefaultExecutor();
      executor.setProcessDestroyer(new ShutdownHookProcessDestroyer());
      executor.setWatchdog(new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT));
      
      try {
        executor.execute(cmdl, new DefaultExecuteResultHandler() {
          @Override
          public void onProcessComplete(final int exitValue) {
            log.info("Elasticsearch quit with exit code: " + exitValue);
          }
          
          @Override
          public void onProcessFailed(final ExecuteException e) {
            if (!stopped) {
              log.error("Elasticsearch execution failed", e);
            }
          }
        });
        f.complete();
      } catch (IOException e) {
        f.fail(e);
      }
    }).toObservable();
  }
  
  /**
   * Wait 60 seconds or until Elasticsearch is up and running, whatever
   * comes first
   * @param client the client to use to check if Elasticsearch is running
   * @return an observable that emits exactly one item when Elasticsearch
   * is running
   */
  public Observable<Void> waitUntilElasticsearchRunning(
      ElasticsearchClient client) {
    final Throwable repeat = new NoStackTraceThrowable("");
    return Observable.<Boolean>unsafeCreate(subscriber -> {
      client.isRunning().subscribe(subscriber);
    }).flatMap(running -> {
      if (!running) {
        return Observable.error(repeat);
      }
      return Observable.just(running);
    }).retryWhen(errors -> {
      Observable<Throwable> o = errors.flatMap(t -> {
        if (t == repeat) {
          // Elasticsearch is still not up, retry
          return Observable.just(t);
        }
        // forward error
        return Observable.error(t);
      });
      // retry for 60 seconds
      return RxUtils.makeRetry(60, 1000, null).call(o);
    }).map(r -> null);
  }
  
  /**
   * Stop a running Elasticsearch instance
   */
  public void stop() {
    if (executor != null && !stopped) {
      stopped = true;
      executor.getWatchdog().destroyProcess();
    }
  }
}

@@ -4022,59 +4022,22 @@
 ble.
-%3CBoolean%3EunsafeCreate(subscriber -%3E %7B%0A      
+defer(
 client
-.
+::
 isRu
@@ -4045,39 +4045,8 @@
 ning
-().subscribe(subscriber);%0A    %7D
 ).fl

Use defer() instead of unsafeCreate()