Skip to content

Commit fa844d0

Browse files
loong10kloong10k
committed
refactor: unify non-blocking sse client api
1 parent fb7e89a commit fa844d0

11 files changed

Lines changed: 833 additions & 169 deletions

‎scripts/evaluate-sdk.zsh‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
#!/bin/zsh
2+
set -uo pipefail
3+
4+
repo_root=${0:a:h:h}
5+
package_root=src/main/java/io/github/easy4j/openclaw
6+
result_dir=$repo_root/target/benchmark
7+
failed=0
8+
9+
for branch in feature/1.0.x feature/2.0.x feature/3.0.x; do
10+
print "STATIC openclaw $branch"
11+
git -C "$repo_root" cat-file -e \
12+
"${branch}:${package_root}/api/OpenClawSseClient.java" || failed=1
13+
git -C "$repo_root" cat-file -e \
14+
"${branch}:${package_root}/api/sse/SseSubscription.java" || failed=1
15+
if ! git -C "$repo_root" grep -q -E \
16+
'public[[:space:]]+OpenClawSseClient[[:space:]]+sse\(\)' \
17+
"$branch" -- "$package_root/OpenClawClient.java"; then
18+
print "FAIL missing OpenClawClient.sse(): $branch"
19+
failed=1
20+
fi
21+
if git -C "$repo_root" grep -n -E \
22+
'eventClient|[[:space:]]events\(\)|public[[:space:]]+void[[:space:]]+stop\(\)|subscribeSessionStream\(|Thread\.sleep|\.execute\(\)|Netty event-loop' \
23+
"$branch" -- "$package_root/api/*HttpClient.java" \
24+
"$package_root/api/*ChatClient.java" "$package_root/api/*SseClient.java"; then
25+
print "FAIL legacy/blocking SSE implementation remains: $branch"
26+
failed=1
27+
fi
28+
done
29+
30+
if [[ "$(git -C "$repo_root" branch --show-current)" != feature/3.0.x ]]; then
31+
print 'FAIL feature/3.0.x must be checked out for latest verification'
32+
failed=1
33+
else
34+
JAVA_HOME=${JAVA_HOME:-$(/usr/libexec/java_home -v 21)}
35+
PATH="$JAVA_HOME/bin:$PATH" mvn -q -f "$repo_root/pom.xml" \
36+
-Dtest=OpenClawSseApiShapeTest,OpenClawNonBlockingConcurrencyTest test || failed=1
37+
fi
38+
39+
python3 "$repo_root/scripts/verify-benchmark-results.py" \
40+
--sdk openclaw \
41+
--results "$result_dir/benchmark-results.csv" \
42+
--branches "$result_dir/branch-verification.tsv" || failed=1
43+
44+
if (( failed != 0 )); then
45+
print 'OPENCLAW_SDK_EVALUATOR_FAIL'
46+
exit 1
47+
fi
48+
print 'OPENCLAW_SDK_EVALUATOR_PASS'

