Skip to content

Commit ad29280

Browse files
loong10kloong10k
committed
perf: bound streams and propagate HTTP cancellation
1 parent 2f11372 commit ad29280

6 files changed

Lines changed: 265 additions & 31 deletions

File tree

‎src/main/java/io/github/easy4j/opencode/HttpCallCancellation.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
package io.github.easy4j.opencode;
2+
3+
/** 将业务层取消信号绑定到一次 OpenCode HTTP 调用。 */
4+
@FunctionalInterface
5+
public interface HttpCallCancellation {
6+
7+
AutoCloseable onCancel(Runnable callback);
8+
9+
default boolean isCancelled() {
10+
return false;
11+
}
12+
}

‎src/main/java/io/github/easy4j/opencode/OpenCodeClient.java‎

Lines changed: 68 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,12 @@
1515
import java.util.List;
1616
import java.util.Map;
1717
import java.util.Objects;
18+
import java.util.concurrent.ExecutorService;
19+
import java.util.concurrent.LinkedBlockingQueue;
20+
import java.util.concurrent.RejectedExecutionException;
21+
import java.util.concurrent.ThreadPoolExecutor;
22+
import java.util.concurrent.TimeUnit;
23+
import java.util.concurrent.atomic.AtomicInteger;
1824

1925
/**
2026
* OpenCode 客户端门面：HTTP Server + SSE 事件流 + 本地 CLI。
@@ -48,14 +54,15 @@ public class OpenCodeClient implements AutoCloseable {
4854
private final OpenCodeHttpClient httpClient;
4955
private final OpenCodeSseClient sseClient;
5056
private final OpenCodeCli cli;
57+
private final ExecutorService streamExecutor;
5158

5259
// ============================================================
5360
// 构造器
5461
// ============================================================
5562

5663
/** 仅 HTTP 子系统（CLI 禁用）。自动创建默认 ObjectMapper 与 OkHttpClient。 */
5764
public OpenCodeClient(OpenCodeHttpClientConfig httpConfig) {
58-
this(httpConfig, new OpenCodeCliConfig(), new ObjectMapper(), new OkHttpClient());
65+
this(httpConfig, new OpenCodeCliConfig(), new ObjectMapper(), null);
5966
}
6067

6168
/** 仅 HTTP 子系统（CLI 禁用），强制注入共享 ObjectMapper 与 OkHttpClient。 */
@@ -65,7 +72,7 @@ public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, ObjectMapper objectMa
6572

6673
/** 仅 CLI 子系统（HTTP 禁用）。自动创建默认 ObjectMapper 与 OkHttpClient。 */
6774
public OpenCodeClient(OpenCodeCliConfig cliConfig) {
68-
this(new OpenCodeHttpClientConfig(), cliConfig, new ObjectMapper(), new OkHttpClient());
75+
this(new OpenCodeHttpClientConfig(), cliConfig, new ObjectMapper(), null);
6976
}
7077

7178
/** 仅 CLI 子系统（HTTP 禁用），强制注入共享 ObjectMapper 与 OkHttpClient。 */
@@ -75,7 +82,7 @@ public OpenCodeClient(OpenCodeCliConfig cliConfig, ObjectMapper objectMapper, Ok
7582

7683
/** HTTP + CLI 子系统。自动创建默认 ObjectMapper 与 OkHttpClient。 */
7784
public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, OpenCodeCliConfig cliConfig) {
78-
this(httpConfig, cliConfig, new ObjectMapper(), new OkHttpClient());
85+
this(httpConfig, cliConfig, new ObjectMapper(), null);
7986
}
8087

