diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamObserver.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamObserver.java index b28499a77f..d20d2f56b1 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamObserver.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamObserver.java @@ -29,6 +29,8 @@ public interface ClientStreamObserver extends CallStreamObserver { * request()} may not be called before the call is started, a number of initial requests may be * specified. */ - void disableAutoRequest(); + default void disableAutoRequest() { + disableAutoFlowControl(); + } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamObserver.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamObserver.java index 3ca3f1d125..9d3a9b2641 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamObserver.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamObserver.java @@ -21,7 +21,8 @@ import org.apache.dubbo.rpc.protocol.tri.observer.CallStreamObserver; public interface ServerStreamObserver extends CallStreamObserver { - - void disableAutoInboundFlowControl(); + default void disableAutoInboundFlowControl() { + disableAutoFlowControl(); + } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/CallStreamObserver.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/CallStreamObserver.java index 8f0037fd92..362ab9ed4b 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/CallStreamObserver.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/CallStreamObserver.java @@ -43,5 +43,12 @@ public interface CallStreamObserver extends StreamObserver { */ void setCompression(String compression); + /** + * Swaps to manual flow control where no message will be delivered to {@link + * StreamObserver#onNext(Object)} unless it is {@link #request request()}ed. Since {@code + * request()} may not be called before the call is started, a number of initial requests may be + * specified. + */ + void disableAutoFlowControl(); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ClientCallToObserverAdapter.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ClientCallToObserverAdapter.java index 8002d2ef67..37f17694c3 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ClientCallToObserverAdapter.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ClientCallToObserverAdapter.java @@ -76,7 +76,7 @@ public class ClientCallToObserverAdapter extends CancelableStreamObserver } @Override - public void disableAutoRequest() { + public void disableAutoFlowControl() { call.setAutoRequest(false); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java index 8329dda8da..edcd555a21 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ServerCallToObserverAdapter.java @@ -105,7 +105,7 @@ public class ServerCallToObserverAdapter extends CancelableStreamObserver } @Override - public void disableAutoInboundFlowControl() { + public void disableAutoFlowControl() { call.disableAutoRequestN(); }