‎scripts/run-concurrency-benchmark.zsh‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
#!/bin/zsh
2+
set -euo pipefail
3+
4+
repo_root=${0:a:h:h}
5+
sdk=openclaw
6+
workload=${1:-}
7+
concurrency=${2:-}
8+
result_file=${3:-$repo_root/target/benchmark/benchmark-results.csv}
9+
lock_dir=/tmp/easy4j-sdk-benchmark-exclusive.lock
10+
11+
if [[ "$workload" != http && "$workload" != sse ]]; then
12+
print -u2 'Usage: run-concurrency-benchmark.zsh <http|sse> <concurrency> [result.csv]'
13+
exit 64
14+
fi
15+
if [[ ! "$concurrency" =~ '^[1-9][0-9]*$' ]]; then
16+
print -u2 'concurrency must be a positive integer'
17+
exit 64
18+
fi
19+
if ! mkdir "$lock_dir" 2>/dev/null; then
20+
print -u2 "another SDK benchmark holds the global lock: $lock_dir"
21+
exit 75
22+
fi
23+
24+
benchmark_pid=
25+
cleanup() {
26+
if [[ -n "$benchmark_pid" ]] && kill -0 "$benchmark_pid" 2>/dev/null; then
27+
kill "$benchmark_pid" 2>/dev/null || true
28+
fi
29+
rmdir "$lock_dir" 2>/dev/null || true
30+
}
31+
trap cleanup EXIT INT TERM
32+
33+
mkdir -p "$repo_root/target/benchmark"
34+
classpath_file=$repo_root/target/benchmark/test-classpath.txt
35+
metrics_file=$repo_root/target/benchmark/${sdk}-${workload}-${concurrency}.metrics
36+
samples_file=$repo_root/target/benchmark/${sdk}-${workload}-${concurrency}.samples
37+
log_file=$repo_root/target/benchmark/${sdk}-${workload}-${concurrency}.log
38+
rm -f "$metrics_file" "$samples_file" "$log_file"
39+
40+
JAVA_HOME=${JAVA_HOME:-$(/usr/libexec/java_home -v 21)}
41+
PATH="$JAVA_HOME/bin:$PATH" mvn -q -f "$repo_root/pom.xml" -DskipTests test-compile \
42+
dependency:build-classpath -Dmdep.outputFile="$classpath_file"
43+
classpath="$repo_root/target/test-classes:$repo_root/target/classes:$(<"$classpath_file")"
44+
started_epoch_ms=$(python3 -c 'import time; print(time.time_ns() // 1000000)')
45+
"$JAVA_HOME/bin/java" -cp "$classpath" io.github.easy4j.openclaw.OpenClawConcurrencyBenchmark \
46+
"$workload" "$concurrency" "$metrics_file" >"$log_file" 2>&1 &
47+
benchmark_pid=$!
48+
49+
while kill -0 "$benchmark_pid" 2>/dev/null; do
50+
ps -o %cpu= -o rss= -p "$benchmark_pid" | awk 'NF == 2 { print $1, $2 }' >>"$samples_file"
51+
sleep 0.2
52+
done
53+
set +e
54+
wait "$benchmark_pid"
55+
benchmark_exit=$?
56+
set -e
57+
benchmark_pid=
58+
ended_epoch_ms=$(python3 -c 'import time; print(time.time_ns() // 1000000)')
59+
60+
if [[ ! -s "$metrics_file" ]]; then
61+
print -u2 "benchmark did not produce metrics; see $log_file"
62+
exit ${benchmark_exit:-1}
63+
fi
64+
operations=$(awk -F= '$1 == "operations" { print $2 }' "$metrics_file")
65+
errors=$(awk -F= '$1 == "errors" { print $2 }' "$metrics_file")
66+
duration_seconds=$(awk -F= '$1 == "duration_seconds" { print $2 }' "$metrics_file")
67+
throughput=$(awk -F= '$1 == "throughput_per_sec" { print $2 }' "$metrics_file")
68+
read avg_cpu peak_cpu peak_rss_mb <<<"$(awk '
69+
BEGIN { sum=0; count=0; peak_cpu=0; peak_rss=0 }
70+
NF == 2 { sum += $1; count++; if ($1 > peak_cpu) peak_cpu=$1; if ($2 > peak_rss) peak_rss=$2 }
71+
END { printf "%.3f %.3f %.3f", count ? sum/count : 0, peak_cpu, peak_rss/1024 }
72+
' "$samples_file")"
73+
benchmark_status=PASS
74+
if (( benchmark_exit != 0 || errors != 0 )) || [[ "$peak_cpu" == 0.000 ]]; then
75+
benchmark_status=FAIL
76+
fi
77+
git_sha=$(git -C "$repo_root" rev-parse HEAD)
78+
if [[ ! -s "$result_file" ]]; then
79+
print 'sdk,workload,concurrency,started_epoch_ms,ended_epoch_ms,duration_seconds,operations,errors,throughput_per_sec,avg_cpu_pct,peak_cpu_pct,peak_rss_mb,status,git_sha' >"$result_file"
80+
fi
81+
print "$sdk,$workload,$concurrency,$started_epoch_ms,$ended_epoch_ms,$duration_seconds,$operations,$errors,$throughput,$avg_cpu,$peak_cpu,$peak_rss_mb,$benchmark_status,$git_sha" >>"$result_file"
82+
print "$sdk $workload concurrency=$concurrency status=$benchmark_status cpu(avg/peak)=$avg_cpu/$peak_cpu rss_peak_mb=$peak_rss_mb throughput=$throughput"
83+
if [[ "$benchmark_status" != PASS ]]; then
84+
print -u2 "benchmark failed; see $log_file"
85+
exit 1
86+
fi