8188
/**
@@ -91,7 +98,6 @@ public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, OpenCodeCliConfig cli
9198
Objects.requireNonNull(httpConfig, "httpConfig");
9299
Objects.requireNonNull(cliConfig, "cliConfig");
93100
Objects.requireNonNull(objectMapper, "objectMapper");
94-
Objects.requireNonNull(httpClient, "httpClient");
95101

96102
boolean httpEnabled = httpConfig.isEnabled();
97103
boolean cliEnabled = cliConfig.isEnabled();
@@ -105,11 +111,12 @@ public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, OpenCodeCliConfig cli
105111
if (httpEnabled) {
106112
this.httpClient = new OpenCodeHttpClient(httpConfig, objectMapper, httpClient);
107113
this.sseClient = new OpenCodeSseClient(httpConfig, objectMapper,
108-
httpClient != null ? httpClient : this.httpClient.getOkHttpClient());
114+
this.httpClient.getOkHttpClient());
109115
} else {
110116
this.httpClient = null;
111117
this.sseClient = null;
112118
}
119+
this.streamExecutor = createStreamExecutor(httpConfig);
113120

114121
// CLI 子系统初始化
115122
if (cliEnabled) {
@@ -124,7 +131,7 @@ public OpenCodeClient(OpenCodeHttpClientConfig httpConfig, OpenCodeCliConfig cli
124131

125132
/** 组合配置，自动创建默认 ObjectMapper 与 OkHttpClient。 */
126133
public OpenCodeClient(OpenCodeClientConfig config) {
127-
this(config, new ObjectMapper(), new OkHttpClient());
134+
this(config, new ObjectMapper(), null);
128135
}
129136

130137
/** 组合配置，强制注入共享 ObjectMapper 与 OkHttpClient。 */
@@ -147,6 +154,21 @@ public OpenCodeClient(OpenCodeClientConfig config,
147154
this.httpClient = httpClient;
148155
this.sseClient = sseClient;
149156
this.cli = cli;
157+
this.streamExecutor = createStreamExecutor(config.getHttp());
158+
}
159+
160+
private static ExecutorService createStreamExecutor(OpenCodeHttpClientConfig config) {
161+
int corePoolSize = Math.max(1, config.getStreamCorePoolSize());
162+
int maxPoolSize = Math.max(corePoolSize, config.getStreamMaxPoolSize());
163+
AtomicInteger threadIndex = new AtomicInteger();
164+
return new ThreadPoolExecutor(corePoolSize, maxPoolSize,
165+
Math.max(1L, config.getStreamKeepAliveMillis()), TimeUnit.MILLISECONDS,
166+
new LinkedBlockingQueue<>(Math.max(1, config.getStreamQueueCapacity())), runnable -> {
167+
Thread thread = new Thread(runnable,
168+
"opencode-stream-consumer-" + threadIndex.incrementAndGet());
169+
thread.setDaemon(true);
170+
return thread;
171+
}, new ThreadPoolExecutor.AbortPolicy());
150172
}
151173

