/**
 * Prepare a Runnable to send an index request to store the element, invalidates the cache on success
 */
private void store(FeatureStoreRequest request, Task task, ActionListener<FeatureStoreResponse> listener) {

    try {
        Optional<ClearCachesNodesRequest> clearCachesNodesRequest = buildClearCache(request);
        IndexRequest indexRequest = buildIndexRequest(task, request);
        client.execute(IndexAction.INSTANCE, indexRequest, wrap(
                (r) -> {
                    // Run and forget, log only if something bad happens
                    // but don't wait for the action to be done nor set the parent task.
                    clearCachesNodesRequest.ifPresent((req) -> clearCachesAction.execute(req, wrap(
                            (r2) -> {
                            },
                            (e) -> logger.error("Failed to clear cache", e))));
                    listener.onResponse(new FeatureStoreResponse(r));
                },
                listener::onFailure));
    } catch (IOException ioe) {
        listener.onFailure(ioe);
    }
}
 

@Test
public void test() throws Exception {
    try {
        DeleteIndexRequestBuilder deleteIndexRequestBuilder = new DeleteIndexRequestBuilder(client("1"), DeleteIndexAction.INSTANCE, "test");
        deleteIndexRequestBuilder.execute().actionGet();
    } catch (Exception e) {
        // ignore
    }
    IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client("1"), IndexAction.INSTANCE);
    indexRequestBuilder
            .setIndex("test")
            .setType("test")
            .setId("1")
            .setSource(jsonBuilder().startObject().field("field", "1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8").endObject())
            .setRefresh(true)
            .execute()
            .actionGet();
    String doc = client("1").prepareSearch("test")
            .setTypes("test")
            .setQuery(matchQuery("field", "1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8"))
            .execute()
            .actionGet()
            .getHits().getAt(0).getSourceAsString();

    assertEquals(doc, "{\"field\":\"1%2fPJJP3JV2C24iDfEu9XpHBaYxXh%2fdHTbmchB35SDznXO2g8Vz4D7GTIvY54iMiX_149c95f02a8\"}");
}
 

@Override
public void createElasticsearchClient(ProcessContext context) throws ProcessException {
    final Client mockClient = mock(Client.class);
    BulkRequestBuilder bulkRequestBuilder = spy(new BulkRequestBuilder(mockClient, BulkAction.INSTANCE));
    if (exceptionToThrow != null) {
        doThrow(exceptionToThrow).when(bulkRequestBuilder).execute();
    } else {
        doReturn(new MockBulkRequestBuilderExecutor(responseHasFailures)).when(bulkRequestBuilder).execute();
    }
    when(mockClient.prepareBulk()).thenReturn(bulkRequestBuilder);

    when(mockClient.prepareIndex(anyString(), anyString(), anyString())).thenAnswer(new Answer<IndexRequestBuilder>() {
        @Override
        public IndexRequestBuilder answer(InvocationOnMock invocationOnMock) throws Throwable {
            Object[] args = invocationOnMock.getArguments();
            String arg1 = (String) args[0];
            if (arg1.isEmpty()) {
                throw new NoNodeAvailableException("Needs index");
            }
            String arg2 = (String) args[1];
            if (arg2.isEmpty()) {
                throw new NoNodeAvailableException("Needs doc type");
            } else {
                IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(mockClient, IndexAction.INSTANCE);
                return indexRequestBuilder;
            }
        }
    });

    esClient.set(mockClient);
}
 

