Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
/*
* Copyright 2025 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package com.google.adk.internal.http;

import static java.util.concurrent.TimeUnit.SECONDS;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import okhttp3.Dispatcher;
import okhttp3.OkHttpClient;

/**
* Creates {@link OkHttpClient}s for the ADK. The default clients are cached per name so the
* dispatcher and connection pool are reused across the ADK; a caller that supplies its own executor
* gets a fresh, non-cached client it owns.
*/
public final class HttpClientFactory {

private static final Map<String, OkHttpClient> sharedClients = new ConcurrentHashMap<>();

private HttpClientFactory() {}

/**
* Returns the shared {@link OkHttpClient} cached by {@code name}, using OkHttp's default
* threading.
*/
public static OkHttpClient getOrCreateSharedHttpClient(String name) {
return sharedClients.computeIfAbsent(name, unused -> new OkHttpClient());
}

/**
* Returns a new {@link OkHttpClient} whose dispatcher runs on {@code executorService}. Pass
* {@link #daemonExecutor} so a standalone JVM can exit once work is done, or a container-managed
* executor in a managed environment. The client is not cached: the caller owns the executor and
* the returned client.
*
* @param executorService executor for the dispatcher.
*/
public static OkHttpClient createHttpClient(ExecutorService executorService) {
return new OkHttpClient.Builder().dispatcher(new Dispatcher(executorService)).build();
}

/**
* Returns an unbounded pool of daemon threads, matching OkHttp's own dispatcher pool but with
* daemon threads so a standalone JVM can exit once work is done. Managed container environments
* should inject their own executor instead of calling this.
*
* @param name prefix for the dispatcher thread names.
*/
public static ExecutorService daemonExecutor(String name) {
return new ThreadPoolExecutor(
0, Integer.MAX_VALUE, 60L, SECONDS, new SynchronousQueue<>(), daemonThreadFactory(name));
}

private static ThreadFactory daemonThreadFactory(String name) {
AtomicInteger count = new AtomicInteger();
return runnable -> {
Thread thread = new Thread(runnable, name + "-" + count.incrementAndGet());
thread.setDaemon(true);
return thread;
};
}
}
91 changes: 73 additions & 18 deletions core/src/main/java/com/google/adk/models/Gemini.java
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,14 @@
import static com.google.common.base.StandardSystemProperty.JAVA_VERSION;

