ElasticClients { @Nonnull public static RestClientBuilder client() { return client("localhost", DEFAULT_PORT); } private  ElasticClients(); @Nonnull static RestClientBuilder client(); @Nonnull static RestClientBuilder client(@Nonnull String location); @Nonnull static RestClientBuilder client(@Nonnull String hostname, int port); @Nonnull static RestClientBuilder client(
            @Nonnull String username,
            @Nonnull String password,
            @Nonnull String hostname,
            int port
    );  }
@Test public void given_clientAsString_whenReadFromElasticSource_thenFinishSuccessfully() { ElasticsearchContainer container = ElasticSupport.elastic.get(); String httpHostAddress = container.getHttpHostAddress(); indexDocument("my-index", ImmutableMap.of("name", "Frantisek")); Pipeline p = Pipeline.create(); p.readFrom(ElasticSources.elastic( () -> ElasticClients.client(httpHostAddress), () -> new SearchRequest("my-index"), SearchHit::getSourceAsString) ).writeTo(Sinks.list(results)); submitJob(p); assertThat(results).hasSize(1); }
@Test public void given_clientAsString_whenReadFromElasticSource_thenFinishSuccessfully() { ElasticsearchContainer container = ElasticSupport.elastic.get(); String httpHostAddress = container.getHttpHostAddress(); indexDocument("my-index", ImmutableMap.of("name", "Frantisek")); Pipeline p = Pipeline.create(); p.readFrom(ElasticSources.elastic( () -> ElasticClients.client(httpHostAddress), SearchHit::getSourceAsString) ).writeTo(Sinks.list(results)); submitJob(p); assertThat(results).hasSize(1); }