fix(stream): use subscribeOn instead of publishOn for blocking stream sources (#410)

AlbumenJ · web-flow · commit 6486596705e3 · 2025-12-31T16:06:09.000+08:00
diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/AnthropicChatModel.java b/agentscope-core/src/main/java/io/agentscope/core/model/AnthropicChatModel.java
@@ -171,7 +171,8 @@ protected Flux<ChatResponse> doStream(
                                     // Convert the SDK's Stream to Flux
                                     return AnthropicResponseParser.parseStreamEvents(
                                                     Flux.fromStream(streamResponse.stream())
-                                                            .publishOn(Schedulers.boundedElastic()),
+                                                            .subscribeOn(
+                                                                    Schedulers.boundedElastic()),
                                                     startTime)
                                             .doFinally(
                                                     signalType -> {
diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/GeminiChatModel.java b/agentscope-core/src/main/java/io/agentscope/core/model/GeminiChatModel.java
@@ -207,7 +207,7 @@ protected Flux<ChatResponse> doStream(
 
                                     // Convert ResponseStream to Flux
                                     return Flux.fromIterable(responseStream)
-                                            .publishOn(Schedulers.boundedElastic())
+                                            .subscribeOn(Schedulers.boundedElastic())
                                             .map(
                                                     response ->
                                                             formatter.parseResponse(
diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/OpenAIChatModel.java b/agentscope-core/src/main/java/io/agentscope/core/model/OpenAIChatModel.java
@@ -171,7 +171,7 @@ protected Flux<ChatResponse> doStream(
 
                                     // Convert the SDK's Stream to Flux
                                     return Flux.fromStream(streamResponse.stream())
-                                            .publishOn(Schedulers.boundedElastic())
+                                            .subscribeOn(Schedulers.boundedElastic())
                                             .map(chunk -> formatter.parseResponse(chunk, startTime))
                                             .filter(Objects::nonNull)
                                             .doFinally(
diff --git a/agentscope-core/src/main/java/io/agentscope/core/model/transport/OkHttpTransport.java b/agentscope-core/src/main/java/io/agentscope/core/model/transport/OkHttpTransport.java
@@ -188,7 +188,7 @@ public Flux<String> stream(HttpRequest request) {
                                 closeQuietly(response);
                             }
                         })
-                .publishOn(Schedulers.boundedElastic());
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     @Override
diff --git a/agentscope-examples/model-request-compression/src/main/java/io/agentscope/examples/compression/extra/CompressingOkHttpTransport.java b/agentscope-examples/model-request-compression/src/main/java/io/agentscope/examples/compression/extra/CompressingOkHttpTransport.java
@@ -244,7 +244,7 @@ public Flux<String> stream(HttpRequest request) {
                                 closeQuietly(response);
                             }
                         })
-                .publishOn(Schedulers.boundedElastic());
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     @Override

Original file line number	Diff line number	Diff line change
`@@ -188,7 +188,7 @@ public Flux<String> stream(HttpRequest request) {`
`188`	`188`	`closeQuietly(response);`
`189`	`189`	`}`
`190`	`190`	`})`
`191`		`- .publishOn(Schedulers.boundedElastic());`
	`191`	`+ .subscribeOn(Schedulers.boundedElastic());`
`192`	`192`	`}`
`193`	`193`
`194`	`194`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -244,7 +244,7 @@ public Flux<String> stream(HttpRequest request) {`
`244`	`244`	`closeQuietly(response);`
`245`	`245`	`}`
`246`	`246`	`})`
`247`		`- .publishOn(Schedulers.boundedElastic());`
	`247`	`+ .subscribeOn(Schedulers.boundedElastic());`
`248`	`248`	`}`
`249`	`249`
`250`	`250`	`@Override`