reslove the check the event content (#362)

Author: Cristic

weevent-processor/src/main/java/com/webank/weevent/processor/mq/CEPRuleMQ.java

@@ -4,7 +4,6 @@
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.BlockingDeque;
@@ -139,7 +138,9 @@ public void onEvent(WeEvent event) {
                             String content = new String(event.getContent());
                             log.info("on event:{},content:{}", event.toString(), content);
 
-                            if ("json".equals(event.getExtensions().get("weevent-format")) && CommonUtil.checkValidJson(content)) {
+
+                            // check the content
+                            if (JSONObject.isValid(content)) {
                                 handleOnEvent(client, event, ruleMap);
                             } else {
                                 handleOnEventOtherPattern(client, event, ruleMap);
@@ -162,8 +163,8 @@ public void onEvent(WeEvent event) {
 
                             String content = new String(event.getContent());
                             log.info("on event:{},content:{}", event.toString(), content);
-
-                            if ("json".equals(event.getExtensions().get("weevent-format")) && CommonUtil.checkValidJson(content)) {
+                            // check the content
+                            if (JSONObject.isValid(content)) {
                                 handleOnEvent(client, event, ruleMap);
                             } else {
                                 handleOnEventOtherPattern(client, event, ruleMap);
@@ -246,7 +247,7 @@ private static void sendMessageToDB(String groupId, WeEvent eventContent, CEPRul
                     // execute the preparedstatement
                     int res = preparedStmt.executeUpdate();
                     if (res > 0) {
-                        log.info("insert db success!!!");
+                        log.info("insert db success...");
                     }
                     preparedStmt.close();
 
@@ -335,8 +336,7 @@ private static void handleOnEvent(IWeEventClient client, WeEvent event, Map<Stri
                             log.info("publish select: {},eventContent:{}", entry.getValue().getSelectField(), eventContent);
 
                             // publish the message
-                            Map<String, String> extensions = new HashMap<>();
-                            WeEvent weEvent = new WeEvent(entry.getValue().getToDestination(), eventContent.getBytes(StandardCharsets.UTF_8), extensions);
+                            WeEvent weEvent = new WeEvent(entry.getValue().getToDestination(), eventContent.getBytes(StandardCharsets.UTF_8), event.getExtensions());
                             log.info("after hitRuleEngine weEvent  groupId: {}, event:{}", groupId, weEvent.toString());
                             IWeEventClient toDestinationClient = getClient(entry.getValue());
                             toDestinationClient.publish(weEvent);

weevent-processor/src/main/java/com/webank/weevent/processor/utils/CommonUtil.java

@@ -140,8 +140,8 @@ public static boolean checkJson(String content, String objJson) {
             List<String> objJsonKeys = getKeys(objJson);
 
             for (String contentKey : contentKeys) {
-                //&& (!ConstantsHelper.EVENT_ID.equals(contentKey))
-                if (!objJsonKeys.contains(contentKey)) {
+
+                if (!((objJsonKeys.contains(contentKey)) || "eventId".equals(contentKey))) {
                     tag = false;
                     break;
                 }