import com.google.adk.Version;
import com.google.adk.internal.http.HttpClientFactory;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import com.google.genai.Client;
import com.google.genai.ResponseStream;
import com.google.genai.types.Candidate;
import com.google.genai.types.ClientOptions;
import com.google.genai.types.Content;
import com.google.genai.types.FinishReason;
import com.google.genai.types.FunctionCall;
Expand All @@ -35,13 +37,16 @@
import com.google.genai.types.Part;
import com.google.genai.types.PartialArg;
import io.reactivex.rxjava3.core.Flowable;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import okhttp3.OkHttpClient;
import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -57,6 +62,52 @@ public class Gemini extends BaseLlm {
private static final Logger logger = LoggerFactory.getLogger(Gemini.class);
private static final ImmutableMap<String, String> TRACKING_HEADERS;

private static OkHttpClient prepareHttpClient(@Nullable ExecutorService executorService) {
OkHttpClient client =
executorService == null
? HttpClientFactory.getOrCreateSharedHttpClient("GeminiApiClient")
: HttpClientFactory.createHttpClient(executorService);
return client
.newBuilder()
.connectTimeout(Duration.ZERO)
.readTimeout(Duration.ZERO)
.writeTimeout(Duration.ZERO)
.build();
}

private static Client buildApiKeyClient(
String apiKey, @Nullable ExecutorService executorService) {
return Client.builder()
.apiKey(apiKey)
.httpOptions(HttpOptions.builder().headers(TRACKING_HEADERS).build())
.clientOptions(
ClientOptions.builder().customHttpClient(prepareHttpClient(executorService)).build())
.build();
}

private static Client buildVertexClient(
VertexCredentials vertexCredentials, @Nullable ExecutorService executorService) {
Client.Builder apiClientBuilder =
Client.builder()
.httpOptions(HttpOptions.builder().headers(TRACKING_HEADERS).build())
.clientOptions(
ClientOptions.builder()
.customHttpClient(prepareHttpClient(executorService))
.build());
vertexCredentials.project().ifPresent(apiClientBuilder::project);
vertexCredentials.location().ifPresent(apiClientBuilder::location);
vertexCredentials.credentials().ifPresent(apiClientBuilder::credentials);
return apiClientBuilder.build();
}

private static Client buildDefaultClient(@Nullable ExecutorService executorService) {
return Client.builder()
.httpOptions(HttpOptions.builder().headers(TRACKING_HEADERS).build())
.clientOptions(
ClientOptions.builder().customHttpClient(prepareHttpClient(executorService)).build())
.build();
}

static {
String frameworkLabel = "google-adk/" + Version.JAVA_ADK_VERSION;
String languageLabel = "gl-java/" + JAVA_VERSION.value();
Expand Down Expand Up @@ -90,11 +141,7 @@ public Gemini(String modelName, Client apiClient) {
public Gemini(String modelName, String apiKey) {
super(modelName);
Objects.requireNonNull(apiKey, "apiKey cannot be null");
this.apiClient =
Client.builder()
.apiKey(apiKey)
.httpOptions(HttpOptions.builder().headers(TRACKING_HEADERS).build())
.build();
this.apiClient = buildApiKeyClient(apiKey, null);
}

/**
Expand All @@ -106,12 +153,7 @@ public Gemini(String modelName, String apiKey) {
public Gemini(String modelName, VertexCredentials vertexCredentials) {
super(modelName);
Objects.requireNonNull(vertexCredentials, "vertexCredentials cannot be null");
Client.Builder apiClientBuilder =
Client.builder().httpOptions(HttpOptions.builder().headers(TRACKING_HEADERS).build());
vertexCredentials.project().ifPresent(apiClientBuilder::project);
vertexCredentials.location().ifPresent(apiClientBuilder::location);
vertexCredentials.credentials().ifPresent(apiClientBuilder::credentials);
this.apiClient = apiClientBuilder.build();
this.apiClient = buildVertexClient(vertexCredentials, null);
}

/**
Expand All @@ -131,6 +173,7 @@ public static class Builder {
private Client apiClient;
private String apiKey;
private VertexCredentials vertexCredentials;
private ExecutorService httpExecutorService;

private Builder() {}

Expand Down Expand Up @@ -186,6 +229,22 @@ public Builder vertexCredentials(VertexCredentials vertexCredentials) {
return this;
}

/**
* Sets the executor for the shared HTTP client's dispatcher. Pass {@link
* HttpClientFactory#daemonExecutor} so a standalone or CLI JVM can exit once work is done, or a
* container-managed executor in a managed environment. Applies only when the client is built
* from an API key, Vertex credentials, or the default; it is ignored when an explicit {@link
* #apiClient} is supplied.
*
* @param httpExecutorService The executor for HTTP dispatcher threads.
* @return This builder.
*/
@CanIgnoreReturnValue
public Builder httpExecutorService(ExecutorService httpExecutorService) {
this.httpExecutorService = httpExecutorService;
return this;
}

/**
* Builds the {@link Gemini} instance.
*
Expand All @@ -198,15 +257,11 @@ public Gemini build() {
if (apiClient != null) {
return new Gemini(modelName, apiClient);
} else if (apiKey != null) {
return new Gemini(modelName, apiKey);
return new Gemini(modelName, buildApiKeyClient(apiKey, httpExecutorService));
} else if (vertexCredentials != null) {
return new Gemini(modelName, vertexCredentials);
return new Gemini(modelName, buildVertexClient(vertexCredentials, httpExecutorService));
} else {
return new Gemini(
modelName,
Client.builder()
.httpOptions(HttpOptions.builder().headers(TRACKING_HEADERS).build())
.build());
return new Gemini(modelName, buildDefaultClient(httpExecutorService));
}
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.adk.JsonBaseModel;
import com.google.adk.internal.http.HttpClientFactory;
import com.google.adk.models.LlmRequest;
import com.google.adk.models.LlmResponse;
import com.google.common.annotations.VisibleForTesting;
Expand All @@ -32,6 +33,7 @@
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.HttpUrl;
Expand All @@ -42,6 +44,7 @@
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.BufferedSource;
import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -68,11 +71,16 @@ public final class ChatCompletionsHttpClient implements ChatCompletionsClient {
private static final Duration DEFAULT_CALL_TIMEOUT = Duration.ofMinutes(5);

/**
* Shared OkHttpClient instance whose connection pool and thread dispatcher are reused across all
* {@link ChatCompletionsHttpClient} instances. Each instance forks this client via {@link
* Returns the OkHttpClient whose connection pool and thread dispatcher back {@link
* ChatCompletionsHttpClient} instances. Without an executor this is the shared client cached by
* name; with one it is a fresh client the caller owns. Each instance forks it via {@link
* OkHttpClient#newBuilder()} to apply per-instance timeouts without leaking pools.
*/
private static final OkHttpClient SHARED_POOL_CLIENT = new OkHttpClient();
private static OkHttpClient prepareHttpClient(@Nullable ExecutorService executorService) {
return executorService == null
? HttpClientFactory.getOrCreateSharedHttpClient("ChatCompletionsHttpClient")
: HttpClientFactory.createHttpClient(executorService);
}

private final OkHttpClient client;
private final HttpUrl completionsUrl;
Expand Down Expand Up @@ -116,7 +124,19 @@ public final class ChatCompletionsHttpClient implements ChatCompletionsClient {
* HTTP(S) URL.
*/
public ChatCompletionsHttpClient(HttpOptions httpOptions) {
this(httpOptions, buildClient(httpOptions));
this(httpOptions, buildClient(httpOptions, null));
}

/**
* Constructs a {@link ChatCompletionsHttpClient} whose HTTP dispatcher runs on {@code
* httpExecutorService}. Pass {@link HttpClientFactory#daemonExecutor} so a standalone or CLI JVM
* can exit once work is done, or a container-managed executor in a managed environment.
*
* @param httpOptions HTTP configuration; see {@link #ChatCompletionsHttpClient(HttpOptions)}.
* @param httpExecutorService executor for the HTTP dispatcher threads.
*/
public ChatCompletionsHttpClient(HttpOptions httpOptions, ExecutorService httpExecutorService) {
this(httpOptions, buildClient(httpOptions, httpExecutorService));
}

private ChatCompletionsHttpClient(HttpOptions httpOptions, OkHttpClient client) {
Expand Down Expand Up @@ -153,12 +173,13 @@ static ChatCompletionsHttpClient forTesting(HttpOptions httpOptions, OkHttpClien
}

/**
* Builds the production OkHttpClient by forking {@link #SHARED_POOL_CLIENT} so the connection
* pool and dispatcher are reused across instances while applying per-instance timeouts.
* Builds the production OkHttpClient by forking the shared pool client so the connection pool and
* dispatcher are reused across instances while applying per-instance timeouts.
*/
private static OkHttpClient buildClient(HttpOptions httpOptions) {
private static OkHttpClient buildClient(
HttpOptions httpOptions, @Nullable ExecutorService executorService) {
Objects.requireNonNull(httpOptions, "httpOptions cannot be null");
OkHttpClient.Builder builder = SHARED_POOL_CLIENT.newBuilder();
OkHttpClient.Builder builder = prepareHttpClient(executorService).newBuilder();
builder.connectTimeout(Duration.ZERO);
builder.readTimeout(Duration.ZERO);
builder.writeTimeout(Duration.ZERO);
Expand Down
7 changes: 6 additions & 1 deletion core/src/main/java/com/google/adk/sessions/ApiClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import static com.google.common.base.StandardSystemProperty.JAVA_VERSION;

import com.google.adk.internal.http.HttpClientFactory;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.common.base.Ascii;
import com.google.common.base.Strings;
Expand Down Expand Up @@ -102,8 +103,12 @@ abstract class ApiClient {
this.httpClient = createHttpClient(httpOptions.timeout().orElse(null));
}

private static OkHttpClient getSharedPoolClient() {
return HttpClientFactory.getOrCreateSharedHttpClient("ApiClient");
}

private OkHttpClient createHttpClient(@Nullable Integer timeout) {
OkHttpClient.Builder builder = new OkHttpClient().newBuilder();
OkHttpClient.Builder builder = getSharedPoolClient().newBuilder();
if (timeout != null) {
builder.connectTimeout(Duration.ofMillis(timeout));
}
Expand Down
Loading
Loading