/**
 * If ES already contains this instance's target index, then do nothing.
 * Otherwise, create the index, then wait {@link #CREATE_SLEEP}.
 * <p>
 * The {@code client} field must point to a live, connected client.
 * The {@code indexName} field must be non-null and point to the name
 * of the index to check for existence or create.
 *
 * @param config the config for this ElasticSearchIndex
 * @throws java.lang.IllegalArgumentException if the index could not be created
 */
private void checkForOrCreateIndex(Configuration config) {
    Preconditions.checkState(null != client);

    //Create index if it does not already exist
    IndicesExistsResponse response = client.admin().indices().exists(new IndicesExistsRequest(indexName)).actionGet();
    if (!response.isExists()) {

        ImmutableSettings.Builder settings = ImmutableSettings.settingsBuilder();

        ElasticSearchSetup.applySettingsFromTitanConf(settings, config, ES_CREATE_EXTRAS_NS);

        CreateIndexResponse create = client.admin().indices().prepareCreate(indexName)
                .setSettings(settings.build()).execute().actionGet();
        try {
            final long sleep = config.get(CREATE_SLEEP);
            log.debug("Sleeping {} ms after {} index creation returned from actionGet()", sleep, indexName);
            Thread.sleep(sleep);
        } catch (InterruptedException e) {
            throw new TitanException("Interrupted while waiting for index to settle in", e);
        }
        if (!create.isAcknowledged()) throw new IllegalArgumentException("Could not create index: " + indexName);
    }
}
 

@Test
public void testDefaultsIcuAnalysis() {
    Index index = new Index("test");

    Settings settings = ImmutableSettings.settingsBuilder()
            .put("path.home", "none")
            .put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT)
            .build();

    Injector parentInjector = new ModulesBuilder().add(new SettingsModule(ImmutableSettings.EMPTY), new EnvironmentModule(new Environment(settings)), new IndicesAnalysisModule()).createInjector();
    Injector injector = new ModulesBuilder().add(
            new IndexSettingsModule(index, settings),
            new IndexNameModule(index),
            new AnalysisModule(ImmutableSettings.EMPTY, parentInjector.getInstance(IndicesAnalysisService.class)).addProcessor(new IKAnalysisBinderProcessor()))
            .createChildInjector(parentInjector);

    AnalysisService analysisService = injector.getInstance(AnalysisService.class);

    TokenizerFactory tokenizerFactory = analysisService.tokenizer("ik_tokenizer");
    MatcherAssert.assertThat(tokenizerFactory, instanceOf(IKTokenizerFactory.class));


}
 

@Override
public void init(Map<String, String> props) {
    String host = props.get(PropConst.HOST);
    if (host == null) {
        throw new IllegalArgumentException("Host does not specified");
    }
    String port = props.get(PropConst.PORT);
    if (port == null) {
        throw new IllegalArgumentException("Port does not specified");
    }
    String clusterName = props.get(PropConst.CLUSTER_NAME);
    if (clusterName == null) {
        throw new IllegalArgumentException("Cluster name does not specified");
    }
    client = (new TransportClient(ImmutableSettings.settingsBuilder().put("cluster.name", clusterName).build()))
            .addTransportAddress(new InetSocketTransportAddress(host, Integer.valueOf(port)));
}
 

@Test
public void testClusterSrvDiscoveryWith5Nodes() throws Exception {
    ImmutableSettings.Builder b = settingsBuilder()
        .put("node.mode", "network")
        .put("discovery.zen.ping.multicast.enabled", "false")
        .put("discovery.type", "srvtest")
        .put(SrvUnicastHostsProvider.DISCOVERY_SRV_QUERY, Constants.TEST_QUERY);

    assertEquals(cluster().size(), 0);

    internalCluster().startNode(b.put("transport.tcp.port", String.valueOf(Constants.NODE_0_TRANSPORT_TCP_PORT)).build());
    internalCluster().startNode(b.put("transport.tcp.port", String.valueOf(Constants.NODE_1_TRANSPORT_TCP_PORT)).build());
    internalCluster().startNode(b.put("transport.tcp.port", String.valueOf(Constants.NODE_2_TRANSPORT_TCP_PORT)).build());
    internalCluster().startNode(b.put("transport.tcp.port", String.valueOf(Constants.NODE_3_TRANSPORT_TCP_PORT)).build());
    internalCluster().startNode(b.put("transport.tcp.port", String.valueOf(Constants.NODE_4_TRANSPORT_TCP_PORT)).build());

    assertEquals(cluster().size(), 5);
}
 

