natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

support catch actual exception (#12446)

This commit is contained in:
icodening 2023-06-07 11:01:33 +08:00 committed by GitHub
parent c8ef10e8a5
commit 00d8a46cd5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 15 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java
View File

@ -120,7 +120,6 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
Channel channel = ctx.channel();
channel.pipeline().addLast(new TripleCommandOutBoundHandler());
channel.pipeline().addLast(new TripleHttp2ClientResponseHandler(createTransportListener()));
channel.closeFuture().addListener(f -> transportException(f.cause()));
}
});
CreateStreamQueueCommand cmd = CreateStreamQueueCommand.create(bootstrap, streamChannelFuture);
@ -149,7 +148,7 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
private void transportException(Throwable cause) {
final TriRpcStatus status = TriRpcStatus.INTERNAL.withDescription("Http2 exception")
.withCause(cause);
listener.onComplete(status, null);
listener.onComplete(status, null, null, false);
}
public ChannelFuture cancelByLocal(TriRpcStatus status) {

18
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleWriteQueue.java
View File

@ -23,6 +23,7 @@ import io.netty.channel.ChannelPromise;
import org.apache.dubbo.common.BatchExecutorQueue;
import org.apache.dubbo.rpc.protocol.tri.command.QueuedCommand;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
public class TripleWriteQueue extends BatchExecutorQueue<QueuedCommand> {
@ -55,13 +56,22 @@ public class TripleWriteQueue extends BatchExecutorQueue<QueuedCommand> {
@Override
protected void prepare(QueuedCommand item) {
item.run(item.channel());
try {
Channel channel = item.channel();
item.run(channel);
} catch (CompletionException e) {
item.promise().tryFailure(e.getCause());
}
}
@Override
protected void flush(QueuedCommand item) {
Channel channel = item.channel();
item.run(channel);
channel.flush();
try {
Channel channel = item.channel();
item.run(channel);
channel.flush();
} catch (CompletionException e) {
item.promise().tryFailure(e.getCause());
}
}
}