[ISSUE #4472]Fix MeshMessageProtocolAdaptor#fromCloudEvent throw NPE

mxsm · mxsm · commit 8f72bf567e58 · 2023-10-06T20:20:41.000+08:00
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/Constants.java
@@ -169,6 +169,13 @@ public class Constants {
 
     public static final int SUCCESS_CODE = 200;
 
+    // protocol desc
+    public static final String PROTOCOL_DESC_GRPC_CLOUD_EVENT = "grpc-cloud-event";
+
+    public static final String PROTOCOL_DESC_HTTP = "http";
+
+    public static final String PROTOCOL_DESC_TCP = "tcp";
+
     /**
      * GRPC PROTOCOL
      */
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/MeshMessageProtocolAdaptor.java b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/MeshMessageProtocolAdaptor.java
@@ -18,6 +18,7 @@
 package org.apache.eventmesh.protocol.meshmessage;
 
 import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.enums.EventMeshProtocolType;
 import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
 import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEventBatch;
 import org.apache.eventmesh.common.protocol.grpc.common.BatchEventMeshCloudEventWrapper;
@@ -109,7 +110,7 @@ public ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) throws Prot
             cloudEvent.getExtension(Constants.PROTOCOL_DESC) == null ? null : cloudEvent.getExtension(Constants.PROTOCOL_DESC).toString();
 
         switch (Objects.requireNonNull(protocolDesc)) {
-            case MeshMessageProtocolConstant.PROTOCOL_DESC_HTTP:
+            case Constants.PROTOCOL_DESC_HTTP:
                 HttpCommand httpCommand = new HttpCommand();
                 Body body = new Body() {
 
@@ -128,9 +129,9 @@ public Map<String, Object> toMap() {
                 body.toMap();
                 httpCommand.setBody(body);
                 return httpCommand;
-            case MeshMessageProtocolConstant.PROTOCOL_DESC_TCP:
+            case Constants.PROTOCOL_DESC_TCP:
                 return TcpMessageProtocolResolver.buildEventMeshMessage(cloudEvent);
-            case MeshMessageProtocolConstant.PROTOCOL_DESC_GRPC_CLOUD_EVENT:
+            case Constants.PROTOCOL_DESC_GRPC_CLOUD_EVENT:
                 return GrpcEventMeshCloudEventProtocolResolver.buildEventMeshCloudEvent(cloudEvent);
             default:
                 throw new ProtocolHandleException(String.format("Unsupported protocolDesc: %s", protocolDesc));
@@ -139,7 +140,7 @@ public Map<String, Object> toMap() {
 
     @Override
     public String getProtocolType() {
-        return MeshMessageProtocolConstant.PROTOCOL_NAME;
+        return EventMeshProtocolType.EVENT_MESH_MESSAGE.protocolTypeName();
     }
 
     private void validateCloudEvent(CloudEvent cloudEvent) {
diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/MeshMessageProtocolConstant.java b/eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/src/main/java/org/apache/eventmesh/protocol/meshmessage/MeshMessageProtocolConstant.java
@@ -22,13 +22,5 @@ public enum MeshMessageProtocolConstant {
 
     public static final String PROTOCOL_NAME = "eventmeshmessage";
 
-    public static final String PROTOCOL_DESC_HTTP = "http";
-
-    public static final String PROTOCOL_DESC_GRPC = "grpc";
-
-    public static final String PROTOCOL_DESC_GRPC_CLOUD_EVENT = "grpc-cloud-event";
-
-    public static final String PROTOCOL_DESC_TCP = "tcp";
-
     public static final String PROTOCOL_KEY_CONTENT = "content";
 }
diff --git a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/grpc/util/EventMeshCloudEventBuilder.java
@@ -153,6 +153,8 @@ private static CloudEvent switchEventMeshMessage2EventMeshCloudEvent(EventMeshMe
         attributeValueMap.put(ProtocolKey.TTL, CloudEventAttributeValue.newBuilder().setCeString(ttl).build());
         attributeValueMap.put(ProtocolKey.SEQ_NUM, CloudEventAttributeValue.newBuilder().setCeString(seqNum).build());
         attributeValueMap.put(ProtocolKey.UNIQUE_ID, CloudEventAttributeValue.newBuilder().setCeString(uniqueId).build());
+        attributeValueMap.put(ProtocolKey.PROTOCOL_DESC,
+            CloudEventAttributeValue.newBuilder().setCeString(Constants.PROTOCOL_DESC_GRPC_CLOUD_EVENT).build());
         attributeValueMap.put(ProtocolKey.PRODUCERGROUP,
             CloudEventAttributeValue.newBuilder().setCeString(clientConfig.getProducerGroup()).build());
         if (null != message.getTopic()) {
@@ -197,7 +199,7 @@ private static CloudEvent switchCloudEvent2EventMeshCloudEvent(io.cloudevents.Cl
         buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.SYS, clientConfig.getSys());
         buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.LANGUAGE, Constants.LANGUAGE_JAVA);
         buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.PROTOCOL_TYPE, protocolType.protocolTypeName());
-        buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.PROTOCOL_DESC, "grpc-cloud-event");
+        buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.PROTOCOL_DESC, Constants.PROTOCOL_DESC_GRPC_CLOUD_EVENT);
         buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.PROTOCOL_VERSION, message.getSpecVersion().toString());
         buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.UNIQUE_ID, RandomStringUtils.generateNum(30));
         buildCloudEventIfAbsent(message, cloudEventBuilder, ProtocolKey.SEQ_NUM, RandomStringUtils.generateNum(30));