@Override
protected Client getTransportClient(Settings.Builder settingsBuilder, String xPackPath,
                                    String username, String password,
                                    List<InetSocketAddress> esHosts, ComponentLog log)
        throws MalformedURLException {
    final Client mockClient = mock(Client.class);
    BulkRequestBuilder bulkRequestBuilder = spy(new BulkRequestBuilder(mockClient, BulkAction.INSTANCE));
    if (exceptionToThrow != null) {
        doThrow(exceptionToThrow).when(bulkRequestBuilder).execute();
    } else {
        doReturn(new MockBulkRequestBuilderExecutor(responseHasFailures, esHosts.get(0))).when(bulkRequestBuilder).execute();
    }
    when(mockClient.prepareBulk()).thenReturn(bulkRequestBuilder);

    when(mockClient.prepareIndex(anyString(), anyString(), anyString())).thenAnswer(new Answer<IndexRequestBuilder>() {
        @Override
        public IndexRequestBuilder answer(InvocationOnMock invocationOnMock) throws Throwable {
            Object[] args = invocationOnMock.getArguments();
            String arg1 = (String) args[0];
            if (arg1.isEmpty()) {
                throw new NoNodeAvailableException("Needs index");
            }
            String arg2 = (String) args[1];
            if (arg2.isEmpty()) {
                throw new NoNodeAvailableException("Needs doc type");
            } else {
                IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(mockClient, IndexAction.INSTANCE);
                return indexRequestBuilder;
            }
        }
    });

    return mockClient;
}
 

@Test
public void one_bulk_insert_on_transport_client_when_bulk_action_threshold_is_reached() {

    // Configure the BulkProcessor to use:
    BulkProcessorConfiguration configuration = new BulkProcessorConfiguration(new BulkProcessingOptionsBuilder().build(), bulkProcessorListener);

    // And create a fake index builder:
    IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client, IndexAction.INSTANCE);

    // The mapping to use:
    IElasticSearchMapping localWeatherDataMapper = new LocalWeatherDataMapper();

    // Index to insert to:
    String indexName = "weather_data";

    // Initialize it with the default settings:
    when(client.settings())
            .thenReturn(Settings.builder().build());

    when(client.prepareIndex())
            .thenReturn(indexRequestBuilder);

    // Create the Test subject:
    ElasticSearchClient<LocalWeatherData> elasticSearchClient = new ElasticSearchClient<>(client, indexName, localWeatherDataMapper, configuration);

    // Create more entities, than Bulk insertion threshold:
    Stream<LocalWeatherData> entitiesStream = getData(configuration.getBulkProcessingOptions().getBulkActions() + 1).stream();

    // Index the Data:
    elasticSearchClient.index(entitiesStream);

    // Verify, that the TransportClient bulk insert has been called:
    verify(client, times(1)).bulk(anyObject(), anyObject());
    verify(bulkProcessorListener, times(1)).beforeBulk(anyLong(), anyObject());
}
 

@Test
public void no_value_inserted_when_not_enough_requests() {

    // Create Mocks:
    Client mockedTransportClient = mock(Client.class);
    BulkProcessor.Listener mockedBulkProcessorListener = mock(BulkProcessor.Listener.class);

    // Configure the BulkProcessor to use:
    BulkProcessorConfiguration configuration = new BulkProcessorConfiguration(new BulkProcessingOptionsBuilder().build(), mockedBulkProcessorListener);

    // And create a fake index builder:
    IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(mockedTransportClient, IndexAction.INSTANCE);

    // The mapping to use:
    IElasticSearchMapping localWeatherDataMapper = new LocalWeatherDataMapper();

    // Index to insert to:
    String indexName = "weather_data";

    // Initialize it with the default settings:
    when(mockedTransportClient.settings())
            .thenReturn(Settings.builder().build());

    when(mockedTransportClient.prepareIndex())
            .thenReturn(indexRequestBuilder);

    // Create the Test subject:
    ElasticSearchClient<LocalWeatherData> elasticSearchClient = new ElasticSearchClient<>(mockedTransportClient, indexName, localWeatherDataMapper, configuration);

    // Create more entities, than Bulk insertion threshold:
    Stream<LocalWeatherData> entitiesStream = getData(configuration.getBulkProcessingOptions().getBulkActions() - 1).stream();

    // Index the Data:
    elasticSearchClient.index(entitiesStream);

    // Verify, that the TransportClient bulk insert has been called:
    verify(mockedTransportClient, times(0)).bulk(anyObject(), anyObject());
    verify(mockedBulkProcessorListener, times(0)).beforeBulk(anyLong(), anyObject());
}
 

