ElasticSearchClient implements IElasticSearchClient<TEntity> { public void index(TEntity entity) { index(Arrays.asList(entity)); } ElasticSearchClient(final Client client, final String indexName, final IElasticSearchMapping mapping, final BulkProcessorConfiguration bulkProcessorConfiguration); void index(TEntity entity); void index(List<TEntity> entities); void index(Stream<TEntity> entities); void flush(); synchronized boolean awaitClose(long timeout, TimeUnit unit); @Override void close();  }
@Test public void one_bulk_insert_on_transport_client_when_bulk_action_threshold_is_reached() { BulkProcessorConfiguration configuration = new BulkProcessorConfiguration(new BulkProcessingOptionsBuilder().build(), bulkProcessorListener); IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client, IndexAction.INSTANCE); IElasticSearchMapping localWeatherDataMapper = new LocalWeatherDataMapper(); String indexName = "weather_data"; when(client.settings()) .thenReturn(Settings.builder().build()); when(client.prepareIndex()) .thenReturn(indexRequestBuilder); ElasticSearchClient<LocalWeatherData> elasticSearchClient = new ElasticSearchClient<>(client, indexName, localWeatherDataMapper, configuration); Stream<LocalWeatherData> entitiesStream = getData(configuration.getBulkProcessingOptions().getBulkActions() + 1).stream(); elasticSearchClient.index(entitiesStream); verify(client, times(1)).bulk(anyObject(), anyObject()); verify(bulkProcessorListener, times(1)).beforeBulk(anyLong(), anyObject()); }
@Test public void no_value_inserted_when_not_enough_requests() { Client mockedTransportClient = mock(Client.class); BulkProcessor.Listener mockedBulkProcessorListener = mock(BulkProcessor.Listener.class); BulkProcessorConfiguration configuration = new BulkProcessorConfiguration(new BulkProcessingOptionsBuilder().build(), mockedBulkProcessorListener); IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(mockedTransportClient, IndexAction.INSTANCE); IElasticSearchMapping localWeatherDataMapper = new LocalWeatherDataMapper(); String indexName = "weather_data"; when(mockedTransportClient.settings()) .thenReturn(Settings.builder().build()); when(mockedTransportClient.prepareIndex()) .thenReturn(indexRequestBuilder); ElasticSearchClient<LocalWeatherData> elasticSearchClient = new ElasticSearchClient<>(mockedTransportClient, indexName, localWeatherDataMapper, configuration); Stream<LocalWeatherData> entitiesStream = getData(configuration.getBulkProcessingOptions().getBulkActions() - 1).stream(); elasticSearchClient.index(entitiesStream); verify(mockedTransportClient, times(0)).bulk(anyObject(), anyObject()); verify(mockedBulkProcessorListener, times(0)).beforeBulk(anyLong(), anyObject()); }