fix: use native support for Rx request instead blocking emitter

Author: bednar

client-reactive/pom.xml

@@ -98,25 +98,6 @@
             <artifactId>influxdb-client-java</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.squareup.retrofit2</groupId>
-            <artifactId>adapter-rxjava3</artifactId>
-            <exclusions>
-                <exclusion>
-                    <groupId>com.squareup.okhttp3</groupId>
-                    <artifactId>okhttp</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.squareup.retrofit2</groupId>
-                    <artifactId>retrofit</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>io.reactivex.rxjava2</groupId>
-                    <artifactId>rxjava</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-
         <dependency>
             <groupId>com.influxdb</groupId>
             <artifactId>influxdb-client-test</artifactId>

client/pom.xml

@@ -147,6 +147,25 @@
             <artifactId>converter-gson</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.squareup.retrofit2</groupId>
+            <artifactId>adapter-rxjava3</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.squareup.okhttp3</groupId>
+                    <artifactId>okhttp</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.squareup.retrofit2</groupId>
+                    <artifactId>retrofit</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>io.reactivex.rxjava2</groupId>
+                    <artifactId>rxjava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <dependency>
             <groupId>com.moandjiezana.toml</groupId>
             <artifactId>toml4j</artifactId>

client/src/main/java/com/influxdb/client/internal/AbstractInfluxDBClient.java

@@ -46,6 +46,7 @@
 import retrofit2.Call;
 import retrofit2.CallAdapter;
 import retrofit2.Retrofit;
+import retrofit2.adapter.rxjava3.RxJava3CallAdapterFactory;
 import retrofit2.converter.gson.GsonConverterFactory;
 import retrofit2.converter.scalars.ScalarsConverterFactory;
 
@@ -75,7 +76,7 @@ public abstract class AbstractInfluxDBClient extends AbstractRestClient {
     protected final Collection<AutoCloseable> autoCloseables = new CopyOnWriteArrayList<>();
 
     public AbstractInfluxDBClient(@Nonnull final InfluxDBClientOptions options, @Nonnull final String clientType) {
-        this(options, clientType, Collections.emptyList());
+        this(options, clientType, Collections.singletonList(RxJava3CallAdapterFactory.createSynchronous()));
     }
 
     public AbstractInfluxDBClient(@Nonnull final InfluxDBClientOptions options,

client/src/main/java/com/influxdb/client/internal/AbstractWriteClient.java

@@ -58,8 +58,6 @@
 import io.reactivex.rxjava3.processors.PublishProcessor;
 import io.reactivex.rxjava3.subjects.PublishSubject;
 import org.reactivestreams.Publisher;
-import retrofit2.Call;
-import retrofit2.Callback;
 import retrofit2.HttpException;
 import retrofit2.Response;
 
@@ -434,21 +432,11 @@ public Maybe<Notification<Response>> apply(final BatchWriteItem batchWrite) {
             String bucket = batchWrite.batchWriteOptions.bucket;
             WritePrecision precision = batchWrite.batchWriteOptions.precision;
 
-            Maybe<Response<Void>> requestSource = Maybe.create(emitter -> service
-                    .postWrite(organization, bucket, content, null,
+            Maybe<Response<Void>> requestSource = service
+                    .postWriteRx(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);
-                        }
-                    }));
+                    .toMaybe();
 
             return requestSource
                     //

pom.xml

@@ -112,8 +112,7 @@
         <dependency.okhttp3.version>4.9.3</dependency.okhttp3.version>
         <dependency.com.squareup.okio>2.10.0</dependency.com.squareup.okio>
         <dependency.gson.version>2.8.9</dependency.gson.version>
-<!--        <dependency.io.reactivex.rxjava2>2.2.21</dependency.io.reactivex.rxjava2>-->
-        <dependency.io.reactivex.rxjava3>3.0.0</dependency.io.reactivex.rxjava3>
+        <dependency.io.reactivex.rxjava3>3.1.3</dependency.io.reactivex.rxjava3>
 
         <plugin.surefire.version>2.22.2</plugin.surefire.version>
         <plugin.javadoc.version>3.2.0</plugin.javadoc.version>