public Client createClient() {
    final AtomicReference<Client> elasticsearchClient = new AtomicReference<>();
    await().atMost(30, TimeUnit.SECONDS).pollDelay(1, TimeUnit.SECONDS).until(() -> {
        Client c = new TransportClient(ImmutableSettings.settingsBuilder().put("cluster.name", elasticsearchClusterName).build()).addTransportAddress(new InetSocketTransportAddress(getIpAddress(), 9300));
        try {
            c.admin().cluster().health(Requests.clusterHealthRequest("_all")).actionGet();
        } catch (ElasticsearchException e) {
            c.close();
            return false;
        }
        elasticsearchClient.set(c);
        return true;
    });
    assertEquals(elasticsearchClusterName, elasticsearchClient.get().admin().cluster().health(Requests.clusterHealthRequest("_all")).actionGet().getClusterName());
    return elasticsearchClient.get();
}
 

/**
 * インデックスを作成する.
 * @param index インデックス名
 * @param mappings マッピング情報
 * @return 非同期応答
 */
public ActionFuture<CreateIndexResponse> createIndex(String index, Map<String, JSONObject> mappings) {
    this.fireEvent(Event.creatingIndex, index);
    CreateIndexRequestBuilder cirb =
            new CreateIndexRequestBuilder(esTransportClient.admin().indices()).setIndex(index);

    // cjkアナライザ設定
    ImmutableSettings.Builder indexSettings = ImmutableSettings.settingsBuilder();
    indexSettings.put("analysis.analyzer.default.type", "cjk");
    cirb.setSettings(indexSettings);

    if (mappings != null) {
        for (Map.Entry<String, JSONObject> ent : mappings.entrySet()) {
            cirb = cirb.addMapping(ent.getKey(), ent.getValue().toString());
        }
    }
    return cirb.execute();
}
 

/**
 * テストケース共通の初期化処理. テスト用のElasticsearchのNodeを初期化する
 * @throws Exception 異常が発生した場合の例外
 */
@BeforeClass
public static void setUpBeforeClass() throws Exception {
    Settings settings = ImmutableSettings.settingsBuilder()
            // .put("node.http.enabled", false)
            .put("cluster.name", TESTING_CLUSTER)
            .put("node.name", "node1")
            .put("gateway.type", "none")
            .put("action.auto_create_index", "false")
            .put("index.store.type", "memory")
            .put("index.number_of_shards", 1)
            .put("index.number_of_replicas", 0)
            .put("transport.tcp.port", "9399")
            .build();
    node = NodeBuilder.nodeBuilder().settings(settings).node();

    esClient = new EsClient(TESTING_CLUSTER, TESTING_HOSTS);
}
 

/**
 * テストケース共通の初期化処理. テスト用のElasticsearchのNodeを初期化する
 * @throws Exception 異常が発生した場合の例外
 */
@BeforeClass
public static void setUpBeforeClass() throws Exception {
    Settings settings = ImmutableSettings.settingsBuilder()
            .put("node.http.enabled", true)
            .put("cluster.name", TESTING_CLUSTER)
            .put("node.name", "node1")
            .put("gateway.type", "none")
            .put("action.auto_create_index", "false")
            .put("index.store.type", "memory")
            .put("index.number_of_shards", 1)
            .put("index.number_of_replicas", 0)
            .put("transport.tcp.port", "9399")
            .build();
    node = NodeBuilder.nodeBuilder().settings(settings).node();

    esClient = new EsClient(TESTING_CLUSTER, TESTING_HOSTS);
}
 

/**
 * テストケース共通の初期化処理. テスト用のElasticsearchのNodeを初期化する
 * @throws Exception 異常が発生した場合の例外
 */
@BeforeClass
public static void setUpBeforeClass() throws Exception {
    Settings settings = ImmutableSettings.settingsBuilder()
            // .put("node.http.enabled", false)
            .put("cluster.name", TESTING_CLUSTER)
            .put("node.name", "node1")
            .put("gateway.type", "none")
            .put("action.auto_create_index", "false")
            .put("index.store.type", "memory")
            .put("index.number_of_shards", 1)
            .put("index.number_of_replicas", 0)
            .put("transport.tcp.port", "9399")
            .build();
    node = NodeBuilder.nodeBuilder().settings(settings).node();

    esClient = new EsClient(TESTING_CLUSTER, TESTING_HOSTS);
}
 

