diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java index 8bb27cbde3..2d1e125cbb 100644 --- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java +++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java @@ -103,4 +103,6 @@ public interface Constants { String H2_SETTINGS_INITIAL_WINDOW_SIZE_KEY = "dubbo.rpc.tri.initial-window-size"; String H2_SETTINGS_MAX_FRAME_SIZE_KEY = "dubbo.rpc.tri.max-frame-size"; String H2_SETTINGS_MAX_HEADER_LIST_SIZE_KEY = "dubbo.rpc.tri.max-header-list-size"; + + String H2_SUPPORT_NO_LOWER_HEADER_KEY = "dubbo.rpc.tri.support-no-lower-header"; } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/RequestMetadata.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/RequestMetadata.java index 0097b78fe4..b1ea629aa9 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/RequestMetadata.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/RequestMetadata.java @@ -47,7 +47,7 @@ public class RequestMetadata { public MethodDescriptor method; public PackableMethod packableMethod; public Map attachments; - + public boolean convertNoLowerHeader; public DefaultHttp2Headers toHeaders() { DefaultHttp2Headers header = new DefaultHttp2Headers(false); @@ -70,7 +70,7 @@ public class RequestMetadata { setIfNotNull(header, TripleHeaderEnum.GRPC_ENCODING.getHeader(), compressor.getMessageEncoding()); } - StreamUtils.convertAttachment(header, attachments); + StreamUtils.convertAttachment(header, attachments, convertNoLowerHeader); return header; } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java index 4927b3e6d1..b01e89cbbe 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java @@ -40,7 +40,11 @@ public enum TripleHeaderEnum { GRPC_ACCEPT_ENCODING("grpc-accept-encoding"), CONSUMER_APP_NAME_KEY("tri-consumer-appname"), SERVICE_VERSION("tri-service-version"), - SERVICE_GROUP("tri-service-group"); + SERVICE_GROUP("tri-service-group"), + + TRI_HEADER_CONVERT("tri-header-convert"), + + ; static final Map enumMap = new HashMap<>(); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java index c9ea8c375c..63df5735a6 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java @@ -239,6 +239,7 @@ public class TripleInvoker extends AbstractInvoker { } else { meta.packableMethod = ReflectionPackableMethod.init(methodDescriptor, url); } + meta.convertNoLowerHeader = TripleProtocol.CONVERT_NO_LOWER_HEADER; meta.method = methodDescriptor; meta.scheme = getSchemeFromUrl(url); // TODO read compressor from config 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 a7b0683d9d..5244e9de32 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 @@ -18,6 +18,7 @@ package org.apache.dubbo.rpc.protocol.tri; import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.config.ConfigurationUtils; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.threadpool.manager.ExecutorRepository; @@ -48,6 +49,8 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.ExecutorService; +import static org.apache.dubbo.rpc.Constants.H2_SUPPORT_NO_LOWER_HEADER_KEY; + public class TripleProtocol extends AbstractProtocol { @@ -57,13 +60,22 @@ public class TripleProtocol extends AbstractProtocol { private final TriBuiltinService triBuiltinService; private final ConnectionManager connectionManager; private final String acceptEncodings; + + /** + * There is only one + */ + public static boolean CONVERT_NO_LOWER_HEADER = false; + private boolean versionChecked = false; + public TripleProtocol(FrameworkModel frameworkModel) { this.frameworkModel = frameworkModel; this.triBuiltinService = new TriBuiltinService(frameworkModel); this.pathResolver = frameworkModel.getExtensionLoader(PathResolver.class) .getDefaultExtension(); + CONVERT_NO_LOWER_HEADER = ConfigurationUtils.getEnvConfiguration(ApplicationModel.defaultModel()) + .getBoolean(H2_SUPPORT_NO_LOWER_HEADER_KEY, false); Set supported = frameworkModel.getExtensionLoader(DeCompressor.class) .getSupportedExtensions(); this.acceptEncodings = String.join(",", supported); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java index 228c651e91..e0c6f61400 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/StreamUtils.java @@ -19,6 +19,8 @@ package org.apache.dubbo.rpc.protocol.tri.stream; import org.apache.dubbo.common.logger.Logger; import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.JsonUtils; +import org.apache.dubbo.rpc.TriRpcStatus; import org.apache.dubbo.rpc.protocol.tri.TripleConstant; import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; @@ -31,6 +33,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.Locale; import java.util.Map; +import java.util.stream.Collectors; public class StreamUtils { @@ -73,14 +76,17 @@ public class StreamUtils { * Parse and put the KV pairs into metadata. Ignore Http2 PseudoHeaderName and internal name. * Only raw byte array or string value will be put. * - * @param headers the metadata holder - * @param attachments KV pairs + * @param headers the metadata holder + * @param attachments KV pairs + * @param needConvertHeaderKey convert flag */ public static void convertAttachment(DefaultHttp2Headers headers, - Map attachments) { + Map attachments, + boolean needConvertHeaderKey) { if (attachments == null) { return; } + for (Map.Entry entry : attachments.entrySet()) { final String key = entry.getKey().toLowerCase(Locale.ROOT); if (Http2Headers.PseudoHeaderName.isPseudoHeader(key)) { @@ -92,6 +98,22 @@ public class StreamUtils { final Object v = entry.getValue(); convertSingleAttachment(headers, key, v); } + if (needConvertHeaderKey) { + Map needConvertKey = attachments.entrySet() + .stream() + .filter(it -> !headers.contains(it.getKey())) + .collect(Collectors.toMap(Map.Entry::getKey, it -> it.getKey().toLowerCase(Locale.ROOT))); + if (!needConvertKey.isEmpty()) { + String needConvertJson = JsonUtils.getJson().toJson(needConvertKey); + headers.add(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader(), TriRpcStatus.encodeMessage(needConvertJson)); + } + } + } + + + public static void convertAttachment(DefaultHttp2Headers headers, + Map attachments) { + convertAttachment(headers, attachments, false); } /** @@ -109,6 +131,8 @@ public class StreamUtils { } else if (v instanceof byte[]) { String str = encodeBase64ASCII((byte[]) v); headers.set(key + TripleConstant.HEADER_BIN_SUFFIX, str); + } else { + LOGGER.warn("Unsupported attachment k: " + key + " class: " + v.getClass().getName()); } } catch (Throwable t) { LOGGER.warn("Meet exception when convert single attachment key:" + key + " value=" + v, @@ -117,4 +141,5 @@ public class StreamUtils { } + } diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java index a3d19abb56..41815f6fab 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java @@ -17,6 +17,9 @@ package org.apache.dubbo.rpc.protocol.tri.stream; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.JsonUtils; import org.apache.dubbo.rpc.TriRpcStatus; import org.apache.dubbo.rpc.model.FrameworkModel; import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; @@ -57,6 +60,8 @@ import java.util.concurrent.Executor; */ public class TripleClientStream extends AbstractStream implements ClientStream { + private static final Logger LOGGER = LoggerFactory.getLogger(TripleClientStream.class); + public final ClientStream.Listener listener; private final WriteQueue writeQueue; private Deframer deframer; @@ -189,7 +194,28 @@ public class TripleClientStream extends AbstractStream implements ClientStream { final Map reserved = filterReservedHeaders(trailers); final Map attachments = headersToMap(trailers); - listener.onComplete(status, attachments, reserved); + final Map finalAttachments = convertNoLowerCaseHeader(attachments); + listener.onComplete(status, finalAttachments, reserved); + } + + private Map convertNoLowerCaseHeader(Map attachments) { + Object obj = attachments.remove(TripleHeaderEnum.TRI_HEADER_CONVERT.getHeader()); + if (obj == null) { + return attachments; + } + if (obj instanceof String) { + String json = TriRpcStatus.decodeMessage((String) obj); + Map map = JsonUtils.getJson().toJavaObject(json, Map.class); + map.forEach((originalKey, lowerCaseKey) -> { + Object val = attachments.remove(lowerCaseKey); + if (val != null) { + attachments.put(originalKey, val); + } + }); + } else { + LOGGER.error("Triple convertNoLowerCaseHeader error, obj is not String"); + } + return attachments; } private TriRpcStatus validateHeaderStatus(Http2Headers headers) { diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java index a9aa76c2aa..f289bb1706 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java @@ -29,6 +29,7 @@ import org.apache.dubbo.rpc.model.FrameworkModel; import org.apache.dubbo.rpc.protocol.tri.ExceptionUtils; import org.apache.dubbo.rpc.protocol.tri.TripleConstant; import org.apache.dubbo.rpc.protocol.tri.TripleHeaderEnum; +import org.apache.dubbo.rpc.protocol.tri.TripleProtocol; import org.apache.dubbo.rpc.protocol.tri.call.ReflectionAbstractServerCall; import org.apache.dubbo.rpc.protocol.tri.call.StubAbstractServerCall; import org.apache.dubbo.rpc.protocol.tri.command.CancelQueueCommand; @@ -173,7 +174,7 @@ public class TripleServerStream extends AbstractStream implements ServerStream { headers.status(HttpResponseStatus.OK.codeAsText()); headers.set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO); } - StreamUtils.convertAttachment(headers, attachments); + StreamUtils.convertAttachment(headers, attachments, TripleProtocol.CONVERT_NO_LOWER_HEADER); headers.set(TripleHeaderEnum.STATUS_KEY.getHeader(), String.valueOf(rpcStatus.code.code)); if (rpcStatus.isOk()) { return headers;