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

Triple bugfix (#7263) 

* binary header value
* Remove channel context
* generic invoke
* mem leak in tri server handler
This commit is contained in:
GuoHao 2021-02-24 22:37:07 +08:00 committed by GitHub
parent d80c45b086
commit adcd2df4ee
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 25 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -161,7 +161,7 @@ public abstract class AbstractStream implements Stream {
if (v instanceof String) {
trailers.addObject(key, v);
} else if (v instanceof byte[]) {
trailers.addObject(key + "-bin", TripleUtil.encodeBase64((byte[]) v));
trailers.add(key + "-bin", TripleUtil.encodeBase64ASCII((byte[]) v));
}
} else {
if (v instanceof String || serializeType == null) {

30
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
View File

@ -213,17 +213,22 @@ public class ServerStream extends AbstractStream implements Stream {
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
ServiceRepository repo = ApplicationModel.getServiceRepository();
final List<MethodDescriptor> methods = serviceDescriptor.getMethods(methodName);
if (methods == null || methods.isEmpty()) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
.withDescription("Method not found:" + methodName + " of service:" + serviceDescriptor.getServiceName()));
return null;
}
if (methods.size() == 1) {
this.methodDescriptor = methods.get(0);
setNeedWrap(TripleUtil.needWrapper(this.methodDescriptor.getParameterClasses()));
} else {
// can not determine which one to invoke when same protobuf method name is used, force wrap it
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
this.methodDescriptor = repo.lookupMethod(GenericService.class.getName(), methodName);
setNeedWrap(true);
} else {
if (methods == null || methods.isEmpty()) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
.withDescription("Method not found:" + methodName + " of service:" + serviceDescriptor.getServiceName()));
return null;
}
if (methods.size() == 1) {
this.methodDescriptor = methods.get(0);
setNeedWrap(TripleUtil.needWrapper(this.methodDescriptor.getParameterClasses()));
} else {
// can not determine which one to invoke when same protobuf method name is used, force wrap it
setNeedWrap(true);
}
}
if (isNeedWrap()) {
loadFromURL(getUrl());
@ -236,9 +241,7 @@ public class ServerStream extends AbstractStream implements Stream {
if (isNeedWrap()) {
final TripleWrapper.TripleRequestWrapper req = TripleUtil.unpack(getData(), TripleWrapper.TripleRequestWrapper.class);
setSerializeType(req.getSerializeType());
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
this.methodDescriptor = repo.lookupMethod(GenericService.class.getName(), methodName);
} else {
if (this.methodDescriptor == null) {
String[] paramTypes = req.getArgTypesList().toArray(new String[req.getArgsCount()]);
for (MethodDescriptor method : methods) {
if (Arrays.equals(method.getCompatibleParamSignatures(), paramTypes)) {
@ -269,7 +272,6 @@ public class ServerStream extends AbstractStream implements Stream {
inv.setParameterTypes(methodDescriptor.getParameterClasses());
inv.setReturnTypes(methodDescriptor.getReturnTypes());
final Map<String, Object> attachments = parseHeadersToMap(getHeaders());
attachments.put(TripleConstant.TRI_CHANNEL_CTX_KEY, ctx);
inv.setObjectAttachments(attachments);
return inv;
}

1
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleConstant.java
View File

@ -17,7 +17,6 @@
package org.apache.dubbo.rpc.protocol.tri;
public interface TripleConstant {
String TRI_CHANNEL_CTX_KEY = "tri-ctx-channel";
String STATUS_KEY = "grpc-status";
String MESSAGE_KEY = "grpc-message";
String CONTENT_TYPE_KEY = "content-type";

9
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
View File

@ -20,8 +20,10 @@ package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.remoting.api.Http2WireProtocol;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http2.Http2FrameCodec;
import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
import io.netty.handler.codec.http2.Http2MultiplexHandler;
@ -47,7 +49,12 @@ public class TripleHttp2Protocol extends Http2WireProtocol {
.frameLogger(SERVER_LOGGER)
.build();
final Http2MultiplexHandler handler = new Http2MultiplexHandler(new TripleServerInitializer());
pipeline.addLast(codec, new TripleServerConnectionHandler(), handler);
pipeline.addLast(codec, new TripleServerConnectionHandler(), handler, new SimpleChannelInboundHandler<Object>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
// empty
}
});
}
@Override