Skip to content

Commit 3fd8cd4

Browse files
loong10kloong10k
committed
perf: bound streams and propagate HTTP cancellation
1 parent 6e69220 commit 3fd8cd4

7 files changed

Lines changed: 125 additions & 10 deletions

File tree

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package io.github.easy4j.openclaw;
2+
3+
/**
4+
* 将调用方取消信号绑定到底层 HTTP Call。
5+
*/
6+
@FunctionalInterface
7+
public interface HttpCallCancellation {
8+
9+
/**
10+
* 注册取消动作。
11+
*
12+
* @param callback 取消时执行的动作
13+
* @return 请求结束后用于注销动作的句柄
14+
*/
15+
AutoCloseable onCancel(Runnable callback);
16+
}

‎src/main/java/io/github/easy4j/openclaw/OpenClawClient.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -448,6 +448,11 @@ public ChatResponse chatCompletion(ChatRequest request) {
448448
return chatClient.chatCompletion(request);
449449
}
450450

451+
/** 发送支持调用方取消的 Chat Completions 请求。 */
452+
public ChatResponse chatCompletion(ChatRequest request, HttpCallCancellation cancellation) {
453+
return chatClient.chatCompletion(request, null, cancellation);
454+
}
455+
451456
/**
452457
* 发送 Chat Completions 请求，携带自定义请求头。
453458
*/
@@ -631,6 +636,10 @@ public ToolInvokeResult toolInvoke(ToolInvokeRequest request) {
631636
return toolsInvokeClient.invoke(request);
632637
}
633638

639+
public ToolInvokeResult toolInvoke(ToolInvokeRequest request, HttpCallCancellation cancellation) {
640+
return toolsInvokeClient.invoke(request, cancellation);
641+
}
642+
634643
// ============================================================
635644
// CLI
636645
// ============================================================

