From 71d4f06c90f6b9214ef0a0a594167f59b96f1f9c Mon Sep 17 00:00:00 2001 From: icodening Date: Sat, 18 May 2024 21:50:03 +0800 Subject: [PATCH 1/5] Fix the issue of errors when running out of HTTP2 stream IDs --- .../rpc/protocol/tri/TripleHttp2Protocol.java | 1 + .../protocol/tri/call/TripleClientCall.java | 96 +++++++++++++++++-- .../tri/command/DataQueueCommand.java | 4 +- .../tri/command/EndStreamQueueCommand.java | 2 +- .../tri/command/HeaderQueueCommand.java | 2 +- .../protocol/tri/command/QueuedCommand.java | 2 +- .../tri/command/TextDataQueueCommand.java | 2 +- .../tri/stream/TripleClientStream.java | 9 +- 8 files changed, 100 insertions(+), 18 deletions(-) diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java index ba1ae2e05e..cd2bb791a4 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java @@ -166,6 +166,7 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter()); List handlers = new ArrayList<>(); handlers.add(new ChannelHandlerPretender(codec)); + handlers.add(new ChannelHandlerPretender(new FlushConsolidationHandler(64, true))); handlers.add(new ChannelHandlerPretender(new Http2MultiplexHandler(new ChannelDuplexHandler()))); handlers.add(new ChannelHandlerPretender(new TriplePingPongHandler(UrlUtils.getCloseTimeout(url)))); handlers.add(new ChannelHandlerPretender(new TripleGoAwayHandler())); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java index 330deaa0bd..cba3b56cdf 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java @@ -16,10 +16,16 @@ */ package org.apache.dubbo.rpc.protocol.tri.call; +import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException; +import io.netty.util.concurrent.Future; + import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.stream.StreamObserver; +import org.apache.dubbo.common.threadpool.serial.SerializingExecutor; +import org.apache.dubbo.remoting.RemotingException; import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient; +import org.apache.dubbo.rpc.RpcException; import org.apache.dubbo.rpc.TriRpcStatus; import org.apache.dubbo.rpc.model.FrameworkModel; import org.apache.dubbo.rpc.protocol.tri.RequestMetadata; @@ -32,6 +38,8 @@ import org.apache.dubbo.rpc.protocol.tri.stream.TripleClientStream; import org.apache.dubbo.rpc.protocol.tri.transport.TripleWriteQueue; import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; import io.netty.channel.Channel; @@ -43,6 +51,7 @@ import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAI import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_STREAM_LISTENER; public class TripleClientCall implements ClientCall, ClientStream.Listener { + private static final Object EMPTY = new Object(); private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(TripleClientCall.class); private final AbstractConnectionClient connectionClient; private final Executor executor; @@ -52,10 +61,12 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { private ClientStream stream; private ClientCall.Listener listener; private boolean canceled; - private boolean headerSent; private boolean autoRequest = true; private boolean done; private Http2Exception.StreamException streamException; + private volatile boolean sendingHeader = false; + private volatile boolean streamCreated = false; + private final Queue pendingTasks = new ConcurrentLinkedQueue<>(); public TripleClientCall( AbstractConnectionClient connectionClient, @@ -63,7 +74,7 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { FrameworkModel frameworkModel, TripleWriteQueue writeQueue) { this.connectionClient = connectionClient; - this.executor = executor; + this.executor = new SerializingExecutor(executor); this.frameworkModel = frameworkModel; this.writeQueue = writeQueue; } @@ -147,7 +158,7 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { return; } // did not create stream - if (!headerSent) { + if (!streamCreated) { return; } canceled = true; @@ -184,9 +195,70 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { } else if (canceled) { throw new IllegalStateException("Call already canceled"); } - if (!headerSent) { - headerSent = true; - stream.sendHeader(requestMetadata.toHeaders()); + doSendMessage(message); + } + + private void doSendMessage(Object message) { + pendingTasks.offer(() -> sendData(message)); + if (streamCreated) { + executor.execute(this::drainTasks); + } + } + + private void sendHeader() { + if (sendingHeader) { + return; + } + if (!streamCreated) { + sendingHeader = true; + Future sendHeaderFuture = stream.sendHeader(requestMetadata.toHeaders()); + sendHeaderFuture.addListener(f -> { + sendingHeader = false; + if (!f.isSuccess()) { + Throwable cause = f.cause(); + if (cause instanceof Http2NoMoreStreamIdsException) { + // stream id used up and channel will be removed + Channel lastChannel = (Channel) connectionClient.getChannel(true); + if (lastChannel != null && lastChannel.isActive() && lastChannel.isOpen()) { + // already reconnected + start(requestMetadata, listener); + return; + } + // blocking reconnect + executor.execute(() -> { + try { + synchronized (connectionClient) { + Channel channel = (Channel) connectionClient.getChannel(true); + if (channel == null) { + connectionClient.reconnect(); + } + } + start(requestMetadata, listener); + } catch (RemotingException e) { + pendingTasks.clear(); + throw new RpcException(e); + } + }); + } + } else { + streamCreated = true; + executor.execute(this::drainTasks); + } + }); + } + } + + private void drainTasks() { + Runnable r; + while ((r = pendingTasks.poll()) != null) { + r.run(); + } + } + + private void sendData(Object message) { + if (EMPTY == message) { + doHalfClose(); + return; } final byte[] data; try { @@ -220,7 +292,14 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { @Override public void halfClose() { - if (!headerSent) { + pendingTasks.offer(this::doHalfClose); + if (streamCreated) { + executor.execute(this::drainTasks); + } + } + + private void doHalfClose() { + if (!streamCreated) { return; } if (canceled) { @@ -243,7 +322,8 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { this.requestMetadata = metadata; this.listener = responseListener; this.stream = new TripleClientStream( - frameworkModel, executor, (Channel) connectionClient.getChannel(true), this, writeQueue); + frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); + this.sendHeader(); return new ClientCallToObserverAdapter<>(this); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/DataQueueCommand.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/DataQueueCommand.java index a9ef447abb..b0d5483dbc 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/DataQueueCommand.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/DataQueueCommand.java @@ -47,13 +47,13 @@ public class DataQueueCommand extends StreamQueueCommand { @Override public void doSend(ChannelHandlerContext ctx, ChannelPromise promise) { if (data == null) { - ctx.write(new DefaultHttp2DataFrame(endStream), promise); + ctx.writeAndFlush(new DefaultHttp2DataFrame(endStream), promise); } else { ByteBuf buf = ctx.alloc().buffer(); buf.writeByte(compressFlag); buf.writeInt(data.length); buf.writeBytes(data); - ctx.write(new DefaultHttp2DataFrame(buf, endStream), promise); + ctx.writeAndFlush(new DefaultHttp2DataFrame(buf, endStream), promise); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/EndStreamQueueCommand.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/EndStreamQueueCommand.java index 2164e02b63..6f85406dde 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/EndStreamQueueCommand.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/EndStreamQueueCommand.java @@ -34,6 +34,6 @@ public class EndStreamQueueCommand extends StreamQueueCommand { @Override public void doSend(ChannelHandlerContext ctx, ChannelPromise promise) { - ctx.write(new DefaultHttp2DataFrame(true), promise); + ctx.writeAndFlush(new DefaultHttp2DataFrame(true), promise); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/HeaderQueueCommand.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/HeaderQueueCommand.java index e221b4370b..a87f09d916 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/HeaderQueueCommand.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/HeaderQueueCommand.java @@ -55,6 +55,6 @@ public class HeaderQueueCommand extends StreamQueueCommand { @Override public void doSend(ChannelHandlerContext ctx, ChannelPromise promise) { - ctx.write(new DefaultHttp2HeadersFrame(headers, endStream), promise); + ctx.writeAndFlush(new DefaultHttp2HeadersFrame(headers, endStream), promise); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/QueuedCommand.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/QueuedCommand.java index f3fe69a8eb..2996b3b0d1 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/QueuedCommand.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/QueuedCommand.java @@ -40,7 +40,7 @@ public abstract class QueuedCommand { public void run(Channel channel) { if (channel.isActive()) { - channel.write(this).addListener(future -> { + channel.writeAndFlush(this).addListener(future -> { if (future.isSuccess()) { promise.setSuccess(); } else { diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/TextDataQueueCommand.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/TextDataQueueCommand.java index c7da28d42d..bf094a5485 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/TextDataQueueCommand.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/TextDataQueueCommand.java @@ -44,6 +44,6 @@ public class TextDataQueueCommand extends StreamQueueCommand { @Override public void doSend(ChannelHandlerContext ctx, ChannelPromise promise) { ByteBuf buf = ByteBufUtil.writeUtf8(ctx.alloc(), data); - ctx.write(new DefaultHttp2DataFrame(buf, endStream), promise); + ctx.writeAndFlush(new DefaultHttp2DataFrame(buf, endStream), promise); } } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java index 8b060ae82d..2cf7a62559 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java @@ -16,6 +16,8 @@ */ package org.apache.dubbo.rpc.protocol.tri.stream; +import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException; + import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -102,12 +104,11 @@ public class TripleClientStream extends AbstractStream implements ClientStream { FrameworkModel frameworkModel, Executor executor, Channel parent, - ClientStream.Listener listener, - TripleWriteQueue writeQueue) { + ClientStream.Listener listener) { super(executor, frameworkModel); this.parent = parent; this.listener = listener; - this.writeQueue = writeQueue; + this.writeQueue = new TripleWriteQueue(); this.streamChannelFuture = initHttp2StreamChannel(parent); } @@ -138,7 +139,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream { } final HeaderQueueCommand headerCmd = HeaderQueueCommand.createHeaders(streamChannelFuture, headers); return writeQueue.enqueueFuture(headerCmd, parent.eventLoop()).addListener(future -> { - if (!future.isSuccess()) { + if (!future.isSuccess() && !(future.cause() instanceof Http2NoMoreStreamIdsException)) { transportException(future.cause()); } }); From 082b4c5af78996c9c5792668c618d5cef8721b42 Mon Sep 17 00:00:00 2001 From: icodening Date: Sat, 18 May 2024 22:03:49 +0800 Subject: [PATCH 2/5] Fix the issue of errors when running out of HTTP2 stream IDs --- .../dubbo/rpc/protocol/tri/call/TripleClientCall.java | 5 ++--- .../rpc/protocol/tri/stream/TripleClientStream.java | 9 ++------- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java index cba3b56cdf..6ac4964d76 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java @@ -16,9 +16,6 @@ */ package org.apache.dubbo.rpc.protocol.tri.call; -import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException; -import io.netty.util.concurrent.Future; - import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.stream.StreamObserver; @@ -44,6 +41,8 @@ import java.util.concurrent.Executor; import io.netty.channel.Channel; import io.netty.handler.codec.http2.Http2Exception; +import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException; +import io.netty.util.concurrent.Future; import static io.netty.handler.codec.http2.Http2Error.FLOW_CONTROL_ERROR; import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_RESPONSE; diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java index 2cf7a62559..65a7bfe8a3 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java @@ -16,8 +16,6 @@ */ package org.apache.dubbo.rpc.protocol.tri.stream; -import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException; - import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; @@ -61,6 +59,7 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.http2.Http2Error; import io.netty.handler.codec.http2.Http2Headers; +import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException; import io.netty.handler.codec.http2.Http2StreamChannel; import io.netty.handler.codec.http2.Http2StreamChannelBootstrap; import io.netty.util.ReferenceCountUtil; @@ -100,11 +99,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream { this.streamChannelFuture = initHttp2StreamChannel(http2StreamChannel); } - public TripleClientStream( - FrameworkModel frameworkModel, - Executor executor, - Channel parent, - ClientStream.Listener listener) { + public TripleClientStream(FrameworkModel frameworkModel, Executor executor, Channel parent, ClientStream.Listener listener) { super(executor, frameworkModel); this.parent = parent; this.listener = listener; From 5b9a9e819a2d19bd42304cf734bbbc86bf636cba Mon Sep 17 00:00:00 2001 From: icodening Date: Sat, 18 May 2024 22:06:55 +0800 Subject: [PATCH 3/5] Fix the issue of errors when running out of HTTP2 stream IDs --- .../apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java index 6ac4964d76..fe7cc9a9bb 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java @@ -320,8 +320,8 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { public StreamObserver start(RequestMetadata metadata, ClientCall.Listener responseListener) { this.requestMetadata = metadata; this.listener = responseListener; - this.stream = new TripleClientStream( - frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); + this.stream = + new TripleClientStream(frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); this.sendHeader(); return new ClientCallToObserverAdapter<>(this); } From 88295cacdc56e6caca9263d78c5c7aff1bbaf506 Mon Sep 17 00:00:00 2001 From: icodening Date: Sat, 18 May 2024 22:15:32 +0800 Subject: [PATCH 4/5] Fix the issue of errors when running out of HTTP2 stream IDs --- .../apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java index fe7cc9a9bb..8510de4b3a 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java @@ -320,8 +320,7 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { public StreamObserver start(RequestMetadata metadata, ClientCall.Listener responseListener) { this.requestMetadata = metadata; this.listener = responseListener; - this.stream = - new TripleClientStream(frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); + this.stream = new TripleClientStream(frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); this.sendHeader(); return new ClientCallToObserverAdapter<>(this); } From c3a5d71d8e16909ddfd53cb627915b6c71533988 Mon Sep 17 00:00:00 2001 From: icodening Date: Sat, 18 May 2024 22:19:48 +0800 Subject: [PATCH 5/5] Fix the issue of errors when running out of HTTP2 stream IDs --- .../apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java | 3 ++- .../dubbo/rpc/protocol/tri/stream/TripleClientStream.java | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java index 8510de4b3a..fe7cc9a9bb 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java @@ -320,7 +320,8 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener { public StreamObserver start(RequestMetadata metadata, ClientCall.Listener responseListener) { this.requestMetadata = metadata; this.listener = responseListener; - this.stream = new TripleClientStream(frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); + this.stream = + new TripleClientStream(frameworkModel, executor, (Channel) connectionClient.getChannel(true), this); this.sendHeader(); return new ClientCallToObserverAdapter<>(this); } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java index 65a7bfe8a3..70a64fa84c 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java @@ -99,7 +99,8 @@ public class TripleClientStream extends AbstractStream implements ClientStream { this.streamChannelFuture = initHttp2StreamChannel(http2StreamChannel); } - public TripleClientStream(FrameworkModel frameworkModel, Executor executor, Channel parent, ClientStream.Listener listener) { + public TripleClientStream( + FrameworkModel frameworkModel, Executor executor, Channel parent, ClientStream.Listener listener) { super(executor, frameworkModel); this.parent = parent; this.listener = listener;