BitoAgent commented on code in PR #13786:
URL: https://github.com/apache/dubbo/pull/13786#discussion_r1573616689
##########
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java:
##########
@@ -130,16 +127,12 @@ private void deliver() {
}
private void processHeader() throws IOException {
- ByteArrayOutputStream bos = new
ByteArrayOutputStream(lengthFieldOffset + lengthFieldLength);
byte[] offsetData = new byte[lengthFieldOffset];
int ignore = accumulate.read(offsetData);
- bos.write(offsetData);
processOffset(new ByteArrayInputStream(offsetData), lengthFieldOffset);
byte[] lengthBytes = new byte[lengthFieldLength];
ignore = accumulate.read(lengthBytes);
- bos.write(lengthBytes);
requiredLength = bytesToInt(lengthBytes);
- this.dataHeader = new ByteArrayInputStream(bos.toByteArray());
// Continue reading the frame body.
state = DecodeState.PAYLOAD;
Review Comment:
**Issue**: The refactoring to remove ByteArrayOutputStream and directly
process byte arrays is a good performance optimization. However, it's crucial
to ensure that the byte arrays are efficiently used and any error handling is
in place for I/O operations. <br> **Fix**: Implement efficient error handling
for I/O operations and ensure that byte arrays are used optimally to prevent
excessive memory usage. <br> **Code Suggestion**:
```
byte[] offsetData = a byte[lengthFieldOffset];
int ignore = accumulate.read(offsetData);
+ try {
+ if (offsetData.length > 0) {
+ processOffset(new ByteArrayInputStream(offsetData),
lengthFieldOffset);
+ } else {
+ throw new IOException("Empty byte array, cannot process offset");
+ }
+ } catch (IOException e) {
+ // handle error
+ }
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java:
##########
@@ -16,123 +16,103 @@
*/
package org.apache.dubbo.rpc.protocol.tri.h12.grpc;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.io.StreamUtils;
+import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.MediaType;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.PackableMethod;
+import org.apache.dubbo.rpc.model.PackableMethodFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.google.protobuf.Message;
-
-import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_METHOD_FACTORY;
public class GrpcCompositeCodec implements HttpMessageCodec {
- private final ProtobufHttpMessageCodec protobufHttpMessageCodec;
+ private static final String PACKABLE_METHOD_CACHE =
"PACKABLE_METHOD_CACHE";
- private final WrapperHttpMessageCodec wrapperHttpMessageCodec;
+ private final URL url;
- public GrpcCompositeCodec(
- ProtobufHttpMessageCodec protobufHttpMessageCodec,
WrapperHttpMessageCodec wrapperHttpMessageCodec) {
- this.protobufHttpMessageCodec = protobufHttpMessageCodec;
- this.wrapperHttpMessageCodec = wrapperHttpMessageCodec;
- }
+ private final FrameworkModel frameworkModel;
+
+ private final String mediaType;
- public void setEncodeTypes(Class<?>[] encodeTypes) {
- this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes);
+ private PackableMethod packableMethod;
+
+ public GrpcCompositeCodec(URL url, FrameworkModel frameworkModel, String
mediaType) {
+ this.url = url;
+ this.frameworkModel = frameworkModel;
+ this.mediaType = mediaType;
}
- public void setDecodeTypes(Class<?>[] decodeTypes) {
- this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes);
+ public void loadPackableMethod(MethodDescriptor methodDescriptor) {
+ if (methodDescriptor instanceof PackableMethod) {
+ packableMethod = (PackableMethod) methodDescriptor;
+ return;
+ }
+ Map<MethodDescriptor, PackableMethod> cacheMap =
(Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
+ .getServiceMetadata()
+ .getAttributeMap()
+ .computeIfAbsent(PACKABLE_METHOD_CACHE, k -> new
ConcurrentHashMap<>());
+ packableMethod = cacheMap.computeIfAbsent(methodDescriptor, md ->
frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType));
}
@Override
public void encode(OutputStream outputStream, Object data, Charset
charset) throws EncodeException {
- // protobuf
- // TODO int compressed =
Identity.MESSAGE_ENCODING.equals(requestMetadata.compressor.getMessageEncoding())
? 0 :
- // 1;
try {
- int compressed = 0;
- outputStream.write(compressed);
- if (isProtobuf(data)) {
- ProtobufWriter.write(protobufHttpMessageCodec, outputStream,
data);
- return;
- }
- // wrapper
- wrapperHttpMessageCodec.encode(outputStream, data);
- } catch (IOException e) {
+ outputStream.write(0);
+ byte[] bytes = packableMethod.packResponse(data);
+ writeLength(outputStream, bytes.length);
+ outputStream.write(bytes);
Review Comment:
**Issue**: The new implementation of the encode method simplifies the logic
by directly writing bytes to the outputStream. However, error handling for
IOExceptions is broadened to catch all Exceptions, which may obscure the root
cause of encoding issues. <br> **Fix**: Narrow the catch block to specifically
handle IOExceptions and add logging or more detailed error handling for
different types of exceptions. <br> **Code Suggestion**:
```
public void encode(OutputStream outputStream, Object data, Charset charset)
throws EncodeException {
try {
byte[] bytes = packableMethod.packResponse(data);
writeLength(outputStream, bytes.length);
outputStream.write(bytes);
} catch (IOException e) {
// Specific handling for IOException
throw new EncodeException("IOException during encoding", e);
} catch (Exception e) {
// Log or handle other exceptions
throw new EncodeException("Unexpected exception during encoding",
e);
}
}
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java:
##########
@@ -86,6 +86,7 @@ protected Executor initializeExecutor(Http2Header metadata) {
return new SerializingExecutor(executorSupport.getExecutor(metadata));
}
+ @Override
protected void doOnMetadata(Http2Header metadata) {
if (metadata.isEndStream()) {
if (!HttpMethods.supportBody(metadata.method())) {
Review Comment:
**Performance Issue**: The introduction of SerializingExecutor within the
initializeExecutor method might introduce unnecessary serialization of tasks
that could be executed concurrently, leading to potential performance
bottlenecks. <br> **Fix**: Consider using a more concurrent approach if the
tasks executed by this executor are independent and can be run in parallel
without causing race conditions or other concurrency-related issues. <br>
**Code Suggestion**:
```
- return new
SerializingExecutor(executorSupport.getExecutor(metadata));
+ return executorSupport.getExecutor(metadata); // Use the executor
directly to allow concurrent task execution
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java:
##########
@@ -16,123 +16,103 @@
*/
package org.apache.dubbo.rpc.protocol.tri.h12.grpc;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.io.StreamUtils;
+import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.MediaType;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.PackableMethod;
+import org.apache.dubbo.rpc.model.PackableMethodFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.google.protobuf.Message;
-
-import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_METHOD_FACTORY;
public class GrpcCompositeCodec implements HttpMessageCodec {
- private final ProtobufHttpMessageCodec protobufHttpMessageCodec;
+ private static final String PACKABLE_METHOD_CACHE =
"PACKABLE_METHOD_CACHE";
- private final WrapperHttpMessageCodec wrapperHttpMessageCodec;
+ private final URL url;
- public GrpcCompositeCodec(
- ProtobufHttpMessageCodec protobufHttpMessageCodec,
WrapperHttpMessageCodec wrapperHttpMessageCodec) {
- this.protobufHttpMessageCodec = protobufHttpMessageCodec;
- this.wrapperHttpMessageCodec = wrapperHttpMessageCodec;
- }
+ private final FrameworkModel frameworkModel;
+
+ private final String mediaType;
- public void setEncodeTypes(Class<?>[] encodeTypes) {
- this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes);
+ private PackableMethod packableMethod;
+
+ public GrpcCompositeCodec(URL url, FrameworkModel frameworkModel, String
mediaType) {
+ this.url = url;
+ this.frameworkModel = frameworkModel;
+ this.mediaType = mediaType;
}
- public void setDecodeTypes(Class<?>[] decodeTypes) {
- this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes);
+ public void loadPackableMethod(MethodDescriptor methodDescriptor) {
+ if (methodDescriptor instanceof PackableMethod) {
+ packableMethod = (PackableMethod) methodDescriptor;
+ return;
+ }
+ Map<MethodDescriptor, PackableMethod> cacheMap =
(Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
+ .getServiceMetadata()
+ .getAttributeMap()
+ .computeIfAbsent(PACKABLE_METHOD_CACHE, k -> new
ConcurrentHashMap<>());
+ packableMethod = cacheMap.computeIfAbsent(methodDescriptor, md ->
frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType));
Review Comment:
**Performance Issue**: The implementation of loadPackableMethod uses
ConcurrentHashMap.computeIfAbsent in a potentially inefficient manner. This
could lead to performance issues, especially under high load, as
computeIfAbsent locks the segment of the map, reducing concurrency. <br>
**Fix**: Consider pre-loading the methods into the cache during initialization
or on a separate thread if possible. Alternatively, use a different caching
strategy that minimizes locking, such as utilizing a loading cache from
libraries like Guava, which can handle concurrent requests more efficiently.
<br> **Code Suggestion**:
```
private final ConcurrentHashMap<MethodDescriptor, PackableMethod> cacheMap
= new ConcurrentHashMap<>();
// Pre-load packable methods into cache
class PackableMethodPreLoader {
public void preloadPackableMethods() {
// Preload methods
}
}
```
##########
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java:
##########
@@ -130,16 +127,12 @@ private void deliver() {
}
private void processHeader() throws IOException {
- ByteArrayOutputStream bos = new
ByteArrayOutputStream(lengthFieldOffset + lengthFieldLength);
byte[] offsetData = new byte[lengthFieldOffset];
int ignore = accumulate.read(offsetData);
- bos.write(offsetData);
processOffset(new ByteArrayInputStream(offsetData), lengthFieldOffset);
byte[] lengthBytes = new byte[lengthFieldLength];
ignore = accumulate.read(lengthBytes);
- bos.write(lengthBytes);
requiredLength = bytesToInt(lengthBytes);
- this.dataHeader = new ByteArrayInputStream(bos.toByteArray());
// Continue reading the frame body.
state = DecodeState.PAYLOAD;
Review Comment:
**Security Issue**: The removal of ByteArrayOutputStream usage and direct
passing of byte arrays without validation can lead to security vulnerabilities
related to improper input validation. <br> **Fix**: Validate the byte arrays
for expected format and size before processing to mitigate potential buffer
overflow or data corruption vulnerabilities. <br> **Code Suggestion**:
```
byte[] offsetData = new byte[lengthFieldOffset];
int ignore = accumulate.read(offsetData);
+ if (!validateByteArray(offsetData)) {
+ throw new InvalidDataException("Invalid byte array format");
+ }
processOffset(new ByteArrayInputStream(offsetData), lengthFieldOffset);
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcRequestHandlerMapping.java:
##########
@@ -42,9 +43,16 @@ protected boolean supportContentType(String contentType) {
@Override
protected void determineHttpMessageCodec(RequestHandler handler, URL url,
HttpRequest request) {
- HttpMessageCodec codec = CODEC_FACTORY.createCodec(url,
getFrameworkModel(), request.contentType());
- handler.setHttpMessageDecoder(codec);
- handler.setHttpMessageEncoder(codec);
+ GrpcCompositeCodec grpcCompositeCodec =
+ (GrpcCompositeCodec) CODEC_FACTORY.createCodec(url,
getFrameworkModel(), request.contentType());
+ MethodDescriptor methodDescriptor =
DescriptorUtils.findMethodDescriptor(
+ handler.getServiceDescriptor(), handler.getMethodName(),
handler.isHasStub());
+ if (methodDescriptor != null) {
+ handler.setMethodDescriptor(methodDescriptor);
+ grpcCompositeCodec.loadPackableMethod(methodDescriptor);
Review Comment:
**Security Issue**: The method 'DescriptorUtils.findMethodDescriptor' is
used without input validation, potentially allowing for injection attacks. <br>
**Fix**: Validate or sanitize 'handler.getMethodName()' and
'handler.isHasStub()' before using them in
'DescriptorUtils.findMethodDescriptor'. <br> **Code Suggestion**:
```
+ String methodName = sanitize(handler.getMethodName());
+ boolean hasStub = handler.isHasStub(); // Ensure 'hasStub' is
validated if necessary
+ MethodDescriptor methodDescriptor =
DescriptorUtils.findMethodDescriptor(
+ handler.getServiceDescriptor(), methodName, hasStub);
+ if (methodDescriptor != null) {
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java:
##########
@@ -16,123 +16,103 @@
*/
package org.apache.dubbo.rpc.protocol.tri.h12.grpc;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.io.StreamUtils;
+import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.MediaType;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.PackableMethod;
+import org.apache.dubbo.rpc.model.PackableMethodFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.google.protobuf.Message;
-
-import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_METHOD_FACTORY;
public class GrpcCompositeCodec implements HttpMessageCodec {
- private final ProtobufHttpMessageCodec protobufHttpMessageCodec;
+ private static final String PACKABLE_METHOD_CACHE =
"PACKABLE_METHOD_CACHE";
- private final WrapperHttpMessageCodec wrapperHttpMessageCodec;
+ private final URL url;
- public GrpcCompositeCodec(
- ProtobufHttpMessageCodec protobufHttpMessageCodec,
WrapperHttpMessageCodec wrapperHttpMessageCodec) {
- this.protobufHttpMessageCodec = protobufHttpMessageCodec;
- this.wrapperHttpMessageCodec = wrapperHttpMessageCodec;
- }
+ private final FrameworkModel frameworkModel;
+
+ private final String mediaType;
- public void setEncodeTypes(Class<?>[] encodeTypes) {
- this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes);
+ private PackableMethod packableMethod;
+
+ public GrpcCompositeCodec(URL url, FrameworkModel frameworkModel, String
mediaType) {
+ this.url = url;
+ this.frameworkModel = frameworkModel;
+ this.mediaType = mediaType;
}
- public void setDecodeTypes(Class<?>[] decodeTypes) {
- this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes);
+ public void loadPackableMethod(MethodDescriptor methodDescriptor) {
+ if (methodDescriptor instanceof PackableMethod) {
+ packableMethod = (PackableMethod) methodDescriptor;
+ return;
+ }
+ Map<MethodDescriptor, PackableMethod> cacheMap =
(Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
+ .getServiceMetadata()
+ .getAttributeMap()
+ .computeIfAbsent(PACKABLE_METHOD_CACHE, k -> new
ConcurrentHashMap<>());
+ packableMethod = cacheMap.computeIfAbsent(methodDescriptor, md ->
frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType));
Review Comment:
**Security Issue**: The use of a ConcurrentHashMap for caching packable
methods without proper synchronization or validation can lead to potential race
conditions and security vulnerabilities, such as unauthorized access or
manipulation of packable methods. <br> **Fix**: Implement proper
synchronization when accessing the ConcurrentHashMap. Additionally, validate
the MethodDescriptor instances to ensure they are authorized for caching and
use within the system. <br> **Code Suggestion**:
```
// Synchronize access to the cacheMap and validate MethodDescriptors
class PackableMethodCache {
private final ConcurrentHashMap<MethodDescriptor, PackableMethod>
cacheMap = new ConcurrentHashMap<>();
public PackableMethod getPackableMethod(MethodDescriptor descriptor) {
synchronized (cacheMap) {
// Validate descriptor
return cacheMap.computeIfAbsent(descriptor,
this::validateAndLoad);
}
}
private PackableMethod validateAndLoad(MethodDescriptor descriptor) {
// Validate and load method
}
}
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java:
##########
@@ -145,39 +146,31 @@ public void onMessage(InputStream inputStream) {
private class DetermineMethodDescriptorListener implements
StreamingDecoder.FragmentListener {
- @Override
- public void onFragmentMessage(InputStream rawMessage) {}
-
@Override
public void onClose() {
getStreamingDecoder().close();
}
@Override
- public void onFragmentMessage(InputStream dataHeader, InputStream
rawMessage) {
+ public void onFragmentMessage(InputStream rawMessage) {
try {
- ByteArrayOutputStream merged =
- new ByteArrayOutputStream(dataHeader.available() +
rawMessage.available());
- StreamUtils.copy(dataHeader, merged);
- byte[] data = StreamUtils.readBytes(rawMessage);
-
RpcInvocationBuildContext context = getContext();
if (null == context.getMethodDescriptor()) {
-
context.setMethodDescriptor(DescriptorUtils.findTripleMethodDescriptor(
- context.getServiceDescriptor(),
context.getMethodName(), data));
+ byte[] data = StreamUtils.readBytes(rawMessage);
+ MethodDescriptor methodDescriptor =
DescriptorUtils.findTripleMethodDescriptor(
+ context.getServiceDescriptor(),
context.getMethodName(), data);
+ context.setMethodDescriptor(methodDescriptor);
Review Comment:
**Security Issue**: The method 'DescriptorUtils.findTripleMethodDescriptor'
is being used without validating the input, which could lead to injection
vulnerabilities if the input is controlled by an attacker. <br> **Fix**: Ensure
that 'context.getMethodName()' and 'data' are properly validated or sanitized
before they are used in 'DescriptorUtils.findTripleMethodDescriptor'. <br>
**Code Suggestion**:
```
Ensure that 'context.getMethodName()' and 'data' are properly validated or
sanitized before they are used in 'DescriptorUtils.findTripleMethodDescriptor'.
+ byte[] data = StreamUtils.readBytes(rawMessage);
+ if(isValidData(data) &&
isValidMethodName(context.getMethodName())) {
+ MethodDescriptor methodDescriptor =
DescriptorUtils.findTripleMethodDescriptor(
+ context.getServiceDescriptor(),
context.getMethodName(), data);
+ context.setMethodDescriptor(methodDescriptor);
+ }
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java:
##########
@@ -16,123 +16,103 @@
*/
package org.apache.dubbo.rpc.protocol.tri.h12.grpc;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.io.StreamUtils;
+import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.MediaType;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.PackableMethod;
+import org.apache.dubbo.rpc.model.PackableMethodFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.google.protobuf.Message;
-
-import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_METHOD_FACTORY;
public class GrpcCompositeCodec implements HttpMessageCodec {
- private final ProtobufHttpMessageCodec protobufHttpMessageCodec;
+ private static final String PACKABLE_METHOD_CACHE =
"PACKABLE_METHOD_CACHE";
- private final WrapperHttpMessageCodec wrapperHttpMessageCodec;
+ private final URL url;
- public GrpcCompositeCodec(
- ProtobufHttpMessageCodec protobufHttpMessageCodec,
WrapperHttpMessageCodec wrapperHttpMessageCodec) {
- this.protobufHttpMessageCodec = protobufHttpMessageCodec;
- this.wrapperHttpMessageCodec = wrapperHttpMessageCodec;
- }
+ private final FrameworkModel frameworkModel;
+
+ private final String mediaType;
- public void setEncodeTypes(Class<?>[] encodeTypes) {
- this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes);
+ private PackableMethod packableMethod;
+
+ public GrpcCompositeCodec(URL url, FrameworkModel frameworkModel, String
mediaType) {
+ this.url = url;
+ this.frameworkModel = frameworkModel;
+ this.mediaType = mediaType;
}
- public void setDecodeTypes(Class<?>[] decodeTypes) {
- this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes);
+ public void loadPackableMethod(MethodDescriptor methodDescriptor) {
+ if (methodDescriptor instanceof PackableMethod) {
+ packableMethod = (PackableMethod) methodDescriptor;
+ return;
+ }
+ Map<MethodDescriptor, PackableMethod> cacheMap =
(Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
+ .getServiceMetadata()
+ .getAttributeMap()
+ .computeIfAbsent(PACKABLE_METHOD_CACHE, k -> new
ConcurrentHashMap<>());
+ packableMethod = cacheMap.computeIfAbsent(methodDescriptor, md ->
frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType));
Review Comment:
**Scalability Issue**: The implementation of loading and caching
PackableMethod instances could lead to excessive memory usage and potential
memory leaks if not properly managed, especially under high load. <br> **Fix**:
Consider implementing a more sophisticated caching strategy that limits the
size of the cache and evicts entries based on a policy (e.g., least recently
used). This can help in preventing the cache from growing indefinitely and
consuming too much memory. <br> **Code Suggestion**:
```
private final Cache<MethodDescriptor, PackableMethod> cache =
CacheBuilder.newBuilder()
.maximumSize(1000)
.expireAfterAccess(10, TimeUnit.MINUTES)
.build();
// Use cache with eviction policy
public PackableMethod getOrLoadPackableMethod(MethodDescriptor descriptor) {
return cache.get(descriptor, () -> loadPackableMethod(descriptor));
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