‎src/main/java/io/github/easy4j/openclaw/OpenClawHttpClientConfig.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,18 @@ public class OpenClawHttpClientConfig {
8989
/** 单主机异步请求最大并发数 */
9090
private int maxRequestsPerHost = 64;
9191

92+
/** SSE 响应消费线程池核心线程数 */
93+
private int sseCorePoolSize = 16;
94+
95+
/** SSE 响应消费线程池最大线程数 */
96+
private int sseMaxPoolSize = 16;
97+
98+
/** SSE 响应消费线程池有界队列容量 */
99+
private int sseQueueCapacity = 128;
100+
101+
/** SSE 响应消费线程空闲保活时间（毫秒） */
102+
private long sseKeepAliveMillis = 60_000L;
103+
92104
/** 遇到失效连接等传输故障时是否允许 OkHttp 自动恢复 */
93105
private boolean retryOnConnectionFailure = true;
94106

‎src/main/java/io/github/easy4j/openclaw/api/OpenClawChatClient.java‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import com.fasterxml.jackson.databind.ObjectMapper;
44
import io.github.easy4j.openclaw.OpenClawHttpClientConfig;
5+
import io.github.easy4j.openclaw.HttpCallCancellation;
56
import io.github.easy4j.openclaw.exception.OpenClawHttpException;
67
import io.github.easy4j.openclaw.util.OpenClawStrings;
78
import io.github.easy4j.openclaw.api.model.*;
@@ -17,8 +18,11 @@
1718
import java.util.Set;
1819
import java.util.concurrent.ConcurrentHashMap;
1920
import java.util.concurrent.ExecutorService;
20-
import java.util.concurrent.Executors;
21+
import java.util.concurrent.LinkedBlockingQueue;
2122
import java.util.concurrent.RejectedExecutionException;
23+
import java.util.concurrent.ThreadPoolExecutor;
24+
import java.util.concurrent.TimeUnit;
25+
import java.util.concurrent.atomic.AtomicInteger;
2226

2327
/**
2428
* Chat Completions API 客户端。
@@ -28,19 +32,32 @@
2832
@Slf4j
2933
public class OpenClawChatClient extends OpenClawHttpClient {
3034

31-
private final ExecutorService streamExecutor = Executors.newCachedThreadPool(runnable -> {
32-
Thread thread = new Thread(runnable, "openclaw-sse-consumer");
33-
thread.setDaemon(true);
34-
return thread;
35-
});
35+
private final ExecutorService streamExecutor;
3636
private final Set<Call> activeStreamCalls = ConcurrentHashMap.newKeySet();
3737

3838
public OpenClawChatClient(OpenClawHttpClientConfig config) {
3939
super(config);
40+
this.streamExecutor = createStreamExecutor(config);
4041
}
4142

4243
public OpenClawChatClient(OpenClawHttpClientConfig config, ObjectMapper objectMapper, OkHttpClient httpClient) {
4344
super(config, objectMapper, httpClient);
45+
this.streamExecutor = createStreamExecutor(config);
46+
}
47+
48+
private static ExecutorService createStreamExecutor(OpenClawHttpClientConfig config) {
49+
int corePoolSize = Math.max(1, config.getSseCorePoolSize());
50+
int maxPoolSize = Math.max(corePoolSize, config.getSseMaxPoolSize());
51+
int queueCapacity = Math.max(1, config.getSseQueueCapacity());
52+
long keepAliveMillis = Math.max(1L, config.getSseKeepAliveMillis());
53+
AtomicInteger threadIndex = new AtomicInteger();
54+
return new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveMillis, TimeUnit.MILLISECONDS,
55+
new LinkedBlockingQueue<>(queueCapacity), runnable -> {
56+
Thread thread = new Thread(runnable,
57+
"openclaw-sse-consumer-" + threadIndex.incrementAndGet());
58+
thread.setDaemon(true);
59+
return thread;
60+
}, new ThreadPoolExecutor.AbortPolicy());
4461
}
4562

4663
// ============================================================
@@ -52,6 +69,12 @@ public ChatResponse chatCompletion(ChatRequest request) {
5269
}
5370

5471
public ChatResponse chatCompletion(ChatRequest request, Map<String, String> headers) {
72+
return chatCompletion(request, headers, null);
73+
}
74+
75+
/** 发送支持调用方取消的 Chat Completion。 */
76+
public ChatResponse chatCompletion(ChatRequest request, Map<String, String> headers,
77+
HttpCallCancellation cancellation) {
5578
Objects.requireNonNull(request, "request");
5679

5780
debug("=== Chat Completion Request ===");
@@ -91,7 +114,7 @@ public ChatResponse chatCompletion(ChatRequest request, Map<String, String> head
91114

92115
String json;
93116
try {
94-
json = postJson(OpenClawConstants.ENDPOINT_CHAT_COMPLETIONS, normalized, headers);
117+
json = postJson(OpenClawConstants.ENDPOINT_CHAT_COMPLETIONS, normalized, headers, cancellation);
95118
} catch (OpenClawHttpException e) {
96119
error("Chat completion failed: status={}, message={}", e.getStatusCode(), e.getMessage());
97120
throw e;

‎src/main/java/io/github/easy4j/openclaw/api/OpenClawHttpClient.java‎

Lines changed: 30 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import com.fasterxml.jackson.databind.DeserializationFeature;
44
import com.fasterxml.jackson.databind.ObjectMapper;
55
import io.github.easy4j.openclaw.OpenClawHttpClientConfig;
6+
import io.github.easy4j.openclaw.HttpCallCancellation;
67
import io.github.easy4j.openclaw.OpenClawOkHttpClientFactory;
78
import io.github.easy4j.openclaw.exception.OpenClawHttpException;
89
import io.github.easy4j.openclaw.util.OpenClawStrings;
@@ -105,6 +106,12 @@ protected String postJson(String path, Object body) {
105106
* POST JSON 请求，带额外请求头。
106107
*/
107108
protected String postJson(String path, Object body, Map<String, String> headers) {
109+
return postJson(path, body, headers, null);
110+
}
111+
112+
/** POST JSON 请求，并将调用方取消信号绑定到底层 Call。 */
113+
protected String postJson(String path, Object body, Map<String, String> headers,
114+
HttpCallCancellation cancellation) {
108115
String url = resolveUrl(path);
109116
debug("POST JSON: path={}, url={}", path, url);
110117

@@ -116,7 +123,7 @@ protected String postJson(String path, Object body, Map<String, String> headers)
116123
.post(RequestBody.create(json, JSON))
117124
.build();
118125

119-
return execute(request, url);
126+
return execute(request, url, cancellation);
120127
} catch (OpenClawHttpException e) {
121128
throw e;
122129
} catch (IOException e) {
@@ -145,10 +152,18 @@ protected String getJson(String path) {
145152
* 执行请求。
146153
*/
147154
protected String execute(Request request, String url) throws IOException {
155+
return execute(request, url, null);
156+
}
157+
158+
/** 执行支持协作式取消的请求。 */
159+
protected String execute(Request request, String url,
160+
HttpCallCancellation cancellation) throws IOException {
148161
debug("Executing request: {} {}", request.method(), request.url());
149162
debug("Request headers: {}", request.headers());
150163

151-
try (Response response = httpClient.newCall(request).execute()) {
164+
Call call = httpClient.newCall(request);
165+
AutoCloseable registration = cancellation != null ? cancellation.onCancel(call::cancel) : null;
166+
try (Response response = call.execute()) {
152167
int status = response.code();
153168
String respBody = response.body() != null ? response.body().string() : "";
154169

@@ -165,6 +180,19 @@ protected String execute(Request request, String url) throws IOException {
165180
throw new OpenClawHttpException("Request returned status " + status, status, respBody);
166181
}
167182
return respBody;
183+
} finally {
184+
closeRegistration(registration);
185+
}
186+
}
187+
188+
private void closeRegistration(AutoCloseable registration) {
189+
if (registration == null) {
190+
return;
191+
}
192+
try {
193+
registration.close();
194+
} catch (Exception error) {
195+
debug("Failed to unregister HTTP cancellation callback: {}", error.getMessage());
168196
}
169197
}
170198

‎src/main/java/io/github/easy4j/openclaw/api/OpenClawToolInvokeClient.java‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import com.fasterxml.jackson.databind.ObjectMapper;
44
import io.github.easy4j.openclaw.OpenClawHttpClientConfig;
5+
import io.github.easy4j.openclaw.HttpCallCancellation;
56
import io.github.easy4j.openclaw.exception.OpenClawHttpException;
67
import io.github.easy4j.openclaw.util.OpenClawStrings;
78
import io.github.easy4j.openclaw.api.model.ToolInvokeRequest;
@@ -28,6 +29,10 @@ public OpenClawToolInvokeClient(OpenClawHttpClientConfig config, ObjectMapper ob
2829
}
2930

3031
public ToolInvokeResult invoke(ToolInvokeRequest request) {
32+
return invoke(request, null);
33+
}
34+
35+
public ToolInvokeResult invoke(ToolInvokeRequest request, HttpCallCancellation cancellation) {
3136
Objects.requireNonNull(request, "request");
3237

3338
debug("=== Tool Invoke Request ===");
@@ -47,7 +52,9 @@ public ToolInvokeResult invoke(ToolInvokeRequest request) {
4752

4853
debug("Sending tool invoke request...");
4954

50-
try (Response response = httpClient.newCall(httpRequest).execute()) {
55+
Call call = httpClient.newCall(httpRequest);
56+
AutoCloseable registration = cancellation != null ? cancellation.onCancel(call::cancel) : null;
57+
try (Response response = call.execute()) {
5158
int status = response.code();
5259
String respBody = response.body() != null ? response.body().string() : "";
5360

@@ -73,11 +80,24 @@ public ToolInvokeResult invoke(ToolInvokeRequest request) {
7380
ToolInvokeResult result = parse(respBody, ToolInvokeResult.class);
7481
debug("Tool invoke success, ok: {}", result.getOk());
7582
return result;
83+
} finally {
84+
closeRegistration(registration);
7685
}
7786
} catch (OpenClawHttpException e) {
7887
throw e;
7988
} catch (Exception e) {
8089
throw new OpenClawHttpException("POST /tools/invoke failed: " + e.getMessage(), e);
8190
}
8291
}
92+
93+
private void closeRegistration(AutoCloseable registration) {
94+
if (registration == null) {
95+
return;
96+
}
97+
try {
98+
registration.close();
99+
} catch (Exception error) {
100+
debug("Failed to unregister tool cancellation callback: {}", error.getMessage());
101+
}
102+
}
83103
}

‎src/test/java/io/github/easy4j/openclaw/OpenClawHttpApiCoverageTest.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,13 @@ void shouldCoverChatModelsStreamingAndValidation() throws Exception {
8787

8888
assertEquals("chat-id", chat.chatCompletion(agentRequest).getId());
8989
assertEquals("chat-id", chat.chatCompletion(modelRequest, map("X-Custom", "value")).getId());
90+
AtomicBoolean cancellationRegistered = new AtomicBoolean();
91+
assertThrows(OpenClawHttpException.class, () -> chat.chatCompletion(agentRequest, null, callback -> {
92+
cancellationRegistered.set(true);
93+
callback.run();
94+
return () -> { };
95+
}));
96+
assertTrue(cancellationRegistered.get());
9097
assertNotNull(chat.listModels());
9198
assertNotNull(chat.getModel("model with space"));
9299
chat.health();

0 commit comments

Comments
 (0)