refactor: avoid using Flowable.fromFuture() which uses blocking get under the hood

PiperOrigin-RevId: 856180759

Author: Mateusz Krawiec

core/pom.xml

@@ -55,7 +55,7 @@
       <artifactId>docker-java-transport-httpclient5</artifactId>
     </dependency>
     <dependency>
-      <groupId> io.modelcontextprotocol.sdk</groupId>
+      <groupId>io.modelcontextprotocol.sdk</groupId>
       <artifactId>mcp</artifactId>
     </dependency>
     <dependency>
@@ -189,6 +189,10 @@
       <artifactId>opentelemetry-sdk-testing</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>net.javacrumbs.future-converter</groupId>
+      <artifactId>future-converter-java8-guava</artifactId>
+    </dependency>
   </dependencies>
   <build>
     <resources>

core/src/main/java/com/google/adk/models/Gemini.java

@@ -17,9 +17,12 @@
 package com.google.adk.models;
 
 import static com.google.common.base.StandardSystemProperty.JAVA_VERSION;
+import static net.javacrumbs.futureconverter.java8guava.FutureConverter.toListenableFuture;
 
 import com.google.adk.Version;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.genai.Client;
 import com.google.genai.ResponseStream;
@@ -32,11 +35,14 @@
 import com.google.genai.types.LiveConnectConfig;
 import com.google.genai.types.Part;
 import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -205,6 +211,23 @@ public Gemini build() {
     }
   }
 
+  private static <T> Single<T> toSingle(ListenableFuture<T> future, Scheduler scheduler) {
+    return Single.create(
+        emitter -> {
+          future.addListener(
+              () -> {
+                try {
+                  emitter.onSuccess(Futures.getDone(future));
+                } catch (ExecutionException e) {
+                  emitter.onError(e.getCause());
+                }
+              },
+              scheduler::scheduleDirect);
+
+          emitter.setCancellable(() -> future.cancel(false));
+        });
+  }
+
   @Override
   public Flowable<LlmResponse> generateContent(LlmRequest llmRequest, boolean stream) {
     llmRequest =
@@ -218,14 +241,17 @@ public Flowable<LlmResponse> generateContent(LlmRequest llmRequest, boolean stre
 
     if (stream) {
       logger.debug("Sending streaming generateContent request to model {}", effectiveModelName);
-      CompletableFuture<ResponseStream<GenerateContentResponse>> streamFuture =
-          apiClient.async.models.generateContentStream(
-              effectiveModelName, llmRequest.contents(), config);
+      ListenableFuture<ResponseStream<GenerateContentResponse>> streamFuture =
+          toListenableFuture(
+              apiClient.async.models.generateContentStream(
+                  effectiveModelName, llmRequest.contents(), config));
 
       return Flowable.defer(
               () ->
                   processRawResponses(
-                      Flowable.fromFuture(streamFuture).flatMapIterable(iterable -> iterable)))
+                      toSingle(streamFuture, Schedulers.io())
+                          .toFlowable()
+                          .flatMapIterable(iterable -> iterable)))
           .filter(
               llmResponse ->
                   llmResponse
@@ -243,12 +269,16 @@ public Flowable<LlmResponse> generateContent(LlmRequest llmRequest, boolean stre
                       .orElse(false));
     } else {
       logger.debug("Sending generateContent request to model {}", effectiveModelName);
-      return Flowable.fromFuture(
-          apiClient
-              .async
-              .models
-              .generateContent(effectiveModelName, llmRequest.contents(), config)
-              .thenApplyAsync(LlmResponse::create));
+      final LlmRequest finalLlmRequest = llmRequest;
+      return toSingle(
+              toListenableFuture(
+                  apiClient
+                      .async
+                      .models
+                      .generateContent(effectiveModelName, finalLlmRequest.contents(), config)
+                      .thenApplyAsync(LlmResponse::create)),
+              Schedulers.io())
+          .toFlowable();
     }
   }
 

pom.xml

@@ -72,6 +72,7 @@
     <anthropic.version>1.4.0</anthropic.version>
     <maven-plugin-tools.version>3.9.0</maven-plugin-tools.version>
     <httpclient5.version>5.4.3</httpclient5.version>
+    <future-converter-java8-guava.version>1.2.0</future-converter-java8-guava.version>
   </properties>
 
   <dependencyManagement>
@@ -274,6 +275,11 @@
         <artifactId>assertj-core</artifactId>
         <version>${assertj.version}</version>
       </dependency>
+      <dependency>
+        <groupId>net.javacrumbs.future-converter</groupId>
+        <artifactId>future-converter-java8-guava</artifactId>
+        <version>${future-converter-java8-guava.version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>