itzap-elasticsearch client library

Vision Statement

Build 7.3.x elasticsearch embedded server into itzap–elasticsearch library to enable easy prototyping and unit/integration testing

How It Works

Elasticsearch embedded server starts a single elasticsearch Node. Elasticsearch server node will be listening for incoming requests on port 9200 Elasticsearch client can be configured to access embedded server using the following url: http://localhost:9200 Elasticsearch embedded server will create data and home temp folders that will be deleted opon application close.

Code

Here is the main class that implements elasticsearch embedded server node

public class EmbeddedElasticSearchServer implements IOData {
  private static final Logger LOGGER = LoggerFactory.getLogger(EmbeddedElasticSearchServer.class);

  private Node instance;
  private int port;
  private final AnyConfig config;

  public EmbeddedElasticSearchServer(AnyConfig config) {
    this.config = config;
  }

  private static class PluginConfigurableNode extends Node {
    PluginConfigurableNode(Settings input,
                           Map<String, String> properties,
                           Path configPath,
                           Supplier<String> defaultNodeName,
                           Collection<Class<? extends Plugin>> classpathPlugins) {
      super(InternalSettingsPreparer.prepareEnvironment(input, properties, configPath, defaultNodeName), classpathPlugins, false);
    }
  }

  @Override
  public synchronized Completable start() {
    return new RunnableCommand<Void>("cmd-start") {
      @Override
      protected Void run() {
        Settings settings = getSettings();

        instance = new PluginConfigurableNode(settings, ImmutableMap.of(),
                                              null, () -> config.getString(EsConfig.CLUSTER_NAME),
                                              singletonList(Netty4Plugin.class));
        try {
          instance.start();
          Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
              if (instance != null) {
                instance.close();
              }
            } catch (IOException e) {
              LOGGER.error("Error closing ElasticSearch");
            }
          }));
          LOGGER.info("ElasticSearch cluster {} started in local mode on port {}", instance.settings().get("cluster.name"),
                      port);
          return null;
        } catch (NodeValidationException e) {
          throw new IZapException("Failed to start embedded elastic search server", e);
        }
      }
    }.toCompletable();
  }

  @Override
  public synchronized Completable stop() {
    return new RunnableCommand<Void>("cmd-stop") {
      @Override
      protected Void run() {
        if (instance != null && !instance.isClosed()) {
          LOGGER.info("Stopping Elastic Search");
          try {
            instance.close();
            instance = null;
            LOGGER.info("Elastic Search on port {} stopped", port);
          } catch (IOException e) {
            throw new IZapException("Failed to close elastic search embedded server", e);
          }
        }

        return null;
      }
    }.toCompletable();
  }

  private Settings getSettings() {
    String clusterName = config.getString(EsConfig.CLUSTER_NAME);
    String host = config.getString(EsConfig.HOST);
    port = config.getInt(EsConfig.PORT);

    try {
      File dataDir = Files.createTempDirectory(clusterName + "_" + System.currentTimeMillis() + "data").toFile();
      FileUtils.forceDeleteOnExit(dataDir);
      cleanDataDir(dataDir.getAbsolutePath());

      File homeDir = Files.createTempDirectory(clusterName + "_" + System.currentTimeMillis() + "-home").toFile();
      cleanDataDir(homeDir.getAbsolutePath());
      FileUtils.forceDeleteOnExit(homeDir);

      Settings.Builder settingsBuilder = Settings.builder()
        .put("cluster.name", clusterName)
        .put("http.host", host)
        .put("http.port", port)
        .put("transport.tcp.port", port + 100)
        .put(EsConfig.DATA_PATH.getName(), dataDir.getAbsolutePath())
        .put(EsConfig.HOME_PATH.getName(), homeDir.getAbsolutePath())
        .put("http.cors.enabled", true)
        .put("node.data", true)
        .put("http.type", "netty4")
        .put("transport.type", "netty4");

      return settingsBuilder.build();
    } catch (IOException e) {
      throw new IZapException("Failed to create temp data/home dir.", e);
    }
  }
}

Elasticsearch embedded server implementation is using RunnableCommand command pattern to implement start and stopmethods. Please note that RunnableCommand returns Completable and not Observable. Sice Observable<Void> is no longer permitted in RxJava2 and start/stop methods have no return values. Here is the unit test for the elasticsearch embedded server

public class EmbeddedElasticSearchServerTest {
    private EmbeddedElasticSearchServer server;

    @Before
    public void setup() {
        AnyConfig config = ConfigBuilder.builder(ConfigType.TYPE_SAFE)
                .setFileName(this.getClass()
                        .getResource("/es-config.properties").getFile())
                .build();

        server = new EmbeddedElasticSearchServer(config);
    }

    @Test
    public void startTest() {
        server.start().blockingGet();
        server.stop().blockingGet();
    }
}

Elasticsearch dependencies

<dependency>
  <groupId>org.elasticsearch.client</groupId>
  <artifactId>elasticsearch-rest-high-level-client</artifactId>
  <version>7.3.1</version>
</dependency>

<dependency>
  <groupId>org.elasticsearch</groupId>
  <artifactId>elasticsearch</artifactId>
  <version>7.3.1</version>
</dependency>

<dependency>
  <groupId>org.elasticsearch.client</groupId>
  <artifactId>transport</artifactId>
  <version>7.3.1</version>
</dependency>

Vision Statement

How It Works

Code

Leave a Reply Cancel reply