public ElasticsearchEmitter(KinesisConnectorConfiguration configuration) {
    Settings settings =
            ImmutableSettings.settingsBuilder()
                    .put(ELASTICSEARCH_CLUSTER_NAME_KEY, configuration.ELASTICSEARCH_CLUSTER_NAME)
                    .put(ELASTICSEARCH_CLIENT_TRANSPORT_SNIFF_KEY, configuration.ELASTICSEARCH_TRANSPORT_SNIFF)
                    .put(ELASTICSEARCH_CLIENT_TRANSPORT_IGNORE_CLUSTER_NAME_KEY,
                            configuration.ELASTICSEARCH_IGNORE_CLUSTER_NAME)
                    .put(ELASTICSEARCH_CLIENT_TRANSPORT_PING_TIMEOUT_KEY, configuration.ELASTICSEARCH_PING_TIMEOUT)
                    .put(ELASTICSEARCH_CLIENT_TRANSPORT_NODES_SAMPLER_INTERVAL_KEY,
                            configuration.ELASTICSEARCH_NODE_SAMPLER_INTERVAL)
                    .build();
    elasticsearchEndpoint = configuration.ELASTICSEARCH_ENDPOINT;
    elasticsearchPort = configuration.ELASTICSEARCH_PORT;
    LOG.info("ElasticsearchEmitter using elasticsearch endpoint " + elasticsearchEndpoint + ":" + elasticsearchPort);
    elasticsearchClient = new TransportClient(settings);
    elasticsearchClient.addTransportAddress(new InetSocketTransportAddress(elasticsearchEndpoint, elasticsearchPort));
}
 

@Test
public void assertTop1OneShard() {
    client.admin().indices().prepareCreate("topk-0").setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)).execute().actionGet();
    
    client.prepareIndex("topk-0", "type0", "doc0").setSource("field0", "foo").execute().actionGet();
    client.prepareIndex("topk-0", "type0", "doc1").setSource("field0", "foo").execute().actionGet();
    client.prepareIndex("topk-0", "type0", "doc2").setSource("field0", "bar").setRefresh(true).execute().actionGet();
   
    SearchResponse searchResponse = client.prepareSearch("topk-0")
            .setQuery(matchAllQuery())
            .addAggregation(new TopKBuilder("topk").field("field0").size(1))
            .execute().actionGet();
    TopK topk = searchResponse.getAggregations().get("topk");
    assertNotNull(topk);
    List<TopK.Bucket> buckets = new ArrayList<>(topk.getBuckets());
    assertEquals(1, buckets.size());
    assertEquals("foo", buckets.get(0).getKey());
    assertEquals(2, buckets.get(0).getDocCount());
}
 

@Test
public void assertTop1NumericalOneShard() {
    client.admin().indices().prepareCreate("topk-0n").setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)).execute().actionGet();
    
    client.prepareIndex("topk-0n", "type0", "doc0").setSource("field0", 42).execute().actionGet();
    client.prepareIndex("topk-0n", "type0", "doc1").setSource("field0", 42).execute().actionGet();
    client.prepareIndex("topk-0n", "type0", "doc2").setSource("field0", 51).setRefresh(true).execute().actionGet();
   
    SearchResponse searchResponse = client.prepareSearch("topk-0n")
            .setQuery(matchAllQuery())
            .addAggregation(new TopKBuilder("topk").field("field0").size(1))
            .execute().actionGet();
    TopK topk = searchResponse.getAggregations().get("topk");
    assertNotNull(topk);
    List<TopK.Bucket> buckets = new ArrayList<>(topk.getBuckets());
    assertEquals(1, buckets.size());
    assertEquals("42", buckets.get(0).getKey());
    assertEquals(2, buckets.get(0).getDocCount());
}
 

@Test
public void assertTop1BooleanOneShard() {
    client.admin().indices().prepareCreate("topk-0b").setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)).execute().actionGet();
    
    client.prepareIndex("topk-0b", "type0", "doc0").setSource("field0", true).execute().actionGet();
    client.prepareIndex("topk-0b", "type0", "doc1").setSource("field0", true).execute().actionGet();
    client.prepareIndex("topk-0b", "type0", "doc2").setSource("field0", false).setRefresh(true).execute().actionGet();
   
    SearchResponse searchResponse = client.prepareSearch("topk-0b")
            .setQuery(matchAllQuery())
            .addAggregation(new TopKBuilder("topk").field("field0").size(1))
            .execute().actionGet();
    TopK topk = searchResponse.getAggregations().get("topk");
    assertNotNull(topk);
    List<TopK.Bucket> buckets = new ArrayList<>(topk.getBuckets());
    assertEquals(1, buckets.size());
    assertEquals("T", buckets.get(0).getKey());
    assertEquals(2, buckets.get(0).getDocCount());
}
 

