chore: use async HTTP call

bednar · bednar · commit b6bf1ee84130 · 2021-11-15T11:21:38.000+01:00
diff --git a/client/src/main/java/com/influxdb/client/internal/AbstractWriteClient.java b/client/src/main/java/com/influxdb/client/internal/AbstractWriteClient.java
@@ -57,6 +57,7 @@
 import io.reactivex.subjects.PublishSubject;
 import org.reactivestreams.Publisher;
 import retrofit2.Call;
+import retrofit2.Callback;
 import retrofit2.HttpException;
 import retrofit2.Response;
 
@@ -450,12 +451,21 @@ public Maybe<Notification<Response>> apply(final BatchWriteItem batchWrite) {
             String bucket = batchWrite.batchWriteOptions.bucket;
             WritePrecision precision = batchWrite.batchWriteOptions.precision;
 
-            Maybe<Response<Void>> requestSource = Maybe
-                    .fromCallable(() -> service
-                            .postWrite(organization, bucket, content, null,
-                                    "identity", "text/plain; charset=utf-8", null,
-                                    "application/json", null, precision))
-                    .map(Call::execute);
+            Maybe<Response<Void>> requestSource = Maybe.create(emitter -> service
+                    .postWrite(organization, bucket, content, null,
+                            "identity", "text/plain; charset=utf-8", null,
+                            "application/json", null, precision)
+                    .enqueue(new Callback<Void>() {
+                        @Override
+                        public void onResponse(@Nonnull final Call<Void> call, @Nonnull final Response<Void> response) {
+                            emitter.onSuccess(response);
+                        }
+
+                        @Override
+                        public void onFailure(@Nonnull final Call<Void> call, @Nonnull final Throwable throwable) {
+                            emitter.onError(throwable);
+                        }
+                    }));
 
             return requestSource
                     //
diff --git a/client/src/test/java/com/influxdb/client/ITWriteQueryApi.java b/client/src/test/java/com/influxdb/client/ITWriteQueryApi.java
@@ -461,12 +461,15 @@ void queryDataFromNewOrganization() {
     }
 
     @Test
-    void flush() throws InterruptedException {
+    void flush() {
 
         String bucketName = bucket.getName();
 
-        writeApi = influxDBClient.makeWriteApi(WriteOptions.builder().batchSize(100_000)
-                .flushInterval(100_000).build());
+        writeApi = influxDBClient.makeWriteApi(WriteOptions.builder().batchSize(1_000_000)
+                .flushInterval(1_000_000).build());
+
+        WriteEventListener<WriteSuccessEvent> successListener = new WriteEventListener<>();
+        writeApi.listenEvents(WriteSuccessEvent.class, successListener);
 
         String record = "h2o_feet,location=coyote_creek level\\ water_level=1.0 1";
 
@@ -476,7 +479,7 @@ void flush() throws InterruptedException {
         Assertions.assertThat(query).hasSize(0);
 
         writeApi.flush();
-        Thread.sleep(10);
+        waitToCallback(successListener.countDownLatch, 10);
 
         query = queryApi.query("from(bucket:\"" + bucketName + "\") |> range(start: 1970-01-01T00:00:00.000000001Z)", organization.getId());
 