‎scripts/verify-benchmark-results.py‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
#!/usr/bin/env python3
2+
"""校验单个 SDK 的分支测试和串行并发压测结果。"""
3+
4+
import argparse
5+
import csv
6+
from pathlib import Path
7+
8+
9+
CONCURRENCY_LEVELS = {"100", "300", "500", "800", "1000"}
10+
WORKLOADS = {"http", "sse"}
11+
BRANCHES = {"feature/1.0.x", "feature/2.0.x", "feature/3.0.x"}
12+
13+
14+
def parse_args():
15+
parser = argparse.ArgumentParser()
16+
parser.add_argument("--sdk", required=True)
17+
parser.add_argument("--results", required=True)
18+
parser.add_argument("--branches", required=True)
19+
return parser.parse_args()
20+
21+
22+
def read_rows(path, delimiter=","):
23+
target = Path(path)
24+
if not target.is_file():
25+
raise SystemExit(f"FAIL missing result file: {target}")
26+
with target.open(newline="", encoding="utf-8") as handle:
27+
return list(csv.DictReader(handle, delimiter=delimiter))
28+
29+
30+
def verify_branches(sdk, path):
31+
rows = read_rows(path, "\t")
32+
expected = {(sdk, branch) for branch in BRANCHES}
33+
actual = {
34+
(row.get("sdk"), row.get("branch"))
35+
for row in rows
36+
if row.get("status") == "PASS"
37+
}
38+
if len(rows) != 3 or actual != expected:
39+
raise SystemExit("FAIL branch verification must contain exactly 3 passing rows")
40+
41+
42+
def verify_benchmarks(sdk, path):
43+
rows = read_rows(path)
44+
expected = {
45+
(sdk, workload, concurrency)
46+
for workload in WORKLOADS
47+
for concurrency in CONCURRENCY_LEVELS
48+
}
49+
actual = {
50+
(row.get("sdk"), row.get("workload"), row.get("concurrency"))
51+
for row in rows
52+
}
53+
if len(rows) != 10 or actual != expected:
54+
raise SystemExit("FAIL benchmark results must contain exactly 10 unique runs")
55+
56+
intervals = []
57+
for row in rows:
58+
if row.get("status") != "PASS" or int(row.get("errors", "-1")) != 0:
59+
raise SystemExit(f"FAIL benchmark errors: {row}")
60+
if float(row.get("avg_cpu_pct", "-1")) < 0:
61+
raise SystemExit(f"FAIL invalid average CPU: {row}")
62+
if float(row.get("peak_cpu_pct", "0")) <= 0:
63+
raise SystemExit(f"FAIL invalid peak CPU: {row}")
64+
if float(row.get("peak_rss_mb", "0")) <= 0:
65+
raise SystemExit(f"FAIL invalid peak RSS: {row}")
66+
if float(row.get("throughput_per_sec", "0")) <= 0:
67+
raise SystemExit(f"FAIL invalid throughput: {row}")
68+
start = int(row["started_epoch_ms"])
69+
end = int(row["ended_epoch_ms"])
70+
if end <= start:
71+
raise SystemExit(f"FAIL invalid interval: {row}")
72+
intervals.append((start, end, row))
73+
74+
intervals.sort(key=lambda value: value[0])
75+
for previous, current in zip(intervals, intervals[1:]):
76+
if current[0] < previous[1]:
77+
raise SystemExit(
78+
f"FAIL overlapping benchmark runs: {previous[2]} and {current[2]}"
79+
)
80+
81+
82+
def main():
83+
args = parse_args()
84+
verify_branches(args.sdk, args.branches)
85+
verify_benchmarks(args.sdk, args.results)
86+
print(f"{args.sdk.upper()}_BENCHMARK_RESULTS_PASS")
87+
88+
89+
if __name__ == "__main__":
90+
main()

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

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@
5757
*
5858
* @see OpenClawGatewayWsClient
5959
* @see OpenClawChatClient
60+
* @see OpenClawSseClient
6061
* @see OpenClawEmbeddingsClient
6162
* @see OpenClawResponsesClient
6263
* @see OpenClawToolInvokeClient
@@ -69,6 +70,7 @@ public class OpenClawClient implements AutoCloseable {
6970

7071
private final OpenClawWebhookClient gatewayHttpClient;
7172
private final OpenClawChatClient chatClient;
73+
private final OpenClawSseClient sseClient;
7274
private final OpenClawEmbeddingsClient embeddingsClient;
7375
private final OpenClawResponsesClient responsesClient;
7476
private final OpenClawToolInvokeClient toolsInvokeClient;
@@ -151,14 +153,17 @@ private OpenClawClient(OpenClawHttpClientConfig httpConfig, OpenClawCliConfig cl
151153
// HTTP 子系统初始化（enabled=false 时不创建，字段为 null）
152154
if (httpEnabled) {
153155
this.gatewayHttpClient = new OpenClawWebhookClient(httpConfig, objectMapper, httpClient);
154-
this.chatClient = new OpenClawChatClient(httpConfig, objectMapper, httpClient);
156+
this.sseClient = new OpenClawSseClient(httpConfig, objectMapper, httpClient);
157+
this.chatClient = new OpenClawChatClient(
158+
httpConfig, objectMapper, httpClient, sseClient);
155159
this.embeddingsClient = new OpenClawEmbeddingsClient(httpConfig, objectMapper, httpClient);
156160
this.responsesClient = new OpenClawResponsesClient(httpConfig, objectMapper, httpClient);
157161
this.toolsInvokeClient = new OpenClawToolInvokeClient(httpConfig, objectMapper, httpClient);
158162
this.wsClient = new OpenClawGatewayWsClient(httpConfig);
159163
} else {
160164
this.gatewayHttpClient = null;
161165
this.chatClient = null;
166+
this.sseClient = null;
162167
this.embeddingsClient = null;
163168
this.responsesClient = null;
164169
this.toolsInvokeClient = null;
@@ -215,13 +220,15 @@ public OpenClawClient(OpenClawHttpClientConfig httpConfig,
215220
OpenClawCliConfig cliConfig,
216221
OpenClawWebhookClient gatewayHttpClient,
217222
OpenClawChatClient chatClient,
223+
OpenClawSseClient sseClient,
218224
OpenClawEmbeddingsClient embeddingsClient,
219225
OpenClawResponsesClient responsesClient,
220226
OpenClawToolInvokeClient toolsInvokeClient,
221227
OpenClawCli cli,
222228
OpenClawGatewayWsClient wsClient) {
223229
this.gatewayHttpClient = gatewayHttpClient;
224230
this.chatClient = chatClient;
231+
this.sseClient = sseClient;
225232
this.embeddingsClient = embeddingsClient;
226233
this.responsesClient = responsesClient;
227234
this.toolsInvokeClient = toolsInvokeClient;
@@ -416,6 +423,15 @@ public OpenClawChatClient chat() {
416423
return chatClient;
417424
}
418425

426+
/**
427+
* 返回 SSE 场景客户端。
428+
*
429+
* @return SSE 客户端；HTTP 子系统未启用时为 {@code null}
430+
*/
431+
public OpenClawSseClient sse() {
432+
return sseClient;
433+
}
434+
419435
/**
420436
* HTTP system {@link OkHttpClient}.
421437
* <p>inject,lifecyclemanaged by caller.</p>
@@ -686,6 +702,7 @@ public OpenClawCli cli() {
686702
@Override
687703
public void close() {
688704
// 逐一释放所有子客户端资源，任一失败不影响其他
705+
closeQuietly(sseClient);
689706
closeQuietly(gatewayHttpClient);
690707
closeQuietly(chatClient);
691708
closeQuietly(embeddingsClient);

0 commit comments

Comments
 (0)