@Test
public void assertTop10of3OneShard() {
    client.admin().indices().prepareCreate("topk-1").setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)).execute().actionGet();
    
    client.prepareIndex("topk-1", "type0", "doc0").setSource("field0", "foo").execute().actionGet();
    client.prepareIndex("topk-1", "type0", "doc1").setSource("field0", "foo").execute().actionGet();
    client.prepareIndex("topk-1", "type0", "doc2").setSource("field0", "bar").setRefresh(true).execute().actionGet();
   
    SearchResponse searchResponse = client.prepareSearch("topk-1")
            .setQuery(matchAllQuery())
            .addAggregation(new TopKBuilder("topk").field("field0").size(10))
            .execute().actionGet();
    TopK topk = searchResponse.getAggregations().get("topk");
    assertNotNull(topk);
    List<TopK.Bucket> buckets = new ArrayList<>(topk.getBuckets());
    assertEquals(2, buckets.size());
    assertEquals("foo", buckets.get(0).getKey());
    assertEquals(2, buckets.get(0).getDocCount());
    assertEquals("bar", buckets.get(1).getKey());
    assertEquals(1, buckets.get(1).getDocCount());
}
 

@Test
public void assertTop10of50OneShard() {
    client.admin().indices().prepareCreate("topk-2").setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 1)).execute().actionGet();
    
    for (int i = 0; i < 50; ++i) { // 50 values
        client.prepareIndex("topk-2", "type0", "doc" + i).setSource("field0", "foo" + i).execute().actionGet();
    }
    client.prepareIndex("topk-2", "type0", "doc50").setSource("field0", "foo0").setRefresh(true).execute().actionGet(); // foo0 twice
   
    SearchResponse searchResponse = client.prepareSearch("topk-2")
            .setQuery(matchAllQuery())
            .addAggregation(new TopKBuilder("topk").field("field0").size(10))
            .execute().actionGet();
    TopK topk = searchResponse.getAggregations().get("topk");
    assertNotNull(topk);
    List<TopK.Bucket> buckets = new ArrayList<>(topk.getBuckets());
    assertEquals(10, buckets.size());
    assertEquals("foo0", buckets.get(0).getKey());
    assertEquals(2, buckets.get(0).getDocCount());
    for (int i = 1; i < 10; ++i) {
        assertEquals(1, buckets.get(i).getDocCount());
    }
}
 

@Test
public void assertTop10of50TwoShard() {
    client.admin().indices().prepareCreate("topk-3").setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_shards", 2)).execute().actionGet();
    
    for (int i = 0; i < 50; ++i) { // 50 values
        client.prepareIndex("topk-3", "type0", "doc" + i).setSource("field0", "foo" + i).setRefresh(true).execute().actionGet();
    }
    for (int i = 50; i < 100; ++i) { // 50 same values
        client.prepareIndex("topk-3", "type0", "doc" + i).setSource("field0", "foo0").setRefresh(true).execute().actionGet();
    }
   
    SearchResponse searchResponse = client.prepareSearch("topk-3")
            .setQuery(matchAllQuery())
            .addAggregation(new TopKBuilder("topk").field("field0").size(10))
            .execute().actionGet();
    assertEquals(100, searchResponse.getHits().getTotalHits());
    TopK topk = searchResponse.getAggregations().get("topk");
    assertNotNull(topk);
    List<TopK.Bucket> buckets = new ArrayList<>(topk.getBuckets());
    assertEquals(10, buckets.size());
    assertEquals("foo0", buckets.get(0).getKey());
    assertEquals(51, buckets.get(0).getDocCount());
}
 

@Before
public void setUp() {
    if (interp == null) {
        resetInterpreter();
    }
    Settings s1 = ImmutableSettings.settingsBuilder()
            .put("cluster.name", "a")
            .put("node.local", false)
            .build();

    esSetup = new StoreEsSetup(s1);
    esSetup.execute(deleteAll(), createIndex("users").withSettings(
            fromClassPath("essetup/settings/test_a.json")).withMapping("d",
            fromClassPath("essetup/mappings/test_a.json")).withData(
            fromClassPath("essetup/data/test_a.json")));
    esSetup.client().admin().indices().prepareRefresh("users").execute().actionGet();

    Settings s2 = ImmutableSettings.settingsBuilder()
            .put("cluster.name", "b")
            .put("node.local", false)
            .build();
    esSetup2 = new StoreEsSetup(s2);
    esSetup2.execute(deleteAll());
}
 

