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

triple support upper header key (#10761) 

* tri support un lower header

* change config only one

* fix ut

* revert format

* revert format

* refactor

* ci
This commit is contained in:
earthchen 2022-10-21 13:59:33 +08:00 committed by GitHub
parent 31e2828d0c
commit f37502c005
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 79 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java
View File

@ -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";
}

4
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/RequestMetadata.java
View File

@ -47,7 +47,7 @@ public class RequestMetadata {
public MethodDescriptor method;
public PackableMethod packableMethod;
public Map<String, Object> 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;
}

6
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHeaderEnum.java
View File

@ -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<String, TripleHeaderEnum> enumMap = new HashMap<>();

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

@ -239,6 +239,7 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
} 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

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

@ -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<String> supported = frameworkModel.getExtensionLoader(DeCompressor.class)
.getSupportedExtensions();
this.acceptEncodings = String.join(",", supported);

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

@ -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<String, Object> attachments) {
Map<String, Object> attachments,
boolean needConvertHeaderKey) {
if (attachments == null) {
return;
}
for (Map.Entry<String, Object> 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<String, String> 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<String, Object> 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 {
}
}

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

@ -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<String, String> reserved = filterReservedHeaders(trailers);
final Map<String, Object> attachments = headersToMap(trailers);
listener.onComplete(status, attachments, reserved);
final Map<String, Object> finalAttachments = convertNoLowerCaseHeader(attachments);
listener.onComplete(status, finalAttachments, reserved);
}
private Map<String, Object> convertNoLowerCaseHeader(Map<String, Object> 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<String, String> 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) {

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

@ -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;