This is an automated email from the ASF dual-hosted git repository.
guohao pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new 3cceefa Add test case for TripleProtocolTest and add unit test for
[Unary]ClintStream, AbstractStream (#9236)
3cceefa is described below
commit 3cceefa23434e1a8f337290617cb5594241b3f16
Author: 灼华 <[email protected]>
AuthorDate: Thu Nov 18 22:27:29 2021 +0800
Add test case for TripleProtocolTest and add unit test for
[Unary]ClintStream, AbstractStream (#9236)
* Add test case for TripleProtocolTest and add unit test for
[Unary]ClintStream, AbstractStream
* Add unit test for WriteQueue and [Unary]ServerStream
* Add license
---
.../dubbo/rpc/protocol/tri/AbstractStream.java | 5 +-
.../rpc/protocol/tri/command/DataQueueCommand.java | 15 +
.../dubbo/rpc/protocol/tri/AbstractStreamTest.java | 118 +++++++
.../dubbo/rpc/protocol/tri/ClientStreamTest.java | 367 +++++++++++++++++++++
.../dubbo/rpc/protocol/tri/ServerStreamTest.java | 291 ++++++++++++++++
.../dubbo/rpc/protocol/tri/TripleProtocolTest.java | 56 ++--
.../dubbo/rpc/protocol/tri/WriteQueueTest.java | 113 +++++++
.../dubbo/rpc/protocol/tri/support/IGreeter.java | 4 +
.../rpc/protocol/tri/support/IGreeterImpl.java | 15 +-
.../tri/support/MockAbstractStreamImpl.java | 48 +++
.../{IGreeterImpl.java => MockStreamObserver.java} | 44 ++-
11 files changed, 1038 insertions(+), 38 deletions(-)
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
index 1220ec7..5ba50e9 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
@@ -374,9 +374,9 @@ public abstract class AbstractStream implements Stream {
if (TripleHeaderEnum.containsExcludeAttachments(key)) {
continue;
}
- if (key.endsWith(TripleConstant.GRPC_BIN_SUFFIX) && key.length() >
4) {
+ if (key.endsWith(TripleConstant.GRPC_BIN_SUFFIX) && key.length() >
TripleConstant.GRPC_BIN_SUFFIX.length()) {
try {
- attachments.put(key.substring(0, key.length() - 4),
decodeASCIIByte(header.getValue()));
+ attachments.put(key.substring(0, key.length() -
TripleConstant.GRPC_BIN_SUFFIX.length()), decodeASCIIByte(header.getValue()));
} catch (Exception e) {
LOGGER.error("Failed to parse response attachment key=" +
key, e);
}
@@ -446,7 +446,6 @@ public abstract class AbstractStream implements Stream {
protected <T> T unpack(InputStream is, Class<T> clz) {
try {
final T req = SingleProtobufUtils.deserialize(is, clz);
- is.close();
return req;
} catch (IOException e) {
throw new RuntimeException("Failed to unpack req", e);
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 bc98842..c1bed7b 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
@@ -81,4 +81,19 @@ public class DataQueueCommand extends
QueuedCommand.AbstractQueuedCommand {
}
return 1;
}
+
+ // for test
+ public byte[] getData() {
+ return data;
+ }
+
+ // for test
+ public boolean isEndStream() {
+ return endStream;
+ }
+
+ // for test
+ public boolean isClient() {
+ return client;
+ }
}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/AbstractStreamTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/AbstractStreamTest.java
new file mode 100644
index 0000000..e730a67
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/AbstractStreamTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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;
+
+import com.google.protobuf.ByteString;
+import io.netty.handler.codec.http2.Http2Headers;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.utils.ReflectUtils;
+import org.apache.dubbo.rpc.protocol.tri.support.MockAbstractStreamImpl;
+import org.apache.dubbo.triple.TripleWrapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum.MESSAGE_KEY;
+import static
org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum.STATUS_DETAIL_KEY;
+import static org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum.STATUS_KEY;
+
+/**
+ * {@link AbstractStream}
+ */
+public class AbstractStreamTest {
+
+ private URL url = URL.valueOf("test://127.0.0.1/test");
+ private AbstractStream stream = new MockAbstractStreamImpl(url);
+
+ @Test
+ public void testTransportError() {
+ Exception exception = getException();
+ OutboundTransportObserver transportObserver =
Mockito.mock(OutboundTransportObserver.class);
+ stream.subscribe(transportObserver);
+ GrpcStatus grpcStatus = GrpcStatus
+ .fromCode(GrpcStatus.Code.INTERNAL)
+ .withDescription("TEST")
+ .withCause(exception);
+ Map<String, Object> attachments = new HashMap<>();
+ attachments.put("strKey", "v1");
+ attachments.put("binKey", new byte[]{1});
+
attachments.put(String.valueOf(Http2Headers.PseudoHeaderName.PATH.value()),
"path");
+ attachments.put(CommonConstants.GROUP_KEY, "group");
+
+ stream.transportError(grpcStatus, attachments, false);
+
+ ArgumentCaptor<DefaultMetadata> metadataArgumentCaptor =
ArgumentCaptor.forClass(DefaultMetadata.class);
+ Mockito.verify(transportObserver,
Mockito.times(2)).onMetadata(metadataArgumentCaptor.capture(),
Mockito.anyBoolean());
+
+ DefaultMetadata defaultMetadata = metadataArgumentCaptor.getValue();
+ Assertions.assertEquals(defaultMetadata.get(STATUS_KEY.getHeader()),
String.valueOf(grpcStatus.code.code));
+ Assertions.assertEquals(defaultMetadata.get(MESSAGE_KEY.getHeader()),
grpcStatus.description);
+
Assertions.assertNotNull(defaultMetadata.get(STATUS_DETAIL_KEY.getHeader()));
+
Assertions.assertTrue(defaultMetadata.contains("strKey".toLowerCase(Locale.ROOT)));
+
Assertions.assertTrue(defaultMetadata.contains("binKey".toLowerCase(Locale.ROOT)
+ TripleConstant.GRPC_BIN_SUFFIX));
+
Assertions.assertFalse(defaultMetadata.contains(String.valueOf(Http2Headers.PseudoHeaderName.PATH.value())));
+
Assertions.assertFalse(defaultMetadata.contains(CommonConstants.GROUP_KEY));
+
+ // test parseMetadataToAttachmentMap
+ Map<String, Object> attachmentMap =
stream.parseMetadataToAttachmentMap(defaultMetadata);
+
Assertions.assertTrue(attachmentMap.containsKey("strKey".toLowerCase(Locale.ROOT)));
+
Assertions.assertTrue(attachmentMap.containsKey("binKey".toLowerCase(Locale.ROOT)));
+
+ }
+
+ @Test
+ public void testPackUnPack() {
+ TripleWrapper.TripleRequestWrapper requestWrapper =
TripleWrapper.TripleRequestWrapper.newBuilder()
+ .addArgTypes(ReflectUtils.getDesc(String.class))
+
.addArgs(ByteString.copyFrom("TEST_ARG".getBytes(StandardCharsets.UTF_8)))
+ .setSerializeType(TripleConstant.HESSIAN4)
+ .build();
+
+ byte[] bytes = stream.pack(requestWrapper);
+ TripleWrapper.TripleRequestWrapper unpackedData = stream.unpack(bytes,
TripleWrapper.TripleRequestWrapper.class);
+
+ Assertions.assertEquals(unpackedData.getArgTypes(0),
requestWrapper.getArgTypes(0));
+ Assertions.assertEquals(unpackedData.getArgs(0),
requestWrapper.getArgs(0));
+ Assertions.assertEquals(unpackedData.getArgs(0),
requestWrapper.getArgs(0));
+ Assertions.assertEquals(unpackedData.getSerializeType(),
requestWrapper.getSerializeType());
+ }
+
+ @Test
+ public void testCodec() {
+ String str = "BCN";
+ String base64ASCII =
stream.encodeBase64ASCII(str.getBytes(StandardCharsets.UTF_8));
+ byte[] bytes = stream.decodeASCIIByte(base64ASCII);
+ Assertions.assertEquals(str, new String(bytes,
StandardCharsets.UTF_8));
+ }
+
+ private Exception getException() {
+ Exception exception = null;
+ try {
+ int count = 1 / 0;
+ } catch (Exception e) {
+ exception = e;
+ }
+ return exception;
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamTest.java
new file mode 100644
index 0000000..b6e87bb
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamTest.java
@@ -0,0 +1,367 @@
+/*
+ * 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;
+
+import com.google.protobuf.ByteString;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultEventLoop;
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.util.Attribute;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.serialize.DefaultMultipleSerialization;
+import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.remoting.api.Connection;
+import org.apache.dubbo.remoting.exchange.Request;
+import org.apache.dubbo.remoting.exchange.support.DefaultFuture2;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.ConsumerModel;
+import org.apache.dubbo.rpc.model.ModuleServiceRepository;
+import org.apache.dubbo.rpc.model.ServiceDescriptor;
+import org.apache.dubbo.rpc.model.ServiceMetadata;
+import org.apache.dubbo.rpc.protocol.tri.command.DataQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.HeaderQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.QueuedCommand;
+import org.apache.dubbo.rpc.protocol.tri.support.IGreeter;
+import org.apache.dubbo.rpc.protocol.tri.support.MockStreamObserver;
+import org.apache.dubbo.triple.TripleWrapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.dubbo.rpc.protocol.tri.Compressor.DEFAULT_COMPRESSOR;
+import static org.apache.dubbo.rpc.protocol.tri.TripleConstant.HTTP_SCHEME;
+import static org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum.GRPC_ENCODING;
+
+/**
+ * {@link ClientStream}
+ * {@link UnaryClientStream}
+ * {@link AbstractClientStream}
+ */
+public class ClientStreamTest {
+
+ private URL url;
+ private ConsumerModel consumerModel;
+ private Invoker<IGreeter> invoker;
+ private Connection connection;
+ private int timeout = 100000;
+ private AtomicInteger writeMethodCalledTimes = new AtomicInteger(0);
+
+
+ @BeforeEach
+ public void init() {
+ url = URL.valueOf("tri://127.0.0.1:9103/" + IGreeter.class.getName());
+ ModuleServiceRepository serviceRepository =
ApplicationModel.defaultModel().getDefaultModule().getServiceRepository();
+ ServiceDescriptor serviceDescriptor =
serviceRepository.registerService(IGreeter.class);
+ consumerModel = new ConsumerModel(url.getServiceKey(), null,
serviceDescriptor, null,
+ new ServiceMetadata(), null);
+ url = url.setServiceModel(consumerModel);
+
+ invoker = Mockito.mock(Invoker.class);
+ Mockito.when(invoker.getUrl()).thenReturn(url);
+
+ connection = Mockito.mock(Connection.class);
+
+ writeMethodCalledTimes.set(0);
+ }
+
+ @Test
+ public void testNewClientStream() throws Exception {
+
+ Method echoMethod = IGreeter.class.getDeclaredMethod("echo",
String.class);
+ RpcInvocation rpcInvocation = new RpcInvocation(consumerModel,
echoMethod, IGreeter.class.getName(), url.getProtocolServiceKey(), new
Object[]{"ECHO"});
+ rpcInvocation.setInvoker(invoker);
+
+ Request request = new Request(1);
+ request.setData(rpcInvocation);
+
+ // test UnaryClientStream
+ AbstractClientStream stream =
AbstractClientStream.newClientStream(request, connection);
+
+ Assertions.assertTrue(stream instanceof UnaryClientStream);
+ UnaryClientStream unaryClientStream = (UnaryClientStream) stream;
+ Assertions.assertEquals(unaryClientStream.getConnection(), connection);
+ Assertions.assertEquals(unaryClientStream.getConsumerModel(),
consumerModel);
+ Assertions.assertEquals(unaryClientStream.getMethodName(),
echoMethod.getName());
+ Assertions.assertEquals(unaryClientStream.getUrl(), url);
+ Assertions.assertEquals(unaryClientStream.getRequestId(),
request.getId());
+ Assertions.assertEquals(unaryClientStream.getRpcInvocation(),
rpcInvocation);
+ Assertions.assertEquals(unaryClientStream.getCompressor(),
Compressor.NONE);
+ Assertions.assertEquals(unaryClientStream.getDeCompressor(),
Compressor.NONE);
+ Assertions.assertEquals(stream.getScheme(), HTTP_SCHEME);
+ Assertions.assertEquals(stream.getAcceptEncoding(), "gzip,identity");
+
Assertions.assertFalse(stream.getCancellationContext().getListeners().isEmpty());
+ Assertions.assertTrue(stream.getMultipleSerialization() instanceof
DefaultMultipleSerialization);
+
+ // test ClientStream
+ Method serverStreamMethod =
IGreeter.class.getDeclaredMethod("serverStream", String.class,
StreamObserver.class);
+ rpcInvocation = new RpcInvocation(consumerModel, serverStreamMethod,
IGreeter.class.getName(), url.getProtocolServiceKey(), new Object[]{null,
null});
+ request.setData(rpcInvocation);
+ rpcInvocation.setInvoker(invoker);
+ stream = AbstractClientStream.newClientStream(request, connection);
+ Assertions.assertTrue(stream instanceof ClientStream);
+ }
+
+ @Test
+ public void testStartCall_UnaryClientStream() throws Throwable {
+ // 1. test startCall
+ Method echoMethod = IGreeter.class.getDeclaredMethod("echo",
String.class);
+ RpcInvocation rpcInvocation = new RpcInvocation(consumerModel,
echoMethod, IGreeter.class.getName(),
+ url.getProtocolServiceKey(), new Object[]{"ECHO"});
+ rpcInvocation.setInvoker(invoker);
+ rpcInvocation.setObjectAttachment(Constants.SERIALIZATION_KEY,
TripleConstant.HESSIAN2);
+ rpcInvocation.setObjectAttachment(CommonConstants.PATH_KEY,
url.getPath());
+ rpcInvocation.put(CommonConstants.TIMEOUT_KEY, timeout);
+ Request request = new Request(1);
+ request.setData(rpcInvocation);
+
+ ExecutorService executor = Mockito.mock(ExecutorService.class);
+ DefaultFuture2 future = DefaultFuture2.newFuture(connection, request,
timeout, executor);
+
+ AbstractClientStream stream =
AbstractClientStream.newClientStream(request, connection);
+
+ ChannelPromise promise = Mockito.mock(ChannelPromise.class);
+ Http2StreamChannel streamChannel = getHttp2StreamChannel(stream);
+
+ // startCall
+ WriteQueue writeQueue = new WriteQueue(streamChannel);
+ stream.startCall(writeQueue, promise);
+ // Wait for the asynchronous operation to complete
+ while (writeMethodCalledTimes.get() != 3) {
+ Thread.sleep(50);
+ }
+
+ Assertions.assertNotNull(stream.outboundTransportObserver());
+ ArgumentCaptor<QueuedCommand> commandArgumentCaptor =
ArgumentCaptor.forClass(QueuedCommand.class);
+ ArgumentCaptor<ChannelPromise> promiseArgumentCaptor =
ArgumentCaptor.forClass(ChannelPromise.class);
+ Mockito.verify(streamChannel,
Mockito.times(3)).write(commandArgumentCaptor.capture(),
promiseArgumentCaptor.capture());
+ List<QueuedCommand> queuedCommands =
commandArgumentCaptor.getAllValues();
+
+ HeaderQueueCommand headerQueueCommand = (HeaderQueueCommand)
queuedCommands.get(0);
+ Http2Headers headers = headerQueueCommand.getHeaders();
+
Assertions.assertEquals(headers.get(Http2Headers.PseudoHeaderName.SCHEME.value()),
HTTP_SCHEME);
+
Assertions.assertEquals(headers.get(Http2Headers.PseudoHeaderName.PATH.value()),
"/" + url.getPath() + "/" + rpcInvocation.getMethodName());
+
Assertions.assertEquals(headers.get(Http2Headers.PseudoHeaderName.AUTHORITY.value()),
url.getAddress());
+
Assertions.assertEquals(headers.get(Http2Headers.PseudoHeaderName.METHOD.value()),
HttpMethod.POST.asciiName());
+ Assertions.assertEquals(headers.get(HttpHeaderNames.TE),
HttpHeaderValues.TRAILERS);
+ Assertions.assertEquals(headers.get(GRPC_ENCODING.getHeader()),
DEFAULT_COMPRESSOR);
+
Assertions.assertEquals(headers.get(TripleHeaderEnum.GRPC_ACCEPT_ENCODING.getHeader()),
stream.getAcceptEncoding());
+
Assertions.assertEquals(headers.get(TripleHeaderEnum.CONTENT_TYPE_KEY.getHeader()),
TripleHeaderEnum.CONTENT_PROTO.getHeader());
+
Assertions.assertEquals(headers.get(TripleHeaderEnum.TIMEOUT.getHeader()),
timeout + "m");
+
+ DataQueueCommand dataQueueCommand1 = (DataQueueCommand)
queuedCommands.get(1);
+ Assertions.assertTrue(dataQueueCommand1.getData().length > 0);
+ Assertions.assertFalse(dataQueueCommand1.isEndStream());
+ Assertions.assertTrue(dataQueueCommand1.isClient());
+ TripleWrapper.TripleRequestWrapper requestWrapper =
stream.unpack(dataQueueCommand1.getData(),
TripleWrapper.TripleRequestWrapper.class);
+ ByteArrayInputStream bais = new
ByteArrayInputStream(requestWrapper.getArgs(0).toByteArray());
+ Object ret = stream.getMultipleSerialization().deserialize(url,
stream.getSerializeType(), requestWrapper.getArgTypes(0), bais);
+ bais.close();
+ Assertions.assertEquals(ret.toString(), "ECHO");
+
+ DataQueueCommand dataQueueCommand2 = (DataQueueCommand)
queuedCommands.get(2);
+ Assertions.assertNull(dataQueueCommand2.getData());
+ Assertions.assertTrue(dataQueueCommand2.isEndStream());
+ // 2. Verify the data from the server()
+ // NOTE: The onXX method of inboundTransportObserver is usually
triggered when receiving server data,
+ // here we manually call to simulate the behavior of the [server ->
client]
+ TransportObserver inboundTransportObserver =
stream.inboundTransportObserver();
+ headers = getHttp2Headers(stream);
+ inboundTransportObserver.onMetadata(new Http2HeaderMeta(headers),
false);
+ Object resp = "RESPONSE";
+ byte[] bytes = getPackedData(stream, resp);
+ inboundTransportObserver.onData(bytes, false);
+
inboundTransportObserver.onMetadata(TripleConstant.getSuccessResponseMeta(),
false); // trailers
+ inboundTransportObserver.onComplete();
+ Object result = future.get();
+ Assertions.assertEquals(((AppResponse) result).recreate(), resp);
+ // TODO onError case
+ }
+
+ @Test
+ public void testStartCall_ClientStream_ServerStream() throws Throwable {
+ // 1. test startCall
+ StreamObserver<String> outboundMessageSubscriber = new
MockStreamObserver();
+ Method serverStreamMethod =
IGreeter.class.getDeclaredMethod("serverStream", String.class,
StreamObserver.class);
+ RpcInvocation rpcInvocation = new RpcInvocation(consumerModel,
serverStreamMethod, IGreeter.class.getName(),
+ url.getProtocolServiceKey(), new Object[]{"stringData",
outboundMessageSubscriber});
+ rpcInvocation.setInvoker(invoker);
+ rpcInvocation.setObjectAttachment(Constants.SERIALIZATION_KEY,
TripleConstant.HESSIAN2);
+ Request request = new Request(1);
+ request.setData(rpcInvocation);
+ ExecutorService executor = Mockito.mock(ExecutorService.class);
+ DefaultFuture2 future = DefaultFuture2.newFuture(connection, request,
timeout, executor);
+ AbstractClientStream stream =
AbstractClientStream.newClientStream(request, connection);
+
+ ChannelPromise promise = Mockito.mock(ChannelPromise.class);
+ Http2StreamChannel streamChannel = getHttp2StreamChannel(stream);
+ // startCall
+ WriteQueue writeQueue = new WriteQueue(streamChannel);
+ stream.startCall(writeQueue, promise);
+ // Wait for the asynchronous operation to complete
+ while (writeMethodCalledTimes.get() != 3) {
+ Thread.sleep(50);
+ }
+ Assertions.assertNull(DefaultFuture2.getFuture(request.getId()));
+ Assertions.assertNull(((AppResponse) future.get()).recreate());
+ // NOTE: Send one header frame and two data frames. The previous test
case has been verified, so I won’t go into details here.
+ Mockito.verify(streamChannel, Mockito.times(3)).write(Mockito.any(),
Mockito.any());
+
+ // 2. Verify the data from the server()
+ // NOTE: The onXX method of inboundTransportObserver is usually
triggered when receiving server data,
+ // here we manually call to simulate the behavior of the [server ->
client]
+ TransportObserver inboundTransportObserver =
stream.inboundTransportObserver();
+ DefaultHttp2Headers headers = getHttp2Headers(stream);
+ inboundTransportObserver.onMetadata(new Http2HeaderMeta(headers),
false);
+ Object resp = "RESPONSE";
+ byte[] bytes = getPackedData(stream, resp);
+ inboundTransportObserver.onData(bytes, false);
+
inboundTransportObserver.onMetadata(TripleConstant.SUCCESS_RESPONSE_META,
false);
+ inboundTransportObserver.onComplete();
+ MockStreamObserver observer = (MockStreamObserver)
outboundMessageSubscriber;
+ observer.getLatch().await(1000, TimeUnit.MILLISECONDS); // Wait for
the asynchronous operation to complete
+ Assertions.assertEquals(observer.getOnNextData(), resp);
+ Assertions.assertTrue(observer.isOnCompleted());
+
+ }
+
+
+ @Test
+ public void testStartCall_ClientStream_BidirectionalStream() throws
Throwable {
+ // 1. test startCall
+ StreamObserver<String> outboundMessageSubscriber = new
MockStreamObserver();
+ Method bidirectionalStreamMethod =
IGreeter.class.getDeclaredMethod("bidirectionalStream", StreamObserver.class);
+ RpcInvocation rpcInvocation = new RpcInvocation(consumerModel,
bidirectionalStreamMethod, IGreeter.class.getName(),
+ url.getProtocolServiceKey(), new
Object[]{outboundMessageSubscriber});
+ rpcInvocation.setInvoker(invoker);
+ rpcInvocation.setObjectAttachment(Constants.SERIALIZATION_KEY,
TripleConstant.HESSIAN2);
+ Request request = new Request(1);
+ request.setData(rpcInvocation);
+ ExecutorService executor = Mockito.mock(ExecutorService.class);
+ DefaultFuture2 future = DefaultFuture2.newFuture(connection, request,
timeout, executor);
+ AbstractClientStream stream =
AbstractClientStream.newClientStream(request, connection);
+ ChannelPromise promise = Mockito.mock(ChannelPromise.class);
+ Http2StreamChannel streamChannel = getHttp2StreamChannel(stream);
+ // startCall
+ WriteQueue writeQueue = new WriteQueue(streamChannel);
+ stream.startCall(writeQueue, promise);
+ // Wait for the asynchronous operation to complete
+ while (DefaultFuture2.getFuture(request.getId()) != null) {
+ Thread.sleep(50);
+ }
+ Assertions.assertNull(DefaultFuture2.getFuture(request.getId()));
+ Assertions.assertEquals(stream.outboundMessageSubscriber(),
outboundMessageSubscriber);
+ Assertions.assertEquals(((AppResponse) future.get()).recreate(),
stream.inboundMessageObserver());
+ Mockito.verify(streamChannel, Mockito.times(0)).write(Mockito.any(),
Mockito.any());
+
+ // 2. Verify the data from the server()
+ // Simulate the client to initiate a call
+ StreamObserver<Object> inboundMessageObserver =
stream.inboundMessageObserver();
+ inboundMessageObserver.onNext("TEST");
+ inboundMessageObserver.onCompleted();
+ while (writeMethodCalledTimes.get() != 3) {
+ Thread.sleep(50);
+ }
+ Mockito.verify(streamChannel, Mockito.times(3)).write(Mockito.any(),
Mockito.any());
+
+ // NOTE: The onXX method of inboundTransportObserver is usually
triggered when receiving server data,
+ // here we manually call to simulate the behavior of the [server ->
client]
+ TransportObserver inboundTransportObserver =
stream.inboundTransportObserver();
+ DefaultHttp2Headers headers = getHttp2Headers(stream);
+ inboundTransportObserver.onMetadata(new Http2HeaderMeta(headers),
false);
+ Object resp = "RESPONSE";
+ byte[] bytes = getPackedData(stream, resp);
+ inboundTransportObserver.onData(bytes, false);
+
inboundTransportObserver.onMetadata(TripleConstant.SUCCESS_RESPONSE_META,
false);
+ inboundTransportObserver.onComplete();
+
+ MockStreamObserver observer = (MockStreamObserver)
outboundMessageSubscriber;
+ observer.getLatch().await(1000, TimeUnit.MILLISECONDS);
+ Assertions.assertEquals(observer.getOnNextData(), resp);
+ Assertions.assertTrue(observer.isOnCompleted());
+ }
+
+
+ private Http2StreamChannel getHttp2StreamChannel(AbstractClientStream
stream) {
+ ChannelFuture channelFuture = Mockito.mock(ChannelFuture.class);
+ Http2StreamChannel streamChannel =
Mockito.mock(Http2StreamChannel.class);
+ Attribute<AbstractClientStream> attribute =
Mockito.mock(Attribute.class);
+ ChannelPromise promise = Mockito.mock(ChannelPromise.class);
+ EventLoop eventLoop = new DefaultEventLoop();
+
+ Mockito.when(attribute.get()).thenReturn(stream);
+
Mockito.when(streamChannel.writeAndFlush(Mockito.any())).thenReturn(channelFuture);
+
Mockito.when(streamChannel.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
+
Mockito.when(streamChannel.attr(TripleConstant.CLIENT_STREAM_KEY)).thenReturn(attribute);
+ Mockito.when(streamChannel.eventLoop()).thenReturn(eventLoop);
+ Mockito.when(streamChannel.newPromise()).thenReturn(promise);
+
+ Mockito.when(streamChannel.write(Mockito.any(),
Mockito.any())).thenAnswer(
+ (Answer<ChannelPromise>) invocationOnMock -> {
+ writeMethodCalledTimes.incrementAndGet();
+ return promise;
+ });
+ return streamChannel;
+ }
+
+ private DefaultHttp2Headers getHttp2Headers(AbstractClientStream stream) {
+ DefaultHttp2Headers headers = new DefaultHttp2Headers(true);
+ headers.set(Http2Headers.PseudoHeaderName.STATUS.value(),
HttpResponseStatus.OK.codeAsText());
+ headers.set(HttpHeaderNames.CONTENT_TYPE,
TripleConstant.CONTENT_PROTO);
+ headers.set(TripleHeaderEnum.GRPC_ENCODING.getHeader(),
stream.getCompressor().getMessageEncoding());
+ headers.set(TripleHeaderEnum.GRPC_ACCEPT_ENCODING.getHeader(),
stream.getAcceptEncoding());
+ return headers;
+ }
+
+ private byte[] getPackedData(AbstractClientStream stream, Object resp)
throws IOException {
+ final TripleWrapper.TripleResponseWrapper.Builder builder =
TripleWrapper.TripleResponseWrapper.newBuilder()
+ .setType(stream.getMethodDescriptor().getReturnClass().getName())
+ .setSerializeType(TripleConstant.HESSIAN4);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ stream.getMultipleSerialization().serialize(url,
TripleConstant.HESSIAN2,
stream.getMethodDescriptor().getReturnClass().getName(), resp, bos);
+ builder.setData(ByteString.copyFrom(bos.toByteArray()));
+ bos.close();
+ TripleWrapper.TripleResponseWrapper responseWrapper = builder.build();
+ byte[] bytes = stream.pack(responseWrapper);
+ return bytes;
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamTest.java
new file mode 100644
index 0000000..327fb45
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/ServerStreamTest.java
@@ -0,0 +1,291 @@
+/*
+ * 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;
+
+import com.google.protobuf.ByteString;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultEventLoop;
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.handler.codec.http2.Http2StreamChannel;
+import io.netty.util.Attribute;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.serialize.MultipleSerialization;
+import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.remoting.Constants;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.ProxyFactory;
+import org.apache.dubbo.rpc.filter.TokenHeaderFilter;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.ModuleServiceRepository;
+import org.apache.dubbo.rpc.model.ProviderModel;
+import org.apache.dubbo.rpc.model.ServiceDescriptor;
+import org.apache.dubbo.rpc.model.ServiceMetadata;
+import org.apache.dubbo.rpc.protocol.tri.command.DataQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.HeaderQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.QueuedCommand;
+import org.apache.dubbo.rpc.protocol.tri.support.IGreeter;
+import org.apache.dubbo.rpc.protocol.tri.support.IGreeterImpl;
+import org.apache.dubbo.rpc.protocol.tri.support.MockStreamObserver;
+import org.apache.dubbo.triple.TripleWrapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.dubbo.rpc.protocol.tri.Compressor.DEFAULT_COMPRESSOR;
+
+/**
+ * {@link ServerStream}
+ * {@link UnaryServerStream}
+ * {@link AbstractServerStream}
+ */
+public class ServerStreamTest {
+ private URL url;
+ private ProviderModel providerModel;
+ private Invoker invoker;
+ private IGreeter serviceImpl;
+ private int timeout = 10000;
+ private AtomicInteger writeMethodCalledTimes = new AtomicInteger(0);
+ private final String REQUEST_MSG = "TEST_DATA";
+
+ @BeforeEach
+ public void init() {
+ serviceImpl = new IGreeterImpl();
+ url = URL.valueOf("tri://127.0.0.1:9103/" + IGreeter.class.getName());
+ ModuleServiceRepository serviceRepository =
ApplicationModel.defaultModel().getDefaultModule().getServiceRepository();
+ ServiceDescriptor serviceDescriptor =
serviceRepository.registerService(IGreeter.class);
+ providerModel = new ProviderModel(
+ url.getServiceKey(),
+ serviceImpl,
+ serviceDescriptor,
+ null,
+ new ServiceMetadata());
+ serviceRepository.registerProvider(providerModel);
+ url = url.setServiceModel(providerModel);
+
+ ProxyFactory proxy =
ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
+ invoker = proxy.getInvoker(serviceImpl, IGreeter.class, url);
+
+ writeMethodCalledTimes.set(0);
+ }
+
+ @Test
+ public void testNewServerStream() {
+ AbstractServerStream unaryServerStream =
AbstractServerStream.newServerStream(invoker.getUrl(), true);
+ Assertions.assertTrue(unaryServerStream instanceof UnaryServerStream);
+
+ AbstractServerStream serverStream =
AbstractServerStream.newServerStream(invoker.getUrl(), false);
+ Assertions.assertTrue(serverStream instanceof ServerStream);
+
+ Assertions.assertTrue(unaryServerStream.getHeaderFilters().get(0)
instanceof TokenHeaderFilter);
+ Assertions.assertEquals(unaryServerStream.getSerializeType(),
Constants.DEFAULT_REMOTING_SERIALIZATION);
+ }
+
+
+ @Test
+ public void testUnaryServerStream() throws Exception {
+
+ AbstractServerStream stream =
AbstractServerStream.newServerStream(invoker.getUrl(), true);
+ Http2StreamChannel channel = getHttp2StreamChannel(stream);
+ WriteQueue writeQueue = new WriteQueue(channel);
+ ServerOutboundTransportObserver outboundTransportObserver = new
ServerOutboundTransportObserver(writeQueue);
+ Method echoMethod = IGreeter.class.getDeclaredMethod("echo", new
Class[]{String.class});
+ String methodName = echoMethod.getName();
+ stream.service(providerModel.getServiceModel())
+ .invoker(invoker)
+ .methodName(methodName)
+ .setDeCompressor(Compressor.NONE)
+ .method(providerModel.getServiceModel().getMethod(methodName, new
Class[]{String.class}))
+ .subscribe(outboundTransportObserver);
+
+ Http2Headers headers = getHttp2Headers(methodName);
+
+ final TransportObserver inboundTransportObserver =
stream.inboundTransportObserver();
+ inboundTransportObserver.onMetadata(new Http2HeaderMeta(headers),
false); // 1
+ byte[] data = getPackedData(stream, new Object[]{REQUEST_MSG}, new
Class[]{String.class});
+ inboundTransportObserver.onData(data, false); // 2
+ inboundTransportObserver.onComplete(); // 3
+
+ // Wait for the asynchronous operation to complete
+ while (writeMethodCalledTimes.get() != 3) {
+ Thread.sleep(50);
+ }
+ ArgumentCaptor<QueuedCommand> commandArgumentCaptor =
ArgumentCaptor.forClass(QueuedCommand.class);
+ ArgumentCaptor<ChannelPromise> promiseArgumentCaptor =
ArgumentCaptor.forClass(ChannelPromise.class);
+ Mockito.verify(channel,
Mockito.times(3)).write(commandArgumentCaptor.capture(),
promiseArgumentCaptor.capture());
+ List<QueuedCommand> queuedCommands =
commandArgumentCaptor.getAllValues();
+
+ HeaderQueueCommand headerQueueCommand1 = (HeaderQueueCommand)
queuedCommands.get(0);
+ Http2Headers headers1 = headerQueueCommand1.getHeaders();
+
Assertions.assertEquals(headers1.get(Http2Headers.PseudoHeaderName.STATUS.value()),
HttpResponseStatus.OK.codeAsText());
+ Assertions.assertEquals(headers1.get(HttpHeaderNames.CONTENT_TYPE),
TripleConstant.CONTENT_PROTO);
+
Assertions.assertEquals(headers1.get(TripleHeaderEnum.GRPC_ENCODING.getHeader()),
stream.getCompressor().getMessageEncoding());
+
Assertions.assertEquals(headers1.get(TripleHeaderEnum.GRPC_ACCEPT_ENCODING.getHeader()),
stream.getAcceptEncoding());
+
+ DataQueueCommand dataQueueCommand = (DataQueueCommand)
queuedCommands.get(1);
+ Assertions.assertTrue(dataQueueCommand.getData().length > 0);
+ Assertions.assertFalse(dataQueueCommand.isEndStream());
+ Assertions.assertFalse(dataQueueCommand.isClient());
+ TripleWrapper.TripleResponseWrapper responseWrapper =
stream.unpack(dataQueueCommand.getData(),
TripleWrapper.TripleResponseWrapper.class);
+ ByteArrayInputStream bais = new
ByteArrayInputStream(responseWrapper.getData().toByteArray());
+ Object ret = stream.getMultipleSerialization().deserialize(url,
stream.getSerializeType(), responseWrapper.getType(), bais);
+ bais.close();
+ Assertions.assertEquals(ret.toString(), REQUEST_MSG);
+
+ HeaderQueueCommand headerQueueCommand2 = (HeaderQueueCommand)
queuedCommands.get(2);
+ Assertions.assertTrue(headerQueueCommand2.isEndStream());
+ Http2Headers headers2 = headerQueueCommand2.getHeaders();
+
Assertions.assertEquals(headers2.get(TripleHeaderEnum.MESSAGE_KEY.getHeader()),
TripleConstant.SUCCESS_RESPONSE_MESSAGE);
+
Assertions.assertEquals(headers2.get(TripleHeaderEnum.STATUS_KEY.getHeader()),
TripleConstant.SUCCESS_RESPONSE_STATUS);
+ }
+
+
+ @Test
+ public void testServerStream() throws Exception {
+ AbstractServerStream stream =
AbstractServerStream.newServerStream(invoker.getUrl(), false);
+ Http2StreamChannel channel = getHttp2StreamChannel(stream);
+ WriteQueue writeQueue = new WriteQueue(channel);
+ ServerOutboundTransportObserver outboundTransportObserver = new
ServerOutboundTransportObserver(writeQueue);
+ Method serverStreamMethod =
IGreeter.class.getDeclaredMethod("serverStream", new Class[]{String.class,
StreamObserver.class});
+ String methodName = serverStreamMethod.getName();
+ stream.service(providerModel.getServiceModel())
+ .invoker(invoker)
+ .methodName(methodName)
+ .setDeCompressor(Compressor.NONE)
+ .method(providerModel.getServiceModel().getMethod(methodName, new
Class[]{String.class, StreamObserver.class}))
+ .subscribe(outboundTransportObserver);
+
+
+ final TransportObserver inboundTransportObserver =
stream.inboundTransportObserver();
+ Http2Headers headers = getHttp2Headers(methodName);
+ inboundTransportObserver.onMetadata(new Http2HeaderMeta(headers),
false); // 1
+ byte[] data = getPackedData(stream, new Object[]{REQUEST_MSG}, new
Class[]{String.class});
+ inboundTransportObserver.onData(data, false); // 2
+ inboundTransportObserver.onComplete(); // 3
+
+ // Wait for the asynchronous operation to complete
+ while (writeMethodCalledTimes.get() != 3) {
+ Thread.sleep(50);
+ }
+ // NOTE: Send two header frames and one data frames. The previous test
case has been verified, so I won’t go into details here.
+ Mockito.verify(channel, Mockito.times(3)).write(Mockito.any(),
Mockito.any());
+ }
+
+ @Test
+ public void testServerStream_biDirectional() throws Exception {
+ AbstractServerStream stream =
AbstractServerStream.newServerStream(invoker.getUrl(), false);
+ Http2StreamChannel channel = getHttp2StreamChannel(stream);
+ WriteQueue writeQueue = new WriteQueue(channel);
+ ServerOutboundTransportObserver outboundTransportObserver = new
ServerOutboundTransportObserver(writeQueue);
+ Method bidirectionalStreamMethod =
IGreeter.class.getDeclaredMethod("bidirectionalStream", new
Class[]{StreamObserver.class});
+ String methodName = bidirectionalStreamMethod.getName();
+ stream.service(providerModel.getServiceModel())
+ .invoker(invoker)
+ .methodName(methodName)
+ .setDeCompressor(Compressor.NONE)
+ .method(providerModel.getServiceModel().getMethod(methodName, new
Class[]{String.class}))
+ .subscribe(outboundTransportObserver);
+
+ final TransportObserver inboundTransportObserver =
stream.inboundTransportObserver();
+ Http2Headers headers = getHttp2Headers(methodName);
+ inboundTransportObserver.onMetadata(new Http2HeaderMeta(headers),
false); // 1
+ byte[] data = getPackedData(stream, new Object[]{REQUEST_MSG}, new
Class[]{String.class});
+ inboundTransportObserver.onData(data, false); // 2
+ inboundTransportObserver.onComplete(); // 3
+
+ // Wait for the asynchronous operation to complete
+ while (writeMethodCalledTimes.get() != 3) {
+ Thread.sleep(50);
+ }
+ // NOTE: Send two header frames and one data frames. The previous test
case has been verified, so I won’t go into details here.
+ Mockito.verify(channel, Mockito.times(3)).write(Mockito.any(),
Mockito.any());
+
+ MockStreamObserver serverOutboundMessageSubscriber =
(MockStreamObserver) ((IGreeterImpl) serviceImpl).getMockStreamObserver();
+ serverOutboundMessageSubscriber.getLatch().await(1000,
TimeUnit.MILLISECONDS);
+
Assertions.assertEquals(serverOutboundMessageSubscriber.getOnNextData(),
REQUEST_MSG);
+ Assertions.assertTrue(serverOutboundMessageSubscriber.isOnCompleted());
+ }
+
+ private Http2Headers getHttp2Headers(String methodName) {
+ Http2Headers headers = new DefaultHttp2Headers();
+ headers.set(Http2Headers.PseudoHeaderName.PATH.value(), "/" +
url.getPath() + "/" + methodName);
+ headers.set(Http2Headers.PseudoHeaderName.METHOD.value(),
HttpMethod.POST.asciiName());
+ headers.set(HttpHeaderNames.TE, HttpHeaderValues.TRAILERS);
+ headers.set(TripleHeaderEnum.GRPC_ENCODING.getHeader(),
DEFAULT_COMPRESSOR);
+ headers.set(TripleHeaderEnum.CONTENT_TYPE_KEY.getHeader(),
TripleHeaderEnum.CONTENT_PROTO.getHeader());
+ headers.set(TripleHeaderEnum.TIMEOUT.getHeader(), timeout + "m");
+ return headers;
+ }
+
+ private Http2StreamChannel getHttp2StreamChannel(AbstractServerStream
stream) {
+ ChannelFuture channelFuture = Mockito.mock(ChannelFuture.class);
+ Http2StreamChannel streamChannel =
Mockito.mock(Http2StreamChannel.class);
+ Attribute<AbstractServerStream> attribute =
Mockito.mock(Attribute.class);
+ ChannelPromise promise = Mockito.mock(ChannelPromise.class);
+ EventLoop eventLoop = new DefaultEventLoop();
+
+ Mockito.when(attribute.get()).thenReturn(stream);
+
Mockito.when(streamChannel.writeAndFlush(Mockito.any())).thenReturn(channelFuture);
+
Mockito.when(streamChannel.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
+
Mockito.when(streamChannel.attr(TripleConstant.SERVER_STREAM_KEY)).thenReturn(attribute);
+ Mockito.when(streamChannel.eventLoop()).thenReturn(eventLoop);
+ Mockito.when(streamChannel.newPromise()).thenReturn(promise);
+
+ Mockito.when(streamChannel.write(Mockito.any(),
Mockito.any())).thenAnswer(
+ (Answer<ChannelPromise>) invocationOnMock -> {
+ writeMethodCalledTimes.incrementAndGet();
+ return promise;
+ });
+ return streamChannel;
+ }
+
+ private byte[] getPackedData(AbstractServerStream stream, Object[] args,
Class[] argsTypes) throws IOException {
+ String serializationName = TripleConstant.HESSIAN2;
+ final TripleWrapper.TripleRequestWrapper.Builder builder =
TripleWrapper.TripleRequestWrapper.newBuilder()
+ .setSerializeType(TripleConstant.HESSIAN4);
+ MultipleSerialization serialization =
stream.getMultipleSerialization();
+ for (int i = 0; i < args.length; i++) {
+ final String clz = argsTypes[i].getName();
+ builder.addArgTypes(clz);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serialization.serialize(url, serializationName, clz, args[i], bos);
+ builder.addArgs(ByteString.copyFrom(bos.toByteArray()));
+ }
+ TripleWrapper.TripleRequestWrapper requestWrapper = builder.build();
+ byte[] bytes = stream.pack(requestWrapper);
+ return bytes;
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocolTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocolTest.java
index 821d6e9..4c7f194 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocolTest.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocolTest.java
@@ -17,7 +17,6 @@
package org.apache.dubbo.rpc.protocol.tri;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.dubbo.common.URL;
@@ -35,6 +34,7 @@ import org.apache.dubbo.rpc.model.ServiceMetadata;
import org.apache.dubbo.rpc.protocol.tri.support.IGreeter;
import org.apache.dubbo.rpc.protocol.tri.support.IGreeterImpl;
+import org.apache.dubbo.rpc.protocol.tri.support.MockStreamObserver;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -42,6 +42,7 @@ import org.junit.jupiter.api.Test;
public class TripleProtocolTest {
private Protocol protocol =
ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
private ProxyFactory proxy =
ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
+ private final String REQUEST_MSG = "hello world";
@Test
public void testDemoProtocol() throws Exception {
@@ -68,32 +69,35 @@ public class TripleProtocolTest {
ConsumerModel consumerModel = new ConsumerModel(url.getServiceKey(),
null, serviceDescriptor, null,
new ServiceMetadata(), null);
url = url.setServiceModel(consumerModel);
- serviceImpl = proxy.getProxy(protocol.refer(IGreeter.class, url));
+ IGreeter greeterProxy = proxy.getProxy(protocol.refer(IGreeter.class,
url));
Thread.sleep(1000);
- Assertions.assertEquals("hello world", serviceImpl.echo("hello
world"));
- // fixme will throw exception
- // Assertions.assertEquals("hello world", serviceImpl.echoAsync("hello
world").get());
- CountDownLatch latch = new CountDownLatch(1);
- serviceImpl.serverStream("hello world", new StreamObserver<String>() {
- @Override
- public void onNext(String data) {
- Assertions.assertEquals("hello world",data);
- }
-
- @Override
- public void onError(Throwable throwable) {
- throwable.printStackTrace();
- }
-
- @Override
- public void onCompleted() {
- latch.countDown();
- System.out.println("onCompleted");
- }
- });
-
- // release CPU to run StreamObserver methods.
- latch.await(1000, TimeUnit.MILLISECONDS);
+
+ // 1. test unaryStream
+ Assertions.assertEquals(REQUEST_MSG, greeterProxy.echo(REQUEST_MSG));
+ Assertions.assertEquals(REQUEST_MSG,
serviceImpl.echoAsync(REQUEST_MSG).get());
+
+ // 2. test serverStream
+ MockStreamObserver outboundMessageSubscriber1 = new
MockStreamObserver();
+ greeterProxy.serverStream(REQUEST_MSG, outboundMessageSubscriber1);
+ outboundMessageSubscriber1.getLatch().await(1000,
TimeUnit.MILLISECONDS);
+ Assertions.assertEquals(outboundMessageSubscriber1.getOnNextData(),
REQUEST_MSG);
+ Assertions.assertTrue(outboundMessageSubscriber1.isOnCompleted());
+
+ // 3. test bidirectionalStream
+ MockStreamObserver outboundMessageSubscriber2 = new
MockStreamObserver();
+ StreamObserver<String> inboundMessageObserver =
greeterProxy.bidirectionalStream(outboundMessageSubscriber2);
+ inboundMessageObserver.onNext(REQUEST_MSG);
+ inboundMessageObserver.onCompleted();
+ outboundMessageSubscriber2.getLatch().await(1000,
TimeUnit.MILLISECONDS);
+ // verify client
+ Assertions.assertEquals(outboundMessageSubscriber2.getOnNextData(),
IGreeter.SERVER_MSG);
+ Assertions.assertTrue(outboundMessageSubscriber2.isOnCompleted());
+ // verify server
+ MockStreamObserver serverOutboundMessageSubscriber =
(MockStreamObserver) ((IGreeterImpl) serviceImpl).getMockStreamObserver();
+ serverOutboundMessageSubscriber.getLatch().await(1000,
TimeUnit.MILLISECONDS);
+
Assertions.assertEquals(serverOutboundMessageSubscriber.getOnNextData(),
REQUEST_MSG);
+ Assertions.assertTrue(serverOutboundMessageSubscriber.isOnCompleted());
+
// resource recycle.
serviceRepository.destroy();
System.out.println("serviceRepository destroyed");
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/WriteQueueTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/WriteQueueTest.java
new file mode 100644
index 0000000..5d0bf30
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/WriteQueueTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultEventLoop;
+import io.netty.channel.EventLoop;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.protocol.tri.command.CancelQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.DataQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.FlushQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.HeaderQueueCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.QueuedCommand;
+import org.apache.dubbo.rpc.protocol.tri.command.TextDataQueueCommand;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.dubbo.rpc.protocol.tri.WriteQueue.DEQUE_CHUNK_SIZE;
+
+
+/**
+ * {@link WriteQueue}
+ */
+public class WriteQueueTest {
+ private AtomicInteger writeMethodCalledTimes = new AtomicInteger(0);
+ private Channel channel;
+
+ @BeforeEach
+ public void init() {
+ channel = Mockito.mock(Channel.class);
+ ChannelPromise promise = Mockito.mock(ChannelPromise.class);
+ EventLoop eventLoop = new DefaultEventLoop();
+ Mockito.when(channel.eventLoop()).thenReturn(eventLoop);
+ Mockito.when(channel.newPromise()).thenReturn(promise);
+ Mockito.when(channel.write(Mockito.any(), Mockito.any())).thenAnswer(
+ (Answer<ChannelPromise>) invocationOnMock -> {
+ writeMethodCalledTimes.incrementAndGet();
+ return promise;
+ });
+
+ writeMethodCalledTimes.set(0);
+ }
+
+ @Test
+ public void test() throws Exception {
+
+ WriteQueue writeQueue = new WriteQueue(channel);
+ writeQueue.enqueue(HeaderQueueCommand.createHeaders(new
DefaultMetadata()), false);
+ writeQueue.enqueue(DataQueueCommand.createGrpcCommand(true), false);
+ GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
+ .withCause(new RpcException())
+ .withDescription("Encode Response data error");
+ writeQueue.enqueue(CancelQueueCommand.createCommand(status), false);
+
writeQueue.enqueue(TextDataQueueCommand.createCommand(status.description,
true), false);
+ writeQueue.enqueue(new FlushQueueCommand(), true);
+
+ while (writeMethodCalledTimes.get() != 5) {
+ Thread.sleep(50);
+ }
+
+ ArgumentCaptor<QueuedCommand> commandArgumentCaptor =
ArgumentCaptor.forClass(QueuedCommand.class);
+ ArgumentCaptor<ChannelPromise> promiseArgumentCaptor =
ArgumentCaptor.forClass(ChannelPromise.class);
+ Mockito.verify(channel,
Mockito.times(5)).write(commandArgumentCaptor.capture(),
promiseArgumentCaptor.capture());
+ List<QueuedCommand> queuedCommands =
commandArgumentCaptor.getAllValues();
+ Assertions.assertEquals(queuedCommands.size(), 5);
+ Assertions.assertTrue(queuedCommands.get(0) instanceof
HeaderQueueCommand);
+ Assertions.assertTrue(queuedCommands.get(1) instanceof
DataQueueCommand);
+ Assertions.assertTrue(queuedCommands.get(2) instanceof
CancelQueueCommand);
+ Assertions.assertTrue(queuedCommands.get(3) instanceof
TextDataQueueCommand);
+ Assertions.assertTrue(queuedCommands.get(4) instanceof
FlushQueueCommand);
+
+ Mockito.verify(channel, Mockito.times(1)).flush();
+ }
+
+ @Test
+ public void testChunk() throws Exception {
+ WriteQueue writeQueue = new WriteQueue(channel);
+ // test deque chunk size
+ writeMethodCalledTimes.set(0);
+ for (int i = 0; i < DEQUE_CHUNK_SIZE; i++) {
+ writeQueue.enqueue(HeaderQueueCommand.createHeaders(new
DefaultMetadata()), false);
+ }
+ writeQueue.enqueue(HeaderQueueCommand.createHeaders(new
DefaultMetadata()), true);
+ while (writeMethodCalledTimes.get() != (DEQUE_CHUNK_SIZE + 1)) {
+ Thread.sleep(50);
+ }
+ Mockito.verify(channel, Mockito.times(DEQUE_CHUNK_SIZE +
1)).write(Mockito.any(), Mockito.any());
+ Mockito.verify(channel, Mockito.times(2)).flush();
+ }
+
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter.java
index 55a5a7c..42a89af 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeter.java
@@ -22,6 +22,9 @@ import org.apache.dubbo.common.stream.StreamObserver;
import java.util.concurrent.CompletableFuture;
public interface IGreeter {
+
+ String SERVER_MSG = "HELLO WORLD";
+
/**
* Use request to respond
*/
@@ -33,5 +36,6 @@ public interface IGreeter {
void serverStream(String str, StreamObserver<String> observer);
+ StreamObserver<String> bidirectionalStream(StreamObserver<String>
observer);
}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
index daeeb0b..c4f14f4 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
@@ -21,6 +21,8 @@ import org.apache.dubbo.common.stream.StreamObserver;
public class IGreeterImpl implements IGreeter {
+ private StreamObserver<String> mockStreamObserver = new
MockStreamObserver();
+
@Override
public String echo(String request) {
return request;
@@ -28,8 +30,19 @@ public class IGreeterImpl implements IGreeter {
@Override
public void serverStream(String str, StreamObserver<String> observer) {
- System.out.println("srt="+str);
+ System.out.println("srt=" + str);
observer.onNext(str);
observer.onCompleted();
}
+
+ @Override
+ public StreamObserver<String> bidirectionalStream(StreamObserver<String>
observer) {
+ observer.onNext(SERVER_MSG);
+ observer.onCompleted();
+ return mockStreamObserver; // This will serve as the server's
outboundMessageSubscriber
+ }
+
+ public StreamObserver<String> getMockStreamObserver() {
+ return mockStreamObserver;
+ }
}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/MockAbstractStreamImpl.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/MockAbstractStreamImpl.java
new file mode 100644
index 0000000..4270e05
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/MockAbstractStreamImpl.java
@@ -0,0 +1,48 @@
+/*
+ * 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.support;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.rpc.protocol.tri.AbstractStream;
+import org.apache.dubbo.rpc.protocol.tri.InboundTransportObserver;
+
+public class MockAbstractStreamImpl extends AbstractStream {
+ public MockAbstractStreamImpl(URL url) {
+ super(url);
+ }
+
+ @Override
+ protected void cancelByRemoteReset() {
+
+ }
+
+ @Override
+ protected void cancelByLocal(Throwable throwable) {
+
+ }
+
+ @Override
+ protected StreamObserver<Object> createStreamObserver() {
+ return null;
+ }
+
+ @Override
+ protected InboundTransportObserver createInboundTransportObserver() {
+ return null;
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/MockStreamObserver.java
similarity index 51%
copy from
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
copy to
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/MockStreamObserver.java
index daeeb0b..f21686d 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/IGreeterImpl.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/support/MockStreamObserver.java
@@ -14,22 +14,50 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.dubbo.rpc.protocol.tri.support;
import org.apache.dubbo.common.stream.StreamObserver;
-public class IGreeterImpl implements IGreeter {
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * Usually use it to simulate a outboundMessageSubscriber
+ */
+public class MockStreamObserver implements StreamObserver<String> {
+ private String onNextData;
+ private Throwable onErrorThrowable;
+ private boolean onCompleted;
+ private CountDownLatch latch = new CountDownLatch(1);
@Override
- public String echo(String request) {
- return request;
+ public void onNext(String data) {
+ onNextData = data;
}
@Override
- public void serverStream(String str, StreamObserver<String> observer) {
- System.out.println("srt="+str);
- observer.onNext(str);
- observer.onCompleted();
+ public void onError(Throwable throwable) {
+ onErrorThrowable = throwable;
+ }
+
+ @Override
+ public void onCompleted() {
+ onCompleted = true;
+ latch.countDown();
+ }
+
+ public String getOnNextData() {
+ return onNextData;
+ }
+
+ public Throwable getOnErrorThrowable() {
+ return onErrorThrowable;
+ }
+
+ public boolean isOnCompleted() {
+ return onCompleted;
+ }
+
+ public CountDownLatch getLatch() {
+ return latch;
}
}
