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

tri create Http2StreamChannel async (#11698) 

* tri create Http2StreamChannel async

* tri create Http2StreamChannel async

* tri create Http2StreamChannel async

---------

Co-authored-by: earthchen <earthchen1996@gmail.com>
This commit is contained in:
icodening 2023-03-03 16:38:11 +08:00 committed by GitHub
parent 48bc7165f7
commit cd3ef2a27f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 300 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
View File

@ -76,7 +76,7 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
private final Set<Invoker<?>> invokers;
private final ExecutorService streamExecutor;
private final String acceptEncodings;
private final TripleWriteQueue writeQueue = new TripleWriteQueue();
private final TripleWriteQueue writeQueue = new TripleWriteQueue(256);
public TripleInvoker(Class<T> serviceType,
URL url,

10
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/CancelQueueCommand.java
View File

@ -21,16 +21,18 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.DefaultHttp2ResetFrame;
import io.netty.handler.codec.http2.Http2Error;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public class CancelQueueCommand extends QueuedCommand {
public class CancelQueueCommand extends StreamQueueCommand {
private final Http2Error error;
public CancelQueueCommand(Http2Error error) {
public CancelQueueCommand(TripleStreamChannelFuture streamChannelFuture, Http2Error error) {
super(streamChannelFuture);
this.error = error;
}
public static CancelQueueCommand createCommand(Http2Error error) {
return new CancelQueueCommand(error);
public static CancelQueueCommand createCommand(TripleStreamChannelFuture streamChannelFuture, Http2Error error) {
return new CancelQueueCommand(streamChannelFuture, error);
}

61
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/CreateStreamQueueCommand.java Normal file
View File

@ -0,0 +1,61 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri.command;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
import io.netty.util.concurrent.Future;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public class CreateStreamQueueCommand extends QueuedCommand {
private final Http2StreamChannelBootstrap bootstrap;
private final TripleStreamChannelFuture streamChannelFuture;
private CreateStreamQueueCommand(Http2StreamChannelBootstrap bootstrap,
TripleStreamChannelFuture streamChannelFuture) {
this.bootstrap = bootstrap;
this.streamChannelFuture = streamChannelFuture;
this.promise(streamChannelFuture.getParentChannel().newPromise());
this.channel(streamChannelFuture.getParentChannel());
}
public static CreateStreamQueueCommand create(Http2StreamChannelBootstrap bootstrap,
TripleStreamChannelFuture streamChannelFuture) {
return new CreateStreamQueueCommand(bootstrap, streamChannelFuture);
}
@Override
public void doSend(ChannelHandlerContext ctx, ChannelPromise promise) {
//NOOP
}
@Override
public void run(Channel channel) {
//work in I/O thread
Future<Http2StreamChannel> future = bootstrap.open();
if (future.isSuccess()) {
streamChannelFuture.complete(future.getNow());
} else {
streamChannelFuture.completeExceptionally(future.cause());
}
}
}

12
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/DataQueueCommand.java
View File

@ -21,8 +21,9 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public class DataQueueCommand extends QueuedCommand {
public class DataQueueCommand extends StreamQueueCommand {
private final byte[] data;
@ -30,15 +31,16 @@ public class DataQueueCommand extends QueuedCommand {
private final boolean endStream;
private DataQueueCommand(byte[] data, int compressFlag, boolean endStream) {
private DataQueueCommand(TripleStreamChannelFuture streamChannelFuture, byte[] data, int compressFlag, boolean endStream) {
super(streamChannelFuture);
this.data = data;
this.compressFlag = compressFlag;
this.endStream = endStream;
}
public static DataQueueCommand createGrpcCommand(byte[] data, boolean endStream,
int compressFlag) {
return new DataQueueCommand(data, compressFlag, endStream);
public static DataQueueCommand create(TripleStreamChannelFuture streamChannelFuture, byte[] data, boolean endStream,
int compressFlag) {
return new DataQueueCommand(streamChannelFuture, data, compressFlag, endStream);
}
@Override

11
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/EndStreamQueueCommand.java
View File

@ -20,11 +20,16 @@ package org.apache.dubbo.rpc.protocol.tri.command;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public class EndStreamQueueCommand extends QueuedCommand {
public class EndStreamQueueCommand extends StreamQueueCommand {
public static EndStreamQueueCommand create() {
return new EndStreamQueueCommand();
public EndStreamQueueCommand(TripleStreamChannelFuture streamChannelFuture) {
super(streamChannelFuture);
}
public static EndStreamQueueCommand create(TripleStreamChannelFuture streamChannelFuture) {
return new EndStreamQueueCommand(streamChannelFuture);
}
@Override

14
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/HeaderQueueCommand.java
View File

@ -21,24 +21,26 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
import io.netty.handler.codec.http2.Http2Headers;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public class HeaderQueueCommand extends QueuedCommand {
public class HeaderQueueCommand extends StreamQueueCommand {
private final Http2Headers headers;
private final boolean endStream;
private HeaderQueueCommand(Http2Headers headers, boolean endStream) {
private HeaderQueueCommand(TripleStreamChannelFuture streamChannelFuture, Http2Headers headers, boolean endStream) {
super(streamChannelFuture);
this.headers = headers;
this.endStream = endStream;
}
public static HeaderQueueCommand createHeaders(Http2Headers headers) {
return new HeaderQueueCommand(headers, false);
public static HeaderQueueCommand createHeaders(TripleStreamChannelFuture streamChannelFuture, Http2Headers headers) {
return new HeaderQueueCommand(streamChannelFuture, headers, false);
}
public static HeaderQueueCommand createHeaders(Http2Headers headers, boolean endStream) {
return new HeaderQueueCommand(headers, endStream);
public static HeaderQueueCommand createHeaders(TripleStreamChannelFuture streamChannelFuture, Http2Headers headers, boolean endStream) {
return new HeaderQueueCommand(streamChannelFuture, headers, endStream);
}
public Http2Headers getHeaders() {

8
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/QueuedCommand.java
View File

@ -41,7 +41,13 @@ public abstract class QueuedCommand {
public void run(Channel channel) {
if (channel.isActive()) {
channel.write(this, promise);
channel.write(this).addListener(future -> {
if (future.isSuccess()) {
promise.setSuccess();
} else {
promise.setFailure(future.cause());
}
});
} else {
promise.trySuccess();
}

47
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/StreamQueueCommand.java Normal file
View File

@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri.command;
import io.netty.channel.Channel;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public abstract class StreamQueueCommand extends QueuedCommand {
protected final TripleStreamChannelFuture streamChannelFuture;
protected StreamQueueCommand(TripleStreamChannelFuture streamChannelFuture) {
Assert.notNull(streamChannelFuture, "streamChannelFuture cannot be null.");
this.streamChannelFuture = streamChannelFuture;
this.promise(streamChannelFuture.getParentChannel().newPromise());
}
@Override
public void run(Channel channel) {
if (streamChannelFuture.isSuccess()) {
super.run(channel);
return;
}
promise().setFailure(streamChannelFuture.cause());
}
@Override
public Channel channel() {
return this.streamChannelFuture.getNow();
}
}

10
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/command/TextDataQueueCommand.java
View File

@ -22,20 +22,22 @@ import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
public class TextDataQueueCommand extends QueuedCommand {
public class TextDataQueueCommand extends StreamQueueCommand {
private final String data;
private final boolean endStream;
private TextDataQueueCommand(String text, boolean endStream) {
private TextDataQueueCommand(TripleStreamChannelFuture streamChannelFuture, String text, boolean endStream) {
super(streamChannelFuture);
this.data = text;
this.endStream = endStream;
}
public static TextDataQueueCommand createCommand(String data, boolean endStream) {
return new TextDataQueueCommand(data, endStream);
public static TextDataQueueCommand createCommand(TripleStreamChannelFuture streamChannelFuture, String data, boolean endStream) {
return new TextDataQueueCommand(streamChannelFuture, data, endStream);
}
@Override

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

@ -25,6 +25,7 @@ import org.apache.dubbo.rpc.protocol.tri.ClassLoadUtil;
import org.apache.dubbo.rpc.protocol.tri.ExceptionUtils;
import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum;
import org.apache.dubbo.rpc.protocol.tri.command.CancelQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.CreateStreamQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.DataQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.EndStreamQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.HeaderQueueCommand;
@ -53,7 +54,6 @@ import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import java.io.IOException;
import java.net.SocketAddress;
@ -79,7 +79,8 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
private final TripleWriteQueue writeQueue;
private Deframer deframer;
private final Channel parent;
private final Http2StreamChannel http2StreamChannel;
private final TripleStreamChannelFuture streamChannelFuture;
private boolean halfClosed;
private boolean rst;
// for test
@ -89,10 +90,10 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
ClientStream.Listener listener,
Http2StreamChannel http2StreamChannel) {
super(executor, frameworkModel);
this.parent = null;
this.parent = http2StreamChannel.parent();
this.listener = listener;
this.writeQueue = writeQueue;
this.http2StreamChannel = http2StreamChannel;
this.streamChannelFuture = initHttp2StreamChannel(http2StreamChannel);
}
public TripleClientStream(FrameworkModel frameworkModel,
@ -104,12 +105,13 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
this.parent = parent;
this.listener = listener;
this.writeQueue = writeQueue;
this.http2StreamChannel = initHttp2StreamChannel(parent);
this.streamChannelFuture = initHttp2StreamChannel(parent);
}
private Http2StreamChannel initHttp2StreamChannel(Channel parent) {
private TripleStreamChannelFuture initHttp2StreamChannel(Channel parent) {
TripleStreamChannelFuture streamChannelFuture = new TripleStreamChannelFuture(parent);
Http2StreamChannelBootstrap bootstrap = new Http2StreamChannelBootstrap(parent);
Future<Http2StreamChannel> future = bootstrap.handler(new ChannelInboundHandlerAdapter() {
bootstrap.handler(new ChannelInboundHandlerAdapter() {
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
Channel channel = ctx.channel();
@ -117,11 +119,10 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
channel.pipeline().addLast(new TripleHttp2ClientResponseHandler(createTransportListener()));
channel.closeFuture().addListener(f -> transportException(f.cause()));
}
}).open().syncUninterruptibly();
if (!future.isSuccess()) {
throw new IllegalStateException("Create remote stream failed. channel:" + parent);
}
return future.getNow();
});
CreateStreamQueueCommand cmd = CreateStreamQueueCommand.create(bootstrap, streamChannelFuture);
this.writeQueue.enqueue(cmd);
return streamChannelFuture;
}
@ -134,9 +135,8 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
if (!checkResult.isSuccess()) {
return checkResult;
}
final HeaderQueueCommand headerCmd = HeaderQueueCommand.createHeaders(headers);
headerCmd.channel(http2StreamChannel);
return writeQueue.enqueue(headerCmd).addListener(future -> {
final HeaderQueueCommand headerCmd = HeaderQueueCommand.createHeaders(streamChannelFuture, headers);
return writeQueue.enqueueFuture(headerCmd, parent.eventLoop()).addListener(future -> {
if (!future.isSuccess()) {
transportException(future.cause());
}
@ -154,10 +154,10 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
if (!checkResult.isSuccess()) {
return checkResult;
}
final CancelQueueCommand cmd = CancelQueueCommand.createCommand(Http2Error.CANCEL);
cmd.channel(http2StreamChannel);
final CancelQueueCommand cmd = CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.CANCEL);
TripleClientStream.this.rst = true;
return this.writeQueue.enqueue(cmd, true);
return this.writeQueue.enqueue(cmd);
}
@Override
@ -172,11 +172,9 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
if (!checkResult.isSuccess()) {
return checkResult;
}
final DataQueueCommand cmd = DataQueueCommand.createGrpcCommand(message, false,
final DataQueueCommand cmd = DataQueueCommand.create(streamChannelFuture, message, false,
compressFlag);
cmd.channel(http2StreamChannel);
return this.writeQueue.enqueue(cmd)
.addListener(future -> {
return this.writeQueue.enqueueFuture(cmd, parent.eventLoop()).addListener(future -> {
if (!future.isSuccess()) {
cancelByLocal(
TriRpcStatus.INTERNAL.withDescription("Client write message failed")
@ -199,19 +197,19 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
if (!checkResult.isSuccess()) {
return checkResult;
}
final EndStreamQueueCommand cmd = EndStreamQueueCommand.create();
cmd.channel(http2StreamChannel);
return this.writeQueue.enqueue(cmd);
final EndStreamQueueCommand cmd = EndStreamQueueCommand.create(streamChannelFuture);
return this.writeQueue.enqueueFuture(cmd, parent.eventLoop()).addListener(future -> {
if (future.isSuccess()) {
halfClosed = true;
}
});
}
private ChannelFuture preCheck() {
if (!http2StreamChannel.isActive()) {
return http2StreamChannel.newFailedFuture(new IOException("stream channel is closed"));
}
if (rst) {
return http2StreamChannel.newFailedFuture(new IOException("stream channel has reset"));
return streamChannelFuture.getNow().newFailedFuture(new IOException("stream channel has reset"));
}
return http2StreamChannel.newSucceededFuture();
return parent.newSucceededFuture();
}
/**
@ -226,22 +224,16 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
private TriRpcStatus transportError;
private DeCompressor decompressor;
private boolean halfClosed;
private boolean headerReceived;
private Http2Headers trailers;
void handleH2TransportError(TriRpcStatus status) {
writeQueue.enqueue(CancelQueueCommand.createCommand(Http2Error.NO_ERROR).channel(http2StreamChannel));
writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.NO_ERROR));
TripleClientStream.this.rst = true;
finishProcess(status, null);
}
void finishProcess(TriRpcStatus status, Http2Headers trailers) {
if (halfClosed) {
return;
}
halfClosed = true;
final Map<String, String> reserved = filterReservedHeaders(trailers);
final Map<String, Object> attachments = headersToMap(trailers, () -> {
return reserved.get(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader());
@ -432,9 +424,9 @@ public class TripleClientStream extends AbstractStream implements ClientStream {
executor.execute(() -> {
if (endStream) {
if (!halfClosed) {
if (http2StreamChannel.isActive() && !rst) {
writeQueue.enqueue(CancelQueueCommand.createCommand(Http2Error.CANCEL).channel(http2StreamChannel),
true);
Http2StreamChannel channel = streamChannelFuture.getNow();
if (channel.isActive() && !rst) {
writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.CANCEL));
rst = true;
}
}

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

@ -85,6 +85,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
private Deframer deframer;
private boolean rst = false;
private final Http2StreamChannel http2StreamChannel;
private final TripleStreamChannelFuture tripleStreamChannelFuture;
public TripleServerStream(Http2StreamChannel channel,
FrameworkModel frameworkModel,
@ -100,6 +101,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
this.writeQueue = writeQueue;
this.remoteAddress = (InetSocketAddress) channel.remoteAddress();
this.http2StreamChannel = channel;
this.tripleStreamChannelFuture = new TripleStreamChannelFuture(channel);
}
@Override
@ -118,7 +120,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
return checkResult;
}
this.rst = true;
return writeQueue.enqueue(CancelQueueCommand.createCommand(cause).channel(http2StreamChannel));
return writeQueue.enqueue(CancelQueueCommand.createCommand(tripleStreamChannelFuture, cause));
}
@Override
@ -137,7 +139,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
return checkResult;
}
headerSent = true;
return writeQueue.enqueue(HeaderQueueCommand.createHeaders(headers, false).channel(http2StreamChannel))
return writeQueue.enqueue(HeaderQueueCommand.createHeaders(tripleStreamChannelFuture, headers, false))
.addListener(f -> {
if (!f.isSuccess()) {
reset(Http2Error.INTERNAL_ERROR);
@ -173,7 +175,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
}
headerSent = true;
trailersSent = true;
return writeQueue.enqueue(HeaderQueueCommand.createHeaders(trailers, true).channel(http2StreamChannel))
return writeQueue.enqueue(HeaderQueueCommand.createHeaders(tripleStreamChannelFuture, trailers, true))
.addListener(f -> {
if (!f.isSuccess()) {
reset(Http2Error.INTERNAL_ERROR);
@ -245,7 +247,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
if (!checkResult.isSuccess()) {
return checkResult;
}
return writeQueue.enqueue(DataQueueCommand.createGrpcCommand(message, false, compressFlag).channel(http2StreamChannel));
return writeQueue.enqueue(DataQueueCommand.create(tripleStreamChannelFuture, message, false, compressFlag));
}
/**
@ -263,8 +265,8 @@ public class TripleServerStream extends AbstractStream implements ServerStream {
.setInt(TripleHeaderEnum.STATUS_KEY.getHeader(), status.code.code)
.set(TripleHeaderEnum.MESSAGE_KEY.getHeader(), status.description)
.set(TripleHeaderEnum.CONTENT_TYPE_KEY.getHeader(), TripleConstant.TEXT_PLAIN_UTF8);
writeQueue.enqueue(HeaderQueueCommand.createHeaders( headers, false).channel(http2StreamChannel));
writeQueue.enqueue(TextDataQueueCommand.createCommand(status.description, true).channel(http2StreamChannel));
writeQueue.enqueue(HeaderQueueCommand.createHeaders(tripleStreamChannelFuture, headers, false));
writeQueue.enqueue(TextDataQueueCommand.createCommand(tripleStreamChannelFuture, status.description, true));
}
/**

66
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleStreamChannelFuture.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri.stream;
import io.netty.channel.Channel;
import io.netty.handler.codec.http2.Http2StreamChannel;
import org.apache.dubbo.common.utils.Assert;
import java.util.concurrent.CompletableFuture;
public class TripleStreamChannelFuture extends CompletableFuture<Http2StreamChannel> {
private final Channel parentChannel;
private Throwable cause;
public TripleStreamChannelFuture(Channel parentChannel) {
Assert.notNull(parentChannel, "parentChannel cannot be null.");
this.parentChannel = parentChannel;
}
public TripleStreamChannelFuture(Http2StreamChannel channel) {
this.complete(channel);
this.parentChannel = channel.parent();
}
public Channel getParentChannel() {
return parentChannel;
}
@Override
public boolean completeExceptionally(Throwable cause) {
boolean result = super.completeExceptionally(cause);
if (result) {
this.cause = cause;
}
return result;
}
public Throwable cause() {
return cause;
}
public boolean isSuccess() {
return isDone() && cause() == null;
}
public Http2StreamChannel getNow() {
return getNow(null);
}
}

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

@ -23,20 +23,33 @@ import io.netty.channel.ChannelPromise;
import org.apache.dubbo.common.BatchExecutorQueue;
import org.apache.dubbo.rpc.protocol.tri.command.QueuedCommand;
import java.util.concurrent.Executor;
public class TripleWriteQueue extends BatchExecutorQueue<QueuedCommand> {
public TripleWriteQueue() {
}
public TripleWriteQueue(int chunkSize) {
super(chunkSize);
}
public ChannelFuture enqueue(QueuedCommand command, boolean rst) {
return enqueue(command);
}
public ChannelFuture enqueue(QueuedCommand command) {
return this.enqueueFuture(command, command.channel().eventLoop());
}
public ChannelFuture enqueueFuture(QueuedCommand command, Executor executor) {
ChannelPromise promise = command.promise();
if (promise == null) {
Channel ch = command.channel();
promise = ch.newPromise();
command.promise(promise);
}
super.enqueue(command, command.channel().eventLoop());
super.enqueue(command, executor);
return promise;
}

22
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStreamTest.java
View File

@ -29,6 +29,7 @@ import org.apache.dubbo.rpc.protocol.tri.RequestMetadata;
import org.apache.dubbo.rpc.protocol.tri.TripleConstant;
import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum;
import org.apache.dubbo.rpc.protocol.tri.command.CancelQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.CreateStreamQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.DataQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.EndStreamQueueCommand;
import org.apache.dubbo.rpc.protocol.tri.command.HeaderQueueCommand;
@ -48,6 +49,8 @@ import io.netty.util.concurrent.ImmediateEventExecutor;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.concurrent.Executor;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.mock;
@ -70,13 +73,16 @@ class TripleClientStreamTest {
MockClientStreamListener listener = new MockClientStreamListener();
TripleWriteQueue writeQueue = mock(TripleWriteQueue.class);
final EmbeddedChannel channel = new EmbeddedChannel();
when(writeQueue.enqueue(any())).thenReturn(channel.newPromise());
when(writeQueue.enqueueFuture(any(QueuedCommand.class), any(Executor.class))).thenReturn(channel.newPromise());
Http2StreamChannel http2StreamChannel = mock(Http2StreamChannel.class);
when(http2StreamChannel.isActive()).thenReturn(true);
when(http2StreamChannel.newSucceededFuture()).thenReturn(channel.newSucceededFuture());
when(http2StreamChannel.eventLoop()).thenReturn(new NioEventLoopGroup().next());
when(http2StreamChannel.newPromise()).thenReturn(channel.newPromise());
when(http2StreamChannel.parent()).thenReturn(channel);
TripleClientStream stream = new TripleClientStream(url.getOrDefaultFrameworkModel(),
ImmediateEventExecutor.INSTANCE, writeQueue, listener, http2StreamChannel);
verify(writeQueue).enqueue(any(CreateStreamQueueCommand.class));
final RequestMetadata requestMetadata = new RequestMetadata();
requestMetadata.method = methodDescriptor;
@ -88,19 +94,19 @@ class TripleClientStreamTest {
requestMetadata.group = url.getGroup();
requestMetadata.version = url.getVersion();
stream.sendHeader(requestMetadata.toHeaders());
verify(writeQueue).enqueue(any(HeaderQueueCommand.class));
verify(writeQueue).enqueueFuture(any(HeaderQueueCommand.class), any(Executor.class));
// no other commands
verify(writeQueue).enqueue(any(QueuedCommand.class));
stream.sendMessage(new byte[0], 0, false);
verify(writeQueue).enqueue(any(DataQueueCommand.class));
verify(writeQueue, times(2)).enqueue(any(QueuedCommand.class));
verify(writeQueue).enqueueFuture(any(DataQueueCommand.class), any(Executor.class));
verify(writeQueue, times(2)).enqueueFuture(any(QueuedCommand.class), any(Executor.class));
stream.halfClose();
verify(writeQueue).enqueue(any(EndStreamQueueCommand.class));
verify(writeQueue, times(3)).enqueue(any(QueuedCommand.class));
verify(writeQueue).enqueueFuture(any(EndStreamQueueCommand.class), any(Executor.class));
verify(writeQueue, times(3)).enqueueFuture(any(QueuedCommand.class), any(Executor.class));
stream.cancelByLocal(TriRpcStatus.CANCELLED);
verify(writeQueue, times(1)).enqueue(any(CancelQueueCommand.class), anyBoolean());
verify(writeQueue, times(3)).enqueue(any(QueuedCommand.class));
verify(writeQueue, times(1)).enqueue(any(CancelQueueCommand.class));
verify(writeQueue, times(3)).enqueueFuture(any(QueuedCommand.class), any(Executor.class));
H2TransportListener transportListener = stream.createTransportListener();
DefaultHttp2Headers headers = new DefaultHttp2Headers();

21
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/transport/WriteQueueTest.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.dubbo.rpc.protocol.tri.transport;
import io.netty.channel.embedded.EmbeddedChannel;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.TriRpcStatus;
import org.apache.dubbo.rpc.protocol.tri.command.CancelQueueCommand;
@ -30,8 +31,10 @@ import io.netty.channel.DefaultEventLoop;
import io.netty.channel.EventLoop;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.Http2Error;
import org.apache.dubbo.rpc.protocol.tri.stream.TripleStreamChannelFuture;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
@ -72,16 +75,19 @@ class WriteQueueTest {
}
@Test
@Disabled
void test() throws Exception {
WriteQueue writeQueue = new WriteQueue();
writeQueue.enqueue(HeaderQueueCommand.createHeaders(new DefaultHttp2Headers()).channel(channel));
writeQueue.enqueue(DataQueueCommand.createGrpcCommand(new byte[0], false, 0).channel(channel));
EmbeddedChannel embeddedChannel = new EmbeddedChannel();
TripleStreamChannelFuture tripleStreamChannelFuture = new TripleStreamChannelFuture(embeddedChannel);
writeQueue.enqueue(HeaderQueueCommand.createHeaders(tripleStreamChannelFuture, new DefaultHttp2Headers()).channel(channel));
writeQueue.enqueue(DataQueueCommand.create(tripleStreamChannelFuture, new byte[0], false, 0).channel(channel));
TriRpcStatus status = TriRpcStatus.UNKNOWN
.withCause(new RpcException())
.withDescription("Encode Response data error");
writeQueue.enqueue(CancelQueueCommand.createCommand(Http2Error.CANCEL).channel(channel));
writeQueue.enqueue(TextDataQueueCommand.createCommand(status.description, true).channel(channel));
writeQueue.enqueue(CancelQueueCommand.createCommand(tripleStreamChannelFuture, Http2Error.CANCEL).channel(channel));
writeQueue.enqueue(TextDataQueueCommand.createCommand(tripleStreamChannelFuture, status.description, true).channel(channel));
while (writeMethodCalledTimes.get() != 4) {
Thread.sleep(50);
@ -99,14 +105,17 @@ class WriteQueueTest {
}
@Test
@Disabled
void testChunk() throws Exception {
WriteQueue writeQueue = new WriteQueue();
// test deque chunk size
EmbeddedChannel embeddedChannel = new EmbeddedChannel();
TripleStreamChannelFuture tripleStreamChannelFuture = new TripleStreamChannelFuture(embeddedChannel);
writeMethodCalledTimes.set(0);
for (int i = 0; i < DEQUE_CHUNK_SIZE; i++) {
writeQueue.enqueue(HeaderQueueCommand.createHeaders(new DefaultHttp2Headers()).channel(channel));
writeQueue.enqueue(HeaderQueueCommand.createHeaders(tripleStreamChannelFuture, new DefaultHttp2Headers()).channel(channel));
}
writeQueue.enqueue(HeaderQueueCommand.createHeaders(new DefaultHttp2Headers()).channel(channel));
writeQueue.enqueue(HeaderQueueCommand.createHeaders(tripleStreamChannelFuture, new DefaultHttp2Headers()).channel(channel));
while (writeMethodCalledTimes.get() != (DEQUE_CHUNK_SIZE + 1)) {
Thread.sleep(50);
}