@Test
public void values_inserted_when_single_entity_is_written() {

    // Configure the BulkProcessor to use:
    BulkProcessorConfiguration configuration = new BulkProcessorConfiguration(new BulkProcessingOptionsBuilder().build(), bulkProcessorListener);

    // And create a fake index builder:
    IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client, IndexAction.INSTANCE);

    // The mapping to use:
    IElasticSearchMapping localWeatherDataMapper = new LocalWeatherDataMapper();

    // Index to insert to:
    String indexName = "weather_data";

    // Initialize it with the default settings:
    when(client.settings())
            .thenReturn(Settings.builder().build());

    when(client.prepareIndex())
            .thenReturn(indexRequestBuilder);

    // Create the Test subject:
    ElasticSearchClient<LocalWeatherData> elasticSearchClient = new ElasticSearchClient<>(client, indexName, localWeatherDataMapper, configuration);

    // Create more entities, than Bulk insertion threshold:
    LocalWeatherData entityToInsert = new LocalWeatherData();

    // Index the Data:
    elasticSearchClient.index(entityToInsert);

    // Flush the Thing:
    elasticSearchClient.flush();

    // Verify, that the TransportClient bulk insert has been called:
    verify(client, times(1)).bulk(anyObject(), anyObject());
    verify(bulkProcessorListener, times(1)).beforeBulk(anyLong(), anyObject());
}
 

@Override
public void closeHistory(String documentId) {
  ElasticsearchDocumentHistory dh = getCachedHistoryIfPresent(documentId);

  if (dh == null) {
    LOGGER.warn(
        "Attempt to close a document {} which is not in cache, thus can't be persisted",
        documentId);
    return;
  }

  try {
    byte[] source = mapper.writeValueAsBytes(new ESHistory(documentId, dh.getAllHistory()));

    new IndexRequestBuilder(elasticsearch.getClient(), IndexAction.INSTANCE)
        .setIndex(documentId)
        .setIndex(esIndex)
        .setType(esType)
        .setId(documentId)
        .setSource(source, XContentType.JSON)
        .get();

  } catch (JsonProcessingException e) {
    LOGGER.warn("Unable to convert history to source, so can't be persisted {}", documentId, e);
  }

  super.closeHistory(documentId);
}
 

public void testLangDetectBinary() throws Exception {
    try {
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client(), CreateIndexAction.INSTANCE).setIndex("test");
        createIndexRequestBuilder.addMapping("someType", jsonBuilder()
                .startObject()
                .startObject("properties")
                .startObject("content")
                .field("type", "binary")
                .startObject("fields")
                .startObject("language")
                .field("type", "langdetect")
                .field("binary", true)
                .endObject()
                .endObject()
                .endObject()
                .endObject()
                .endObject());
        createIndexRequestBuilder.execute().actionGet();
        IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client(), IndexAction.INSTANCE)
                .setIndex("test").setType("someType").setId("1")
                //\"God Save the Queen\" (alternatively \"God Save the King\"
                .setSource("content", "IkdvZCBTYXZlIHRoZSBRdWVlbiIgKGFsdGVybmF0aXZlbHkgIkdvZCBTYXZlIHRoZSBLaW5nIg==");
        indexRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                .execute().actionGet();
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.termQuery("content.language", "en"))
                .addStoredField("content.language");
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        assertEquals(1L, searchResponse.getHits().getTotalHits());
        assertEquals("en", searchResponse.getHits().getAt(0).field("content.language").getValue());
    } finally {
        DeleteIndexRequestBuilder deleteIndexRequestBuilder =
                new DeleteIndexRequestBuilder(client(), DeleteIndexAction.INSTANCE, "test");
        deleteIndexRequestBuilder.execute().actionGet();
    }
}
 

