RemoteElasticsearchClient implements ElasticsearchClient { @Override public Single<JsonObject> bulkInsert(String type, List<Tuple2<String, JsonObject>> documents) { String uri = "/" + index + "/" + type + "/_bulk"; Buffer body = Buffer.buffer(); for (Tuple2<String, JsonObject> e : documents) { String id = e.v1; Buffer source = e.v2.toBuffer(); JsonObject subject = new JsonObject().put("_id", id); body.appendString("{\"index\":") .appendBuffer(subject.toBuffer()) .appendString("}\n") .appendBuffer(source) .appendString("\n"); } return client.performRequest(HttpMethod.POST, uri, body); } RemoteElasticsearchClient(List<URI> hosts, String index,
      Duration autoUpdateHostsInterval, boolean compressRequestBodies, Vertx vertx); @Override void close(); @Override Single<JsonObject> bulkInsert(String type,
      List<Tuple2<String, JsonObject>> documents); @Override Single<JsonObject> beginScroll(String type, JsonObject query,
      JsonObject postFilter, JsonObject aggregations, JsonObject parameters, String timeout); @Override Single<JsonObject> continueScroll(String scrollId, String timeout); @Override Single<JsonObject> search(String type, JsonObject query,
      JsonObject postFilter, JsonObject aggregations, JsonObject parameters); @Override Single<Long> count(String type, JsonObject query); @Override Single<JsonObject> updateByQuery(String type, JsonObject postFilter,
      JsonObject script); @Override Single<JsonObject> bulkDelete(String type, JsonArray ids); @Override Single<Boolean> indexExists(); @Override Single<Boolean> typeExists(String type); @Override Single<Boolean> createIndex(); @Override Single<Boolean> createIndex(JsonObject settings); @Override Completable ensureIndex(); @Override Single<Boolean> putMapping(String type, JsonObject mapping); @Override Completable ensureMapping(String type, JsonObject mapping); @Override Single<JsonObject> getMapping(String type); @Override Single<JsonObject> getMapping(String type, String field); @Override Single<Boolean> isRunning();  }
@Test public void bulkInsert(TestContext context) { String url = "/" + INDEX + "/" + TYPE + "/_bulk"; wireMockRule1.stubFor(post(urlEqualTo(url)) .willReturn(aResponse() .withStatus(200) .withBody("{}"))); List<Tuple2<String, JsonObject>> documents = new ArrayList<>(); documents.add(Tuple.tuple("A", new JsonObject().put("name", "Elvis"))); documents.add(Tuple.tuple("B", new JsonObject().put("name", "Max"))); Async async = context.async(); client.bulkInsert(TYPE, documents).subscribe(res -> { wireMockRule1.verify(postRequestedFor(urlEqualTo(url)) .withRequestBody(equalToJson("{\"index\":{\"_id\":\"A\"}}\n" + "{\"name\":\"Elvis\"}\n" + "{\"index\":{\"_id\":\"B\"}}\n" + "{\"name\":\"Max\"}\n"))); context.assertEquals(0, res.size()); async.complete(); }, context::fail); }