Tri backpressure

[EarthChen]

What is the purpose of the change?

pre pr #15957

Checklist

• Make sure there is a GitHub_issue field for the change.
• Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• Write necessary unit-test to verify your logic correction. If the new feature or significant change is committed, please remember to add sample in dubbo samples project.
• Make sure gitHub actions can pass. Why the workflow is failing and how to fix it?

[EarthChen]

Add two APIs, namely ServerCallStreamObserver and ClientCallStreamObserver, to replace ServerStreamObserver and ClientStreamObserver. This is for aligning with gRPC's APIs and enabling cross-layer access between the remote and RPC layers.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.67347% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 60.70%. Comparing base (0bc5260) to head (c7e27b8).
⚠️ Report is 3 commits behind head on 3.3.

Files with missing lines	Patch %	Lines
.../dubbo/common/stream/ServerCallStreamObserver.java	0.00%	2 Missing ⚠️
.../dubbo/rpc/protocol/tri/call/TripleClientCall.java	66.66%	1 Missing and 1 partial ⚠️
.../protocol/tri/command/InitOnReadyQueueCommand.java	83.33%	1 Missing and 1 partial ⚠️
...dubbo/reactive/AbstractTripleReactorPublisher.java	83.33%	0 Missing and 1 partial ⚠️
...remoting/http12/h2/Http2ServerChannelObserver.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##                3.3   #15967      +/-   ##
============================================
- Coverage     60.70%   60.70%   -0.01%     
- Complexity    11708    11727      +19     
============================================
  Files          1946     1948       +2     
  Lines         88701    88732      +31     
  Branches      13374    13379       +5     
============================================
+ Hits          53849    53863      +14     
- Misses        29342    29351       +9     
- Partials       5510     5518       +8     

Flag	Coverage Δ
integration-tests-java21	32.22% <38.77%> (+0.01%)	⬆️
integration-tests-java8	32.40% <38.77%> (-0.02%)	⬇️
samples-tests-java21	32.16% <83.67%> (+0.07%)	⬆️
samples-tests-java8	29.70% <83.67%> (+0.04%)	⬆️
unit-tests-java11	58.97% <46.93%> (-0.02%)	⬇️
unit-tests-java17	58.45% <46.93%> (-0.03%)	⬇️
unit-tests-java21	58.44% <46.93%> (-0.03%)	⬇️
unit-tests-java25	58.43% <46.93%> (-0.02%)	⬇️
unit-tests-java8	59.00% <46.93%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[EarthChen]

test on apache/dubbo-samples#1278

[Copilot]

Pull request overview

This pull request implements backpressure handling for the Triple protocol's streaming functionality, aligning Dubbo's streaming API with gRPC's flow control patterns. The changes introduce a new mechanism to trigger initial onReady() callbacks and refactor stream observer interfaces for better gRPC compatibility.

• Adds InitOnReadyQueueCommand to trigger initial onReady() callbacks when channels are writable from creation
• Refactors stream observer interfaces to align with gRPC's pattern (ClientResponseObserver, ClientCallStreamObserver, ServerCallStreamObserver)
• Changes ClientCall.start() from returning StreamObserver<Object> to void, moving adapter creation to the invoker layer

Reviewed changes

Copilot reviewed 42 out of 42 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
InitOnReadyQueueCommand.java	New command to trigger initial onReady callback after stream creation
AbstractTripleClientStream.java	Enqueues InitOnReadyQueueCommand after stream channel initialization
TripleClientCall.java	Changes start() to void, adds null check for stream in isReady()
ClientCall.java	Updates start() signature to return void
TripleInvoker.java	Refactors streaming call setup with beforeStart() callbacks and adapter creation
CallStreamObserver.java	Moves to common package, adds default implementations for optional methods
ClientCallStreamObserver.java	New interface for client-side flow control, mirrors gRPC API
ServerCallStreamObserver.java	New interface for server-side flow control, mirrors gRPC API
ClientResponseObserver.java	New gRPC-compatible interface with beforeStart() callback
ClientStreamObserver.java	Deprecated in favor of ClientCallStreamObserver
ServerStreamObserver.java	Deprecated in favor of ServerCallStreamObserver
ClientCallToObserverAdapter.java	Implements new interfaces for compatibility
Http2ServerStreamObserver.java	Implements ServerCallStreamObserver
Http2ServerChannelObserver.java	Implements ServerCallStreamObserver with documentation
ClientTripleReactorPublisher.java	Implements ClientResponseObserver, refactors beforeStart()
AbstractTripleReactorPublisher.java	Adds onReadyHandler setup in onSubscribe()
TripleClientStreamTest.java	Updates test expectations for new InitOnReadyQueueCommand
BackpressureTest.java	Updates mock to match void start() signature
TripleInvokerTest.java	Updates mock to use doNothing() for void start()
Reactive/Mutiny plugin files	Updates imports to use common.stream interfaces

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The comment on line 78 appears to be incomplete or has a grammatical error. The sentence starts with "is triggered by onReady" but doesn't have a subject. The comment should be revised to clearly state what is triggered by onReady versus onStart.

Suggested change

	// is created before any data is sent
	// is triggered by onReady, not by onStart (which requires server headers).
	// is created before any data is sent.
	// The callback is triggered by onReady, not by onStart (which requires server headers).

[Copilot]

The comment states that enqueue should have been called twice for CreateStreamQueueCommand and InitOnReadyQueueCommand, but this comment appears after sendHeader and sendMessage have been called. At this point, enqueue has still only been called twice (both during stream initialization), so the comment is accurate but potentially misleading. The comment seems to suggest these calls happened after sendHeader and sendMessage, but they actually happened during stream construction.

[Copilot]

There's a potential race condition in the onReadyHandler. If the handler is invoked concurrently by multiple threads, line 82 could capture a non-null value while another thread is executing line 84 (setting onSubscribe to null). This could result in the callback being executed multiple times. Consider using a compareAndSet pattern or ensuring the handler is only invoked once at the framework level.

Suggested change

	// Only execute the callback once (on first onReady)
	Consumer<CallStreamObserver<?>> callback = onSubscribe;
	if (callback != null && subscription.isReady()) {
	onSubscribe = null; // Clear to prevent re-execution
	callback.accept(subscription);
	// Only execute the callback once (on first onReady), safely under synchronization
	synchronized (AbstractTripleReactorPublisher.this) {
	Consumer<CallStreamObserver<?>> callback = onSubscribe;
	if (callback != null && subscription.isReady()) {
	onSubscribe = null; // Clear to prevent re-execution
	callback.accept(subscription);
	}

[wangchengming666]

LGTM

[RainYuY]

LGTM, but interface default method should be fixed.

[RainYuY]

and here

[RainYuY]

LGTM

[zrlw]

LGTM