modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: implement demand-based request handling in ResponseSubscribers (#516)

Browse source 
- Add hasRequestedDemand volatile boolean flag to AggregateSubscriber and BodilessResponseLineSubscriber
- Modify hookOnSubscribe to request Long.MAX_VALUE only on first demand request
- Guard event emission in hookOnComplete with hasRequestedDemand check
- Prevents unnecessary data processing when no downstream demand exists
- Improves backpressure handling and resource efficiency

Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
Christian Tzolov 2025-08-26 10:30:31 +02:00 • committed by Christian Tzolov
commit 0f635eec67
1 changed files with 27 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

36
mcp/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java
View file

@ -223,6 +223,8 @@ class ResponseSubscribers {
*/ */
private ResponseInfo responseInfo; private ResponseInfo responseInfo;
volatile boolean hasRequestedDemand = false;
/** /**
* Creates a new JsonLineSubscriber that will emit parsed JSON-RPC messages. * Creates a new JsonLineSubscriber that will emit parsed JSON-RPC messages.
* @param sink the {@link FluxSink} to emit parsed {@link ResponseEvent} objects * @param sink the {@link FluxSink} to emit parsed {@link ResponseEvent} objects
@ -236,7 +238,13 @@ class ResponseSubscribers {
@Override @Override
protected void hookOnSubscribe(Subscription subscription) { protected void hookOnSubscribe(Subscription subscription) {
sink.onRequest(subscription::request);
sink.onRequest(n -> {
if (!hasRequestedDemand) {
subscription.request(Long.MAX_VALUE);
}
hasRequestedDemand = true;
});
// Register disposal callback to cancel subscription when Flux is disposed // Register disposal callback to cancel subscription when Flux is disposed
sink.onDispose(subscription::cancel); sink.onDispose(subscription::cancel);
@ -249,8 +257,11 @@ class ResponseSubscribers {
@Override @Override
protected void hookOnComplete() { protected void hookOnComplete() {
String data = this.eventBuilder.toString();
this.sink.next(new AggregateResponseEvent(responseInfo, data)); if (hasRequestedDemand) {
String data = this.eventBuilder.toString();
this.sink.next(new AggregateResponseEvent(responseInfo, data));
}
this.sink.complete(); this.sink.complete();
} }
@ -271,6 +282,8 @@ class ResponseSubscribers {
private final ResponseInfo responseInfo; private final ResponseInfo responseInfo;
volatile boolean hasRequestedDemand = false;
public BodilessResponseLineSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) { public BodilessResponseLineSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) {
this.sink = sink; this.sink = sink;
this.responseInfo = responseInfo; this.responseInfo = responseInfo;
@ -280,7 +293,10 @@ class ResponseSubscribers {
protected void hookOnSubscribe(Subscription subscription) { protected void hookOnSubscribe(Subscription subscription) {
sink.onRequest(n -> { sink.onRequest(n -> {
subscription.request(n); if (!hasRequestedDemand) {
subscription.request(Long.MAX_VALUE);
}
hasRequestedDemand = true;
}); });
// Register disposal callback to cancel subscription when Flux is disposed // Register disposal callback to cancel subscription when Flux is disposed
@ -291,11 +307,13 @@ class ResponseSubscribers {
@Override @Override
protected void hookOnComplete() { protected void hookOnComplete() {
// emit dummy event to be able to inspect the response info if (hasRequestedDemand) {
// this is a shortcut allowing for a more streamlined processing using // emit dummy event to be able to inspect the response info
// operator composition instead of having to deal with the CompletableFuture // this is a shortcut allowing for a more streamlined processing using
// along the Subscriber for inspecting the result // operator composition instead of having to deal with the
this.sink.next(new DummyEvent(responseInfo)); // CompletableFuture along the Subscriber for inspecting the result
this.sink.next(new DummyEvent(responseInfo));
}
this.sink.complete(); this.sink.complete();
} }