@Override
public void createElasticsearchClient(ProcessContext context) throws ProcessException {
    final Client mockClient = mock(Client.class);
    BulkRequestBuilder bulkRequestBuilder = spy(new BulkRequestBuilder(mockClient, BulkAction.INSTANCE));
    if (exceptionToThrow != null) {
        doThrow(exceptionToThrow).when(bulkRequestBuilder).execute();
    } else {
        doReturn(new MockBulkRequestBuilderExecutor(responseHasFailures)).when(bulkRequestBuilder).execute();
    }
    when(mockClient.prepareBulk()).thenReturn(bulkRequestBuilder);

    when(mockClient.prepareIndex(anyString(), anyString(), anyString())).thenAnswer(new Answer<IndexRequestBuilder>() {
        @Override
        public IndexRequestBuilder answer(InvocationOnMock invocationOnMock) throws Throwable {
            Object[] args = invocationOnMock.getArguments();
            String arg1 = (String) args[0];
            if (arg1.isEmpty()) {
                throw new NoNodeAvailableException("Needs index");
            }
            String arg2 = (String) args[1];
            if (arg2.isEmpty()) {
                throw new NoNodeAvailableException("Needs doc type");
            } else {
                IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(mockClient, IndexAction.INSTANCE);
                return indexRequestBuilder;
            }
        }
    });

    esClient.set(mockClient);
}
 

@Override
protected Client getTransportClient(Settings.Builder settingsBuilder, String xPackPath,
                                    String username, String password,
                                    List<InetSocketAddress> esHosts, ComponentLog log)
        throws MalformedURLException {
    final Client mockClient = mock(Client.class);
    BulkRequestBuilder bulkRequestBuilder = spy(new BulkRequestBuilder(mockClient, BulkAction.INSTANCE));
    if (exceptionToThrow != null) {
        doThrow(exceptionToThrow).when(bulkRequestBuilder).execute();
    } else {
        doReturn(new MockBulkRequestBuilderExecutor(responseHasFailures, esHosts.get(0))).when(bulkRequestBuilder).execute();
    }
    when(mockClient.prepareBulk()).thenReturn(bulkRequestBuilder);

    when(mockClient.prepareIndex(anyString(), anyString(), anyString())).thenAnswer(new Answer<IndexRequestBuilder>() {
        @Override
        public IndexRequestBuilder answer(InvocationOnMock invocationOnMock) throws Throwable {
            Object[] args = invocationOnMock.getArguments();
            String arg1 = (String) args[0];
            if (arg1.isEmpty()) {
                throw new NoNodeAvailableException("Needs index");
            }
            String arg2 = (String) args[1];
            if (arg2.isEmpty()) {
                throw new NoNodeAvailableException("Needs doc type");
            } else {
                IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(mockClient, IndexAction.INSTANCE);
                return indexRequestBuilder;
            }
        }
    });

    return mockClient;
}
 

@Test(expected = ActionRequestValidationException.class)
public void testIngest3() {
    Client client = client("1");
    IndexRequest r = new IndexRequestBuilder(client, IndexAction.INSTANCE).request();
    IngestRequestBuilder builder = new IngestRequestBuilder(client, IngestAction.INSTANCE)
            .add(r);
    client.execute(IngestAction.INSTANCE, builder.request()).actionGet();
}
 

@Test(expected = ClusterBlockException.class)
public void testClusterBlockNodeClient() throws Exception {
    IngestRequestBuilder brb = client("1").prepareExecute(IngestAction.INSTANCE);
    XContentBuilder builder = jsonBuilder().startObject().field("field", "value").endObject();
    String jsonString = builder.string();
    IndexRequest indexRequest = client("1").prepareExecute(IndexAction.INSTANCE)
            .setIndex("test").setType("test").setId("1").setSource(jsonString).request();
    brb.add(indexRequest);
    brb.execute().actionGet();
}
 

