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 94e26b4 Add unit test for rpc-triple module (#9203)
94e26b4 is described below
commit 94e26b4c4a2379ec01fa29ee2d9daf0baad41dcc
Author: 灼华 <[email protected]>
AuthorDate: Fri Nov 5 10:48:32 2021 +0800
Add unit test for rpc-triple module (#9203)
* Add unit test for rpc-triple module
* Fix build
* Fix and add unit test
Co-authored-by: guohao <[email protected]>
---
.../threadpool/serial/SerializingExecutor.java | 2 +-
.../exchange/PortUnificationExchanger.java | 5 +-
.../apache/dubbo/rpc/CancellationContextTest.java | 61 ++++++++++
.../apache/dubbo/rpc/protocol/tri/GrpcStatus.java | 2 +-
.../rpc/protocol/tri/SingleProtobufUtils.java | 17 ++-
.../dubbo/rpc/protocol/tri/TripleProtocol.java | 3 -
.../dubbo/rpc/protocol/tri/UnaryClientStream.java | 2 +-
.../dubbo/rpc/protocol/tri/UnaryServerStream.java | 2 +-
.../protocol/tri/service/TriBuiltinService.java | 2 +-
.../rpc/protocol/tri/service/TriHealthImpl.java | 2 +-
.../rpc/protocol/tri/GrpcDataDecoderTest.java | 101 ++++++++++++++++
.../rpc/protocol/tri/SingleProtobufUtilsTest.java | 77 ++++++++++++
.../tri/TripleHttp2ClientResponseHandlerTest.java | 107 +++++++++++++++++
.../tri/service/TriBuiltinServiceTest.java | 43 +++++++
.../protocol/tri/service/TriHealthImplTest.java | 129 +++++++++++++++++++++
15 files changed, 534 insertions(+), 21 deletions(-)
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
index e106523..bc82f26 100644
---
a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
+++
b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
@@ -80,7 +80,7 @@ public final class SerializingExecutor implements Executor,
Runnable {
// ConcurrentLinkedQueue claims that null elements are
not allowed, but seems to not
// throw if the item to remove is null. If removable
is present in the queue twice,
// the wrong one may be removed. It doesn't seem
possible for this case to exist today.
- // This is important to run in case of
RejectedExectuionException, so that future calls
+ // This is important to run in case of
RejectedExecutionException, so that future calls
// to execute don't succeed and accidentally run a
previous runnable.
runQueue.remove(removable);
}
diff --git
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
index 3b0fd38..fc00d4c 100644
---
a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
+++
b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/PortUnificationExchanger.java
@@ -21,15 +21,14 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.remoting.api.PortUnificationServer;
-import io.netty.util.internal.PlatformDependent;
-
import java.util.ArrayList;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class PortUnificationExchanger {
private static final Logger log =
LoggerFactory.getLogger(PortUnificationExchanger.class);
- private static final ConcurrentMap<String, PortUnificationServer> servers
= PlatformDependent.newConcurrentHashMap();
+ private static final ConcurrentMap<String, PortUnificationServer> servers
= new ConcurrentHashMap<>();
public static void bind(URL url) {
servers.computeIfAbsent(url.getAddress(), addr -> {
diff --git
a/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/CancellationContextTest.java
b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/CancellationContextTest.java
new file mode 100644
index 0000000..aeffbcb
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-api/src/test/java/org/apache/dubbo/rpc/CancellationContextTest.java
@@ -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;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * {@link CancellationContext}
+ */
+public class CancellationContextTest {
+
+ @Test
+ public void test() throws Exception {
+ CancellationContext cancellationContext = new CancellationContext();
+
+ CountDownLatch latch = new CountDownLatch(2);
+ cancellationContext.addListener(rpcServiceContext -> {
+ latch.countDown();
+ });
+ cancellationContext.addListener(rpcServiceContext -> {
+ latch.countDown();
+ });
+ RuntimeException runtimeException = new RuntimeException();
+ cancellationContext.cancel(runtimeException);
+ latch.await();
+ Assertions.assertNull(cancellationContext.getListeners());
+ Assertions.assertTrue(cancellationContext.isCancelled());
+ Assertions.assertEquals(cancellationContext.getCancellationCause(),
runtimeException);
+ }
+
+ @Test
+ public void testAddListenerAfterCancel() throws Exception {
+ CancellationContext cancellationContext = new CancellationContext();
+
+ cancellationContext.cancel(null);
+ CountDownLatch latch = new CountDownLatch(1);
+ cancellationContext.addListener(rpcServiceContext -> {
+ latch.countDown();
+ });
+ latch.await();
+ }
+
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
index 56a3a21..b1eab70 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
@@ -150,7 +150,7 @@ public class GrpcStatus {
}
public static String fromMessage(String raw) {
- if (raw == null || raw.isEmpty()) {
+ if (StringUtils.isEmpty(raw)) {
return "";
}
return QueryStringDecoder.decodeComponent(raw);
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
index aca2b09..ab73a52 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtils.java
@@ -40,10 +40,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class SingleProtobufUtils {
- private static final ConcurrentHashMap<Class<?>, Message> instCache = new
ConcurrentHashMap<>();
- private static final ExtensionRegistryLite globalRegistry =
- ExtensionRegistryLite.getEmptyRegistry();
- private static final ConcurrentMap<Class<?>, SingleMessageMarshaller<?>>
marshallers = new ConcurrentHashMap<>();
+ private static final ConcurrentHashMap<Class<?>, Message> INST_CACHE = new
ConcurrentHashMap<>();
+ private static final ExtensionRegistryLite GLOBAL_REGISTRY =
ExtensionRegistryLite.getEmptyRegistry();
+ private static final ConcurrentMap<Class<?>, SingleMessageMarshaller<?>>
MARSHALLER_CACHE = new ConcurrentHashMap<>();
static {
// Built-in types need to be registered in advance
@@ -67,12 +66,12 @@ public class SingleProtobufUtils {
}
public static <T extends MessageLite> void marshaller(T defaultInstance) {
- marshallers.put(defaultInstance.getClass(), new
SingleMessageMarshaller<>(defaultInstance));
+ MARSHALLER_CACHE.put(defaultInstance.getClass(), new
SingleMessageMarshaller<>(defaultInstance));
}
@SuppressWarnings("all")
public static Message defaultInst(Class<?> clz) {
- Message defaultInst = instCache.get(clz);
+ Message defaultInst = INST_CACHE.get(clz);
if (defaultInst != null) {
return defaultInst;
}
@@ -81,7 +80,7 @@ public class SingleProtobufUtils {
} catch (IllegalAccessException | InvocationTargetException |
NoSuchMethodException e) {
throw new RuntimeException("Create default protobuf instance
failed ", e);
}
- instCache.put(clz, defaultInst);
+ INST_CACHE.put(clz, defaultInst);
return defaultInst;
}
@@ -110,7 +109,7 @@ public class SingleProtobufUtils {
}
private static SingleMessageMarshaller<?> getMarshaller(Class<?> clz) {
- return marshallers.computeIfAbsent(clz, k -> new
SingleMessageMarshaller(k));
+ return MARSHALLER_CACHE.computeIfAbsent(clz, k -> new
SingleMessageMarshaller(k));
}
public static final class SingleMessageMarshaller<T extends MessageLite> {
@@ -128,7 +127,7 @@ public class SingleProtobufUtils {
}
public T parse(InputStream stream) throws
InvalidProtocolBufferException {
- return parser.parseFrom(stream, globalRegistry);
+ return parser.parseFrom(stream, GLOBAL_REGISTRY);
}
}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
index d2b124a..f9d1187 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
@@ -37,9 +37,6 @@ import grpc.health.v1.HealthCheckResponse;
import static
org.apache.dubbo.common.constants.CommonConstants.DEFAULT_CLIENT_THREADPOOL;
import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
-/**
- *
- */
public class TripleProtocol extends AbstractProtocol implements Protocol {
private static final Logger logger =
LoggerFactory.getLogger(TripleProtocol.class);
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryClientStream.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryClientStream.java
index 30e8b94..f1eae16 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryClientStream.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryClientStream.java
@@ -69,7 +69,7 @@ public class UnaryClientStream extends AbstractClientStream
implements Stream {
return map;
}
- private class ClientUnaryInboundTransportObserver extends
ServerUnaryInboundTransportObserver implements TransportObserver {
+ private class ClientUnaryInboundTransportObserver extends
ServerUnaryInboundTransportObserver {
@Override
public void onComplete() {
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
index 21c418a..688091c 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java
@@ -45,7 +45,7 @@ public class UnaryServerStream extends AbstractServerStream
implements Stream {
return new UnaryServerTransportObserver();
}
- private class UnaryServerTransportObserver extends
ServerUnaryInboundTransportObserver implements TransportObserver {
+ private class UnaryServerTransportObserver extends
ServerUnaryInboundTransportObserver {
@Override
public void onError(GrpcStatus status) {
transportError(status);
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinService.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinService.java
index a0d6a6c..72143a4 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinService.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinService.java
@@ -36,7 +36,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.common.constants.CommonConstants.ANYHOST_VALUE;
/**
- * tri internal service like grpc internal service
+ * tri internal service like grpc internal service
**/
public class TriBuiltinService {
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImpl.java
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImpl.java
index 59bda55..2ec9e06 100644
---
a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImpl.java
+++
b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImpl.java
@@ -44,7 +44,7 @@ public class TriHealthImpl implements Health {
private final Object watchLock = new Object();
// Technically a Multimap<String, StreamObserver<HealthCheckResponse>>.
The Boolean value is not
- // used. The StreamObservers need to be kept in a identity-equality set,
to make sure
+ // used. The StreamObservers need to be kept in an identity-equality set,
to make sure
// user-defined equals() doesn't confuse our book-keeping of the
StreamObservers. Constructing
// such Multimap would require extra lines and the end result is not
significantly simpler, thus I
// would rather not have the Guava collections dependency.
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoderTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoderTest.java
new file mode 100644
index 0000000..ff4772f
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoderTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.Attribute;
+import org.apache.dubbo.rpc.RpcException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * {@link GrpcDataDecoder}
+ */
+public class GrpcDataDecoderTest {
+
+ @Test
+ public void test() throws Exception {
+ GrpcDataDecoder grpcDataDecoder = new GrpcDataDecoder(1024, true);
+ ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class);
+ Channel channel = Mockito.mock(Channel.class);
+ Mockito.when(ctx.channel()).thenReturn(channel);
+ ByteBuf in = ByteBufAllocator.DEFAULT.buffer();
+ List<Object> out = new ArrayList<>();
+
+ // test frame header malformed
+ in.writeByte(0xFE);
+ try {
+ grpcDataDecoder.decode(ctx, in, out);
+ Assertions.fail();
+ } catch (Exception e) {
+ Assertions.assertTrue(e instanceof RpcException);
+ RpcException rpcException = (RpcException) e;
+ Assertions.assertEquals(rpcException.getCode(),
GrpcStatus.Code.INTERNAL.code);
+ Assertions.assertEquals(rpcException.getMessage(), "gRPC frame
header malformed: reserved bits not zero");
+ }
+ in.clear();
+
+ // test message exceeds maximum size
+ in.writeByte(0);
+ in.writeInt(2048);
+ try {
+ grpcDataDecoder.decode(ctx, in, out);
+ } catch (Exception e) {
+ Assertions.assertTrue(e instanceof RpcException);
+ RpcException rpcException = (RpcException) e;
+ Assertions.assertEquals(rpcException.getCode(),
GrpcStatus.Code.RESOURCE_EXHAUSTED.code);
+ Assertions.assertEquals(rpcException.getMessage(), "gRPC message
exceeds maximum size 1024: 2048");
+ }
+
+ // test normal
+ in.writeByte(1);
+ in.writeInt(1024);
+ byte[] bytes = new byte[1024];
+ Arrays.fill(bytes, (byte) 1);
+ in.writeBytes(bytes);
+ in.markReaderIndex();
+ AbstractClientStream stream = Mockito.mock(AbstractClientStream.class);
+ Mockito.when(stream.getDeCompressor()).thenReturn(Compressor.NONE);
+ Attribute<AbstractClientStream> attribute =
Mockito.mock(Attribute.class);
+ Mockito.when(attribute.get()).thenReturn(stream);
+
Mockito.when(channel.attr(TripleConstant.CLIENT_STREAM_KEY)).thenReturn(attribute);
+ grpcDataDecoder.decode(ctx, in, out);
+ Assertions.assertFalse(out.isEmpty());
+ Assertions.assertArrayEquals((byte[]) out.get(0), bytes);
+
+ // test compressor not found
+ in.resetReaderIndex();
+ Mockito.when(stream.getDeCompressor()).thenReturn(null);
+ try {
+ grpcDataDecoder.decode(ctx, in, out);
+ } catch (Exception e) {
+ Assertions.assertTrue(e instanceof RpcException);
+ RpcException rpcException = (RpcException) e;
+ Assertions.assertEquals(rpcException.getCode(),
GrpcStatus.Code.UNIMPLEMENTED.code);
+ Assertions.assertEquals(rpcException.getMessage(), "gRPC message
compressor not found");
+ }
+
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtilsTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtilsTest.java
new file mode 100644
index 0000000..b74342e
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufUtilsTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.BoolValue;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.DoubleValue;
+import com.google.protobuf.Empty;
+import com.google.protobuf.EnumValue;
+import com.google.protobuf.FloatValue;
+import com.google.protobuf.Int32Value;
+import com.google.protobuf.Int64Value;
+import com.google.protobuf.ListValue;
+import com.google.protobuf.Message;
+import com.google.protobuf.Parser;
+import com.google.protobuf.StringValue;
+import grpc.health.v1.HealthCheckRequest;
+import grpc.health.v1.HealthCheckResponse;
+import org.apache.dubbo.triple.TripleWrapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+/**
+ * {@link SingleProtobufUtils}
+ */
+public class SingleProtobufUtilsTest {
+
+ @Test
+ public void test() throws IOException {
+
Assertions.assertFalse(SingleProtobufUtils.isSupported(SingleProtobufUtilsTest.class));
+ Assertions.assertTrue(SingleProtobufUtils.isSupported(Empty.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(BoolValue.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(Int32Value.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(Int64Value.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(FloatValue.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(DoubleValue.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(BytesValue.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(StringValue.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(EnumValue.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(ListValue.class));
+
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(HealthCheckResponse.class));
+
Assertions.assertTrue(SingleProtobufUtils.isSupported(HealthCheckRequest.class));
+
+ Message message =
SingleProtobufUtils.defaultInst(HealthCheckRequest.class);
+ Assertions.assertNotNull(message);
+ Parser<HealthCheckRequest> parser =
SingleProtobufUtils.getParser(HealthCheckRequest.class);
+ Assertions.assertNotNull(parser);
+
+ TripleWrapper.TripleRequestWrapper requestWrapper =
TripleWrapper.TripleRequestWrapper.newBuilder()
+ .setSerializeType("hessian4").build();
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ SingleProtobufUtils.serialize(requestWrapper,bos);
+
+ ByteArrayInputStream bis = new ByteArrayInputStream(bos.toByteArray());
+ TripleWrapper.TripleRequestWrapper tripleRequestWrapper =
SingleProtobufUtils.deserialize(bis, TripleWrapper.TripleRequestWrapper.class);
+
Assertions.assertEquals(tripleRequestWrapper.getSerializeType(),"hessian4");
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandlerTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandlerTest.java
new file mode 100644
index 0000000..a7660dd
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandlerTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.DefaultHttp2GoAwayFrame;
+import io.netty.handler.codec.http2.DefaultHttp2Headers;
+import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
+import io.netty.handler.codec.http2.DefaultHttp2ResetFrame;
+import io.netty.handler.codec.http2.Http2Error;
+import io.netty.handler.codec.http2.Http2GoAwayFrame;
+import io.netty.handler.codec.http2.Http2Headers;
+import io.netty.util.Attribute;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.rpc.RpcException;
+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;
+
+/**
+ * {@link TripleHttp2ClientResponseHandler }
+ */
+public class TripleHttp2ClientResponseHandlerTest {
+ private TripleHttp2ClientResponseHandler handler;
+ private ChannelHandlerContext ctx;
+ private AbstractClientStream stream;
+ private InboundTransportObserver transportObserver;
+ private StreamObserver streamObserver;
+
+
+ @BeforeEach
+ public void init() {
+ handler = new TripleHttp2ClientResponseHandler();
+
+ ctx = Mockito.mock(ChannelHandlerContext.class);
+ Channel channel = Mockito.mock(Channel.class);
+ Mockito.when(ctx.channel()).thenReturn(channel);
+
+ stream = Mockito.mock(AbstractClientStream.class);
+ transportObserver = Mockito.mock(InboundTransportObserver.class);
+ streamObserver = Mockito.mock(StreamObserver.class);
+
Mockito.when(stream.inboundTransportObserver()).thenReturn(transportObserver);
+
Mockito.when(stream.inboundMessageObserver()).thenReturn(streamObserver);
+
Mockito.when(stream.getUrl()).thenReturn(URL.valueOf("test://127.0.0.1/test"));
+ Mockito.when(stream.getDeCompressor()).thenReturn(Compressor.NONE);
+
+ Attribute<AbstractClientStream> attribute =
Mockito.mock(Attribute.class);
+ Mockito.when(attribute.get()).thenReturn(stream);
+
Mockito.when(channel.attr(TripleConstant.CLIENT_STREAM_KEY)).thenReturn(attribute);
+ }
+
+ @Test
+ public void testUserEventTriggered() throws Exception {
+ // test Http2GoAwayFrame
+ Http2GoAwayFrame goAwayFrame = new
DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR, ByteBufUtil
+ .writeAscii(ByteBufAllocator.DEFAULT, "app_requested"));
+ handler.userEventTriggered(ctx, goAwayFrame);
+ Mockito.verify(ctx, Mockito.times(1)).close();
+
+ // test Http2ResetFrame
+ DefaultHttp2ResetFrame resetFrame = new
DefaultHttp2ResetFrame(Http2Error.CANCEL);
+ handler.userEventTriggered(ctx, resetFrame);
+ Mockito.verify(ctx, Mockito.times(2)).close();
+ Mockito.verify(stream, Mockito.times(1)).cancelByRemote();
+ }
+
+ @Test
+ public void testChannelRead0() throws Exception {
+ final Http2Headers headers = new DefaultHttp2Headers(true);
+ headers.set(TripleHeaderEnum.GRPC_ENCODING.getHeader(), "gzip");
+ DefaultHttp2HeadersFrame headersFrame = new
DefaultHttp2HeadersFrame(headers, true);
+ handler.channelRead0(ctx, headersFrame);
+ Mockito.verify(transportObserver,
Mockito.times(1)).onMetadata(Mockito.any(), Mockito.anyBoolean());
+ Mockito.verify(transportObserver, Mockito.times(1)).onComplete();
+ }
+
+ @Test
+ public void testExceptionCaught() {
+ RuntimeException exception = new RuntimeException();
+ handler.exceptionCaught(ctx, exception);
+ ArgumentCaptor<RpcException> rpcExceptionArgumentCaptor =
ArgumentCaptor.forClass(RpcException.class);
+ Mockito.verify(streamObserver,
Mockito.times(1)).onError(rpcExceptionArgumentCaptor.capture());
+
Assertions.assertEquals(rpcExceptionArgumentCaptor.getValue().getCode(),
GrpcStatus.Code.INTERNAL.code);
+ Mockito.verify(ctx, Mockito.times(1)).close();
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinServiceTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinServiceTest.java
new file mode 100644
index 0000000..747686c
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/service/TriBuiltinServiceTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.service;
+
+import grpc.health.v1.Health;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.ModuleServiceRepository;
+import org.apache.dubbo.rpc.protocol.tri.PathResolver;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * {@link TriBuiltinService}
+ */
+public class TriBuiltinServiceTest {
+
+ @Test
+ public void test() {
+ FrameworkModel frameworkModel = new FrameworkModel();
+ TriBuiltinService triBuiltinService = new
TriBuiltinService(frameworkModel);
+ Assertions.assertNotNull(triBuiltinService.getHealthStatusManager());
+ PathResolver pathResolver =
frameworkModel.getExtensionLoader(PathResolver.class).getDefaultExtension();
+ Assertions.assertNotNull(pathResolver.resolve(Health.class.getName()));
+
+ ModuleServiceRepository repository =
frameworkModel.getInternalApplicationModel().getInternalModule().getServiceRepository();
+ Assertions.assertFalse(repository.getAllServices().isEmpty());
+ Assertions.assertFalse(repository.getExportedServices().isEmpty());
+ }
+}
diff --git
a/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImplTest.java
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImplTest.java
new file mode 100644
index 0000000..e62ee62
--- /dev/null
+++
b/dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/service/TriHealthImplTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.service;
+
+import grpc.health.v1.HealthCheckRequest;
+import grpc.health.v1.HealthCheckResponse;
+import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+
+/**
+ * {@link TriHealthImpl}
+ */
+public class TriHealthImplTest {
+
+ @Test
+ public void testCheck() {
+ TriHealthImpl triHealth = new TriHealthImpl();
+
+ HealthCheckRequest request = HealthCheckRequest.newBuilder().build();
+ HealthCheckResponse response = triHealth.check(request);
+ Assertions.assertEquals(response.getStatus(),
HealthCheckResponse.ServingStatus.SERVING);
+
+ HealthCheckRequest badRequest =
HealthCheckRequest.newBuilder().setService("test").build();
+ Assertions.assertThrows(RpcException.class, () ->
triHealth.check(badRequest));
+ }
+
+ @Test
+ public void testWatch() throws Exception {
+ TriHealthImpl triHealth = new TriHealthImpl();
+
+ HealthCheckRequest request = HealthCheckRequest.newBuilder().build();
+ StreamObserver<HealthCheckResponse> streamObserver = new
MockStreamObserver();
+
+ // test watch
+ triHealth.watch(request, streamObserver);
+ MockStreamObserver mockStreamObserver = (MockStreamObserver)
streamObserver;
+
Assertions.assertNotNull(RpcContext.getCancellationContext().getListeners());
+ HashMap<String, IdentityHashMap<StreamObserver<HealthCheckResponse>,
Boolean>> watches = getWatches(triHealth);
+ Assertions.assertTrue(watches.containsKey(request.getService()));
+
Assertions.assertTrue(watches.get(request.getService()).containsKey(streamObserver));
+
Assertions.assertTrue(watches.get(request.getService()).get(streamObserver));
+ Assertions.assertEquals(mockStreamObserver.getCount(), 1);
+ Assertions.assertEquals(mockStreamObserver.getResponse().getStatus(),
HealthCheckResponse.ServingStatus.SERVING);
+
+ // test setStatus
+ triHealth.setStatus(request.getService(),
HealthCheckResponse.ServingStatus.SERVICE_UNKNOWN);
+ Assertions.assertEquals(mockStreamObserver.getCount(), 2);
+ Assertions.assertEquals(mockStreamObserver.getResponse().getStatus(),
HealthCheckResponse.ServingStatus.SERVICE_UNKNOWN);
+
+ triHealth.enterTerminalState();
+ Assertions.assertEquals(mockStreamObserver.getCount(), 3);
+ Assertions.assertEquals(mockStreamObserver.getResponse().getStatus(),
HealthCheckResponse.ServingStatus.NOT_SERVING);
+
+ // test clearStatus
+ turnOffTerminal(triHealth);
+ triHealth.clearStatus(request.getService());
+ Assertions.assertEquals(mockStreamObserver.getCount(), 4);
+ Assertions.assertEquals(mockStreamObserver.getResponse().getStatus(),
HealthCheckResponse.ServingStatus.SERVICE_UNKNOWN);
+
+ // test listener
+ RpcContext.getCancellationContext().close();
+ Assertions.assertTrue(watches.isEmpty());
+ }
+
+ private void turnOffTerminal(TriHealthImpl triHealth) throws
NoSuchFieldException, IllegalAccessException {
+ Field terminalField =
triHealth.getClass().getDeclaredField("terminal");
+ terminalField.setAccessible(true);
+ terminalField.set(triHealth, false);
+ }
+
+ private HashMap<String,
IdentityHashMap<StreamObserver<HealthCheckResponse>, Boolean>>
getWatches(TriHealthImpl triHealth) throws Exception {
+ Field watchersField =
triHealth.getClass().getDeclaredField("watchers");
+ watchersField.setAccessible(true);
+ HashMap<String, IdentityHashMap<StreamObserver<HealthCheckResponse>,
Boolean>> watches =
+ (HashMap<String,
IdentityHashMap<StreamObserver<HealthCheckResponse>, Boolean>>)
watchersField.get(triHealth);
+ return watches;
+ }
+
+ class MockStreamObserver implements StreamObserver<HealthCheckResponse> {
+
+ private int count = 0;
+ private HealthCheckResponse response;
+
+ @Override
+ public void onNext(HealthCheckResponse data) {
+ count++;
+ response = data;
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+
+ }
+
+ @Override
+ public void onCompleted() {
+
+ }
+
+ public int getCount() {
+ return count;
+ }
+
+ public HealthCheckResponse getResponse() {
+ return response;
+ }
+ }
+}