152174
// ============================================================
@@ -218,6 +240,11 @@ private void copyHttpConfig(OpenCodeHttpClientConfig src) {
218240
this.config.getHttp().setKeepAliveDurationMillis(src.getKeepAliveDurationMillis());
219241
this.config.getHttp().setMaxRequests(src.getMaxRequests());
220242
this.config.getHttp().setMaxRequestsPerHost(src.getMaxRequestsPerHost());
243+
this.config.getHttp().setStreamCorePoolSize(src.getStreamCorePoolSize());
244+
this.config.getHttp().setStreamMaxPoolSize(src.getStreamMaxPoolSize());
245+
this.config.getHttp().setStreamQueueCapacity(src.getStreamQueueCapacity());
246+
this.config.getHttp().setStreamKeepAliveMillis(src.getStreamKeepAliveMillis());
247+
this.config.getHttp().setSseEventQueueCapacity(src.getSseEventQueueCapacity());
221248
this.config.getHttp().setRetryOnConnectionFailure(src.isRetryOnConnectionFailure());
222249
this.config.getHttp().setVerifySsl(src.isVerifySsl());
223250
this.config.getHttp().setDefaultModel(src.getDefaultModel());
@@ -287,6 +314,11 @@ public PromptResult chatCompletionWithSession(PromptRequest request, String sess
287314
return httpClient.chatCompletionWithSession(request, sessionKey);
288315
}
289316

317+
public PromptResult chatCompletionWithSession(PromptRequest request, String sessionKey,
318+
HttpCallCancellation cancellation) {
319+
return httpClient.chatCompletionWithSession(request, sessionKey, cancellation);
320+
}
321+
290322
public PromptResult chatCompletionWithSession(String text, String sessionKey) {
291323
return httpClient.chatCompletionWithSession(PromptRequest.ofText(text), sessionKey);
292324
}
@@ -315,6 +347,13 @@ public ChatResponse chatCompletionWithSession(ChatRequest request, String sessio
315347
return ChatMessageMapper.toChatResponse(result);
316348
}
317349

350+
public ChatResponse chatCompletionWithSession(ChatRequest request, String sessionKey,
351+
HttpCallCancellation cancellation) {
352+
PromptRequest promptRequest = ChatMessageMapper.toPromptRequest(request);
353+
PromptResult result = httpClient.chatCompletionWithSession(promptRequest, sessionKey, cancellation);
354+
return ChatMessageMapper.toChatResponse(result);
355+
}
356+
318357
public ChatStreamingResponse chatCompletionStream(ChatRequest request, String sessionKey) {
319358
return chatCompletionStream(request, sessionKey, null);
320359
}
@@ -326,9 +365,12 @@ public ChatStreamingResponse chatCompletionStream(ChatRequest request, String se
326365

327366
ChatStreamingResponse stream = new ChatStreamingResponse();
328367

329-
java.util.concurrent.BlockingQueue<Event> queue = sseClient.subscribeQueue(context);
368+
OpenCodeSseClient.QueueSubscription subscription =
369+
sseClient.subscribeQueueSubscription(context);
370+
java.util.concurrent.BlockingQueue<Event> queue = subscription.getQueue();
330371

331-
java.util.concurrent.CompletableFuture.runAsync(() -> {
372+
try {
373+
streamExecutor.submit(() -> {
332374
try {
333375
long deadline = System.currentTimeMillis() + (config.getCli().getTimeout() * 1000L);
334376
while (!stream.isDone() && System.currentTimeMillis() < deadline) {
@@ -379,10 +421,25 @@ public ChatStreamingResponse chatCompletionStream(ChatRequest request, String se
379421
stream.fail(e);
380422
} catch (Exception e) {
381423
stream.fail(e);
424+
} finally {
425+
subscription.close();
382426
}
383-
});
427+
});
428+
} catch (RejectedExecutionException error) {
429+
subscription.close();
430+
stream.fail(new IllegalStateException("OpenCode stream executor is full", error));
431+
return stream;
432+
}
384433

385-
httpClient.promptAsync(sessionId, promptRequest, context);
434+
try {
435+
if (!httpClient.promptAsync(sessionId, promptRequest, context)) {
436+
subscription.close();
437+
stream.fail(new IllegalStateException("OpenCode async prompt was rejected"));
438+
}
439+
} catch (RuntimeException error) {
440+
subscription.close();
441+
stream.fail(error);
442+
}
386443

387444
return stream;
388445
}
@@ -895,6 +952,7 @@ public io.github.easy4j.opencode.cli.OpenCodeCliResult cliPr(int number) {
895952

896953
@Override
897954
public void close() {
955+
streamExecutor.shutdownNow();
898956
if (httpClient != null) httpClient.close();
899957
if (sseClient != null) sseClient.close();
900958
}

‎src/main/java/io/github/easy4j/opencode/OpenCodeHttpClientConfig.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,21 @@ public class OpenCodeHttpClientConfig {
8585
*/
8686
private int maxRequestsPerHost = 128;
8787

88+
/** 流式事件处理线程数。 */
89+
private int streamCorePoolSize = 32;
90+
91+
/** 流式事件处理最大线程数。 */
92+
private int streamMaxPoolSize = 32;
93+
94+
/** 流式事件处理有界队列容量。 */
95+
private int streamQueueCapacity = 128;
96+
97+
/** 流式事件处理线程空闲保活时间（毫秒）。 */
98+
private long streamKeepAliveMillis = 60_000L;
99+
100+
/** 单个 SSE 订阅的事件缓存上限。 */
101+
private int sseEventQueueCapacity = 1_024;
102+
88103
/**
89104
* 遇到失效连接等传输故障时是否允许 OkHttp 自动恢复。
90105
*/

0 commit comments

Comments
 (0)