@BeforeSuite
public static void init() throws IOException, ExecutionException, InterruptedException, ParseException {

    File file = new File(DB_FOLDER_NAME);
    FileUtils.deleteDirectory(file);

    Settings settings = ImmutableSettings.settingsBuilder()
            .put("path.logs", "")
            .put("path.data", DB_FOLDER_NAME)
            .build();

    node = nodeBuilder().settings(settings).data(true).local(true).clusterName(HOST).node();
    client = node.client();

    LOG.info("Started local node at " + DB_FOLDER_NAME + " settings " + node.settings().getAsMap());

}
 

protected Client createClient() {
	if (client == null) {
		if (logger.isDebugEnabled()) {
			logger.debug("Creating client for Search!");
		}
		// Try starting search client at context loading
		try {
			final Settings settings = ImmutableSettings
					.settingsBuilder()
					.put(ElasticSearchReservedWords.CLUSTER_NAME.getText(),
							searchServerClusterName).build();
			TransportClient transportClient = new TransportClient(settings);
			transportClient = transportClient
					.addTransportAddress(new InetSocketTransportAddress(
							"localhost", 9300));
			if (transportClient.connectedNodes().size() == 0) {
				logger.error("There are no active nodes available for the transport, it will be automatically added once nodes are live!");
			}
			client = transportClient;
		} catch (final Exception ex) {
			// ignore any exception, dont want to stop context loading
			logger.error("Error occured while creating search client!", ex);
		}
	}
	return client;
}
 

/**
	 * 创建TransportClient
	 * @return
	 */
	private static Client createTransportClient() {
		//创建settings
		Settings settings = ImmutableSettings.settingsBuilder()
			.put("cluster.name", "elasticsearch")//设置集群名称
//		    .put("shield.user", "admin:sysadmin")
			.build();
		Client client = null;
		try {
			client = new org.elasticsearch.client.transport.TransportClient(settings)
				.addTransportAddress(new InetSocketTransportAddress("127.0.0.1", 9300));
		} catch (Exception e) {
			e.printStackTrace();
		}
		return client;
	}
 

@Bean
public Client elasticsearchClient(){

    Settings settings = ImmutableSettings.settingsBuilder()
            .put("client.transport.ignore_cluster_name", true)
            .put("cluster.name", "elasticsearch")
            .build();
    TransportClient tc = new TransportClient(settings);

    for (String elasticSearchHost : configurationContext.getElasticSearchHosts()) {
        String[] elements = elasticSearchHost.split(":");
        tc.addTransportAddress(new InetSocketTransportAddress(elements[0], Integer.parseInt(elements[1])));
    }


    log.debug("Creating Spring Bean for elasticsearchClient");

    return  tc;

}
 

void createNodes() throws Exception {
  Settings settings = ImmutableSettings
      .settingsBuilder()
      .put("number_of_shards", 1)
      .put("number_of_replicas", 0)
      .put("routing.hash.type", "simple")
      .put("gateway.type", "none")
      .put("path.data", "target/es-test")
      .build();

  node = NodeBuilder.nodeBuilder().settings(settings).local(true).node();
  client = node.client();

  client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute()
      .actionGet();
}
 

/**
 * Open client to elaticsearch cluster
 * 
 * @param clusterName
 */
private void openClient(String clusterName) {
  logger.info("Using ElasticSearch hostnames: {} ",
      Arrays.toString(serverAddresses));
  Settings settings = ImmutableSettings.settingsBuilder()
      .put("cluster.name", clusterName).build();

  TransportClient transportClient = new TransportClient(settings);
  for (InetSocketTransportAddress host : serverAddresses) {
    transportClient.addTransportAddress(host);
  }
  if (client != null) {
    client.close();
  }
  client = transportClient;
}
 

void createNodes() throws Exception {
  Settings settings = ImmutableSettings
      .settingsBuilder()
      .put("number_of_shards", 1)
      .put("number_of_replicas", 0)
      .put("routing.hash.type", "simple")
      .put("gateway.type", "none")
      .put("path.data", "target/es-test")
      .build();

  node = NodeBuilder.nodeBuilder().settings(settings).local(true).node();
  client = node.client();

  client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute()
      .actionGet();
}
 