@Override
public ActionFuture<IndexResponse> index(final IndexRequest request) {
    return execute(IndexAction.INSTANCE, request);
}
 

@Override
public void index(final IndexRequest request, final ActionListener<IndexResponse> listener) {
    execute(IndexAction.INSTANCE, request, listener);
}
 

@Override
public IndexRequestBuilder prepareIndex() {
    return new IndexRequestBuilder(this, IndexAction.INSTANCE, null);
}
 

public void testChineseLanguageCode() throws Exception {
    try {
        XContentBuilder builder = jsonBuilder()
                .startObject()
                .startObject("properties")
                .startObject("content")
                .field("type", "text")
                .startObject("fields")
                .startObject("language")
                .field("type", "langdetect")
                .array("languages", "zh-cn", "en", "de")
                .endObject()
                .endObject()
                .endObject()
                .endObject()
                .endObject();
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client(), CreateIndexAction.INSTANCE);
        createIndexRequestBuilder.setIndex("test").addMapping("someType", builder).execute().actionGet();
        String source = "位于美国首都华盛顿都会圈的希望中文学校５日晚举办活动庆祝建立２０周年。" +
                "从中国大陆留学生为子女学中文而自发建立的学习班，到学生规模在全美名列前茅的中文学校，" +
                "这个平台的发展也折射出美国的中文教育热度逐步提升。\n" +
                "希望中文学校是大华盛顿地区最大中文学校，现有７个校区逾４０００名学生，" +
                "规模在美国东部数一数二。" +
                "不过，见证了希望中文学校２０年发展的人们起初根本无法想象这个小小的中文教育平台能发展到今日之规模。";
        IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client(), IndexAction.INSTANCE)
                .setIndex("test").setType("someType").setId("1")
                .setSource("content", source);
        indexRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                .execute().actionGet();
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.termQuery("content.language", "zh-cn"))
                .addStoredField("content.language");
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        assertEquals(1L, searchResponse.getHits().getTotalHits());
        assertEquals("zh-cn", searchResponse.getHits().getAt(0).field("content.language").getValue());
    } finally {
        DeleteIndexRequestBuilder deleteIndexRequestBuilder =
                new DeleteIndexRequestBuilder(client(), DeleteIndexAction.INSTANCE, "test");
        deleteIndexRequestBuilder.execute().actionGet();
    }
}
 

public void testGermanLanguageCode() throws Exception {
    try {
        XContentBuilder builder = jsonBuilder()
                .startObject()
                .startObject("properties")
                .startObject("content")
                .field("type", "text")
                .startObject("fields")
                .startObject("language")
                .field("type", "langdetect")
                .array("languages", "zh-cn", "en", "de")
                .endObject()
                .endObject()
                .endObject()
                .endObject()
                .endObject();
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client(), CreateIndexAction.INSTANCE);
        createIndexRequestBuilder.setIndex("test").addMapping("someType", builder).execute().actionGet();
        String source = "Einigkeit und Recht und Freiheit\n" +
                "für das deutsche Vaterland!\n" +
                "Danach lasst uns alle streben\n" +
                "brüderlich mit Herz und Hand!";
        IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client(), IndexAction.INSTANCE)
                .setIndex("test").setType("someType").setId("1")
                .setSource("content", source);
        indexRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                .execute().actionGet();
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.termQuery("content.language", "de"))
                .addStoredField("content.language");
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        assertEquals(1L, searchResponse.getHits().getTotalHits());
        assertEquals("de", searchResponse.getHits().getAt(0).field("content.language").getValue());
    } finally {
        DeleteIndexRequestBuilder deleteIndexRequestBuilder =
                new DeleteIndexRequestBuilder(client(), DeleteIndexAction.INSTANCE, "test");
        deleteIndexRequestBuilder.execute().actionGet();
    }
}