public void testDefaults()
{
	Map<String, Object> map = new HashMap<>();		
	RiverSettings settings = new RiverSettings(
		ImmutableSettings.settingsBuilder().build(), 
		map);
	KafkaRiverConfig c = new KafkaRiverConfig(settings);
	
	assertEquals("localhost", c.brokerHost);
	assertEquals(9092, c.brokerPort);
	assertEquals("localhost", c.zookeeper);
	assertEquals(10485760, c.bulkSize);
	assertEquals(0, c.partition);
	assertEquals("default_topic", c.topic);
	assertEquals(10000, c.bulkTimeout.millis());
	
	assertEquals("org.elasticsearch.river.kafka.JsonMessageHandlerFactory", c.factoryClass);
	
	assertEquals(-1, c.statsdPort);
	assertNull(c.statsdHost);
	assertNull(c.statsdPrefix);
	
}
 

protected ShellNativeClient newTransportClient(TransportAddress... addresses) {

        Settings settings = ImmutableSettings.settingsBuilder().put("client.transport.ignore_cluster_name", true).build();
        org.elasticsearch.client.transport.TransportClient client = new TransportClient(settings).addTransportAddresses(addresses);

        //if no connected node we can already close the (useless) client
        if (client.connectedNodes().size() == 0) {
            client.close();
            return null;
        }

        AbstractClient<TransportClient, JsonInput, JsonOutput> shellClient = clientWrapper.wrapEsTransportClient(client);
        resourceRegistry.registerResource(shellClient);
        ShellNativeClient shellNativeClient = clientWrapper.wrapShellClient(shellClient);
        clientScopeSynchronizerRunner.startSynchronizer(shellNativeClient);
        return shellNativeClient;
    }
 

@Override
public Client start() {
    transportPort = elasticSearchFig.getTransportPort();
    httpPort = elasticSearchFig.getHttpPort();
    host = elasticSearchFig.getTransportHost();
    clusterName = elasticSearchFig.getClusterName();

    Settings settings = ImmutableSettings.settingsBuilder().put( "cluster.name", clusterName ).build();
    LOG.info( "Connecting Elasticsearch on {}", elasticSearchFig.getTransportHost() + ":" +
            elasticSearchFig.getTransportPort() );
    nodeList = getNodeList();
    TransportClient transportClient = new TransportClient( settings );
    for ( ElasticSearchNode elasticSearchNode : nodeList ) {
        LOG.debug( "Adding transport address with host {} and port {}", elasticSearchNode.getTransportHost()
                , elasticSearchNode.getTransportPort() );
        transportClient.addTransportAddress( new InetSocketTransportAddress( elasticSearchNode.getTransportHost(),
                elasticSearchNode.getTransportPort() ) );
    }

    client = transportClient;
    return client;
}
 

protected Settings buildNodeSettings() {
    // Build settings
    ImmutableSettings.Builder builder = ImmutableSettings.settingsBuilder()
            .put("node.name", "node-test-" + System.currentTimeMillis())
            .put("node.data", true)
            .put("cluster.name", "cluster-test-" + NetworkUtils.getLocalAddress().getHostName())
            .put("path.data", "./target/elasticsearch-test/data")
            .put("path.work", "./target/elasticsearch-test/work")
            .put("path.logs", "./target/elasticsearch-test/logs")
            .put("index.number_of_shards", "1")
            .put("index.number_of_replicas", "0")
            .put("cluster.routing.schedule", "50ms")
            .put("node.local", true);

    if (settings != null) {
        builder.put(settings);
    }

    return builder.build();
}
 

@Override
public void start(File tmpDataFolder, String clusterName) throws Exception {
	if (node == null) {
		node = nodeBuilder()
			.settings(ImmutableSettings.settingsBuilder()
				.put("http.enabled", false)
				.put("path.data", tmpDataFolder.getAbsolutePath()))
			.clusterName(clusterName)
			.local(true)
			.node();

		node.start();
	}
}
 

public EsStore(String host, int port) {
  String clusterName = Configuration.getProperties().getString("es_cluster_name", "soundwave");
  Settings settings = ImmutableSettings.settingsBuilder().put("cluster.name", clusterName)
      .put("client.transport.sniff", false).build();
  esClient = new TransportClient(settings)
      .addTransportAddress(
          new InetSocketTransportAddress(host, port));
}