[ISSUE #4577] Implement FilterEngine for EventMesh Filters (#4578)

* update ci.yml

* add filters

* add filters

* add filters

* apply filters under http processor

* update filterEngine

* [ISSUE #4577] Implement FilterEngine for EventMesh Filters

* refactor ci.yml

* fix ci check error

* fix ci check error

* fix ci check error

* fix ci check error

* remove duplicate method in PatternBuilder

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/http/common/EventMeshRetCode.java

@@ -42,6 +42,8 @@ public enum EventMeshRetCode {
     EVENTMESH_HEARTBEAT_ERR(21, "eventMesh heartbeat error"),
     EVENTMESH_ACL_ERR(22, "eventMesh acl error"),
     EVENTMESH_HTTP_MES_SEND_OVER_LIMIT_ERR(23, "eventMesh http msg send over the limit"),
+
+    EVENTMESH_FILTER_MSG_ERR(24, "eventMesh filter async msg error"),
     EVENTMESH_OPERATE_FAIL(100, "operate fail");
 
     private final Integer retCode;

eventmesh-filter/src/main/java/org/apache/eventmesh/filter/PatternEntry.java

@@ -26,26 +26,31 @@
 
 public class PatternEntry {
 
-    private final String patternPath;
+    private String patternName;
+
+    private String patternPath;
 
     private final List<Condition> conditionList = new ArrayList<>();
 
-    public PatternEntry(final String patternPath) {
+    public PatternEntry(final String patternName, final String patternPath) {
+        this.patternName = patternName;
         this.patternPath = patternPath;
     }
 
-    public void addRuleCondition(Condition patternCondition) {
+    public void addCondition(Condition patternCondition) {
         this.conditionList.add(patternCondition);
     }
 
     public String getPatternName() {
-        return "123";
+        return patternName;
     }
 
     public String getPatternPath() {
         return patternPath;
     }
 
+    // default filter type is OR
+    // todo: extend the filter type with AND
     public boolean match(JsonNode jsonElement) {
         for (final Condition patternCondition : conditionList) {
             if (patternCondition.match(jsonElement)) {

eventmesh-filter/src/main/java/org/apache/eventmesh/filter/patternbuild/PatternBuilder.java

@@ -74,21 +74,7 @@ public static Pattern build(String jsonStr) {
             }
 
             // iter all requiredField
-            parseRequiredField("$." + key, value, pattern);
-            if (value.isEmpty()) {
-                // Empty array
-                throw new JsonException("INVALID_PATTERN_VALUE ");
-            }
-            PatternEntry patternEntry = new PatternEntry("$." + key);
-            for (final JsonNode objNode : value) {
-                // {
-                // "suffix":".jpg"
-                // }
-                Condition condition = parseCondition(objNode);
-                patternEntry.addRuleCondition(condition);
-            }
-
-            pattern.addRequiredFieldList(patternEntry);
+            parseRequiredField(key, "$." + key, value, pattern);
         }
 
         return pattern;
@@ -112,12 +98,12 @@ private static void parseDataField(JsonNode jsonNode, Pattern pattern) {
                 String key = entry.getKey();
                 // [{"anything-but":"initializing"}] [{"anything-but":123}]}
                 JsonNode value = entry.getValue();
-                PatternEntry patternEntry = new PatternEntry(elepath + "." + key);
+                PatternEntry patternEntry = new PatternEntry(key, elepath + "." + key);
                 if (!value.isObject()) {
                     if (value.isArray()) {
                         for (JsonNode node11 : value) {
                             // {"anything-but":"initializing"}
-                            patternEntry.addRuleCondition(parseCondition(node11));
+                            patternEntry.addCondition(parseCondition(node11));
                         }
                     }
                     pattern.addDataList(patternEntry);
@@ -146,15 +132,15 @@ private static Condition parseCondition(JsonNode jsonNode) {
         return null;
     }
 
-    private static void parseRequiredField(String path, JsonNode jsonNode, Pattern pattern) {
+    private static void parseRequiredField(String patternName, String patternPath, JsonNode jsonNode, Pattern pattern) {
         if (jsonNode.isEmpty()) {
             // Empty array
             throw new JsonException("INVALID_PATTERN_VALUE ");
         }
-        PatternEntry patternEntry = new PatternEntry(path);
+        PatternEntry patternEntry = new PatternEntry(patternName, patternPath);
         for (final JsonNode objNode : jsonNode) {
             Condition condition = parseCondition(objNode);
-            patternEntry.addRuleCondition(condition);
+            patternEntry.addCondition(condition);
         }
 
         pattern.addRequiredFieldList(patternEntry);

eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/MetaService.java

@@ -54,7 +54,9 @@ public interface MetaService {
 
     void registerMetadata(Map<String, String> metadataMap);
 
-    String getMetaData(String key);
+    Map<String, String> getMetaData(String key, boolean fuzzyEnabled);
+
+    void getMetaDataWithListener(MetaServiceListener metaServiceListener, String key);
 
     void updateMetaData(Map<String, String> metadataMap);
 

eventmesh-meta/eventmesh-meta-api/src/main/java/org/apache/eventmesh/api/meta/MetaServiceListener.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.api.meta;
+
+/**
+ * MetaServiceListener
+ */
+public interface MetaServiceListener {
+
+    void onChange(String key, String value);
+}

eventmesh-meta/eventmesh-meta-consul/src/main/java/org/apache/eventmesh/meta/consul/service/ConsulMetaService.java

@@ -19,6 +19,7 @@
 
 import org.apache.eventmesh.api.exception.MetaException;
 import org.apache.eventmesh.api.meta.MetaService;
+import org.apache.eventmesh.api.meta.MetaServiceListener;
 import org.apache.eventmesh.api.meta.dto.EventMeshDataInfo;
 import org.apache.eventmesh.api.meta.dto.EventMeshRegisterInfo;
 import org.apache.eventmesh.api.meta.dto.EventMeshUnRegisterInfo;
@@ -162,6 +163,12 @@ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throw
         return true;
     }
 
+    // todo: to be implemented
+    @Override
+    public void getMetaDataWithListener(MetaServiceListener metaServiceListener, String key) {
+
+    }
+
     @Override
     public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws MetaException {
         HealthServicesRequest request = HealthServicesRequest.newBuilder().setPassing(true).setToken(token).build();
@@ -192,7 +199,7 @@ public void registerMetadata(Map<String, String> metadataMap) {
     }
 
     @Override
-    public String getMetaData(String key) {
+    public Map<String, String> getMetaData(String key, boolean fuzzyEnabled) {
         return null;
     }
 

eventmesh-meta/eventmesh-meta-etcd/src/main/java/org/apache/eventmesh/meta/etcd/service/EtcdMetaService.java

@@ -19,6 +19,7 @@
 
 import org.apache.eventmesh.api.exception.MetaException;
 import org.apache.eventmesh.api.meta.MetaService;
+import org.apache.eventmesh.api.meta.MetaServiceListener;
 import org.apache.eventmesh.api.meta.dto.EventMeshDataInfo;
 import org.apache.eventmesh.api.meta.dto.EventMeshRegisterInfo;
 import org.apache.eventmesh.api.meta.dto.EventMeshUnRegisterInfo;
@@ -196,10 +197,16 @@ public void registerMetadata(Map<String, String> metadataMap) {
     }
 
     @Override
-    public String getMetaData(String key) {
+    public Map<String, String> getMetaData(String key, boolean fuzzyEnabled) {
         return null;
     }
 
+    // todo: to be implemented
+    @Override
+    public void getMetaDataWithListener(MetaServiceListener metaServiceListener, String key) {
+
+    }
+
     @Override
     public void updateMetaData(Map<String, String> metadataMap) {
         String etcdMetaKey = instanceIp + "-" + group;

eventmesh-meta/eventmesh-meta-nacos/src/main/java/org/apache/eventmesh/meta/nacos/service/NacosMetaService.java

@@ -19,6 +19,7 @@
 
 import org.apache.eventmesh.api.exception.MetaException;
 import org.apache.eventmesh.api.meta.MetaService;
+import org.apache.eventmesh.api.meta.MetaServiceListener;
 import org.apache.eventmesh.api.meta.config.EventMeshMetaConfig;
 import org.apache.eventmesh.api.meta.dto.EventMeshDataInfo;
 import org.apache.eventmesh.api.meta.dto.EventMeshRegisterInfo;
@@ -31,26 +32,38 @@
 import org.apache.eventmesh.meta.nacos.constant.NacosConstant;
 
 import org.apache.commons.lang3.StringUtils;
-
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.alibaba.nacos.api.NacosFactory;
 import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.config.listener.Listener;
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.NamingService;
 import com.alibaba.nacos.api.naming.pojo.Instance;
 import com.alibaba.nacos.client.naming.utils.UtilAndComs;
 import com.alibaba.nacos.common.utils.CollectionUtils;
 import com.alibaba.nacos.common.utils.JacksonUtils;
+import com.fasterxml.jackson.databind.JsonNode;
 
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -85,6 +98,8 @@ public class NacosMetaService implements MetaService {
 
     private ConcurrentMap<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;
 
+    private MetaServiceListener metaServiceListener;
+
     @Override
     public void init() throws MetaException {
 
@@ -173,6 +188,26 @@ private Properties buildProperties() {
         return properties;
     }
 
+    @Override
+    public void getMetaDataWithListener(MetaServiceListener metaServiceListener, String key) {
+        try {
+            nacosConfigService.addListener(key, group, new Listener() {
+
+                @Override
+                public Executor getExecutor() {
+                    return null;
+                }
+
+                @Override
+                public void receiveConfigInfo(String configInfo) {
+                    metaServiceListener.onChange(key, configInfo);
+                }
+            });
+        } catch (Exception e) {
+            throw new RuntimeException("add nacos listener for key " + key + "error", e);
+        }
+    }
+
     @Override
     public void shutdown() throws MetaException {
         if (!initStatus.compareAndSet(true, false)) {
@@ -242,16 +277,67 @@ public void registerMetadata(Map<String, String> metadataMap) {
         }
     }
 
+    // implement with http
     @Override
-    public String getMetaData(String key) {
-        try {
-            return this.nacosConfigService.getConfig(key, group, 5000L);
-        } catch (NacosException e) {
+    public Map<String, String> getMetaData(String key, boolean fuzzyEnabled) {
+        if (fuzzyEnabled) {
+            key = key + "*";
+        }
+        int pageNo = 1;
+        int pageSize = 100;
+
+        Map<String, String> result = new HashMap<>();
+        Map<String, String> tmpMap;
+        do {
+            tmpMap = getResultFromNacos(pageNo, pageSize, key, group, fuzzyEnabled);
+            result.putAll(tmpMap);
+        } while (!(tmpMap.size() < pageSize));
+        return result;
+    }
+
+    private Map<String, String> getResultFromNacos(int pageNo, int pageSize, String key, String group, boolean fuzzyEnabled) {
+        Map<String, String> result = new HashMap<>();
+        try (CloseableHttpClient httpclient = HttpClients.createDefault()) {
+            URIBuilder uriBuilder = new URIBuilder("http://" + serverAddr + "/nacos/v1/cs/configs")
+                .setParameter("dataId", key)
+                .setParameter("group", group)
+                .setParameter("pageNo", String.valueOf(pageNo))
+                .setParameter("pageSize", String.valueOf(pageSize));
+            if (fuzzyEnabled) {
+                uriBuilder.setParameter("search", "blur");
+            }
+            URI uri = uriBuilder.build();
+            HttpGet httpGet = new HttpGet(uri);
+            try (CloseableHttpResponse closeableHttpResponse = httpclient.execute(httpGet)) {
+                if (closeableHttpResponse.getStatusLine().getStatusCode() == 200) {
+                    String response = EntityUtils.toString(closeableHttpResponse.getEntity(), StandardCharsets.UTF_8);
+                    result = processResponse(response);
+                }
+            } catch (Exception e) {
+                log.error("get metaData fail", e);
+                throw new RuntimeException(e);
+            }
+            return result;
+        } catch (Exception e) {
             log.error("get metaData fail", e);
             throw new RuntimeException(e);
         }
     }
 
+    private Map<String, String> processResponse(String response) {
+        Map<String, String> result = new HashMap<>();
+        JsonNode jsonNode = JacksonUtils.toObj(response);
+        JsonNode jsonNodeArray = jsonNode.get("pageItems");
+        if (jsonNodeArray.isArray()) {
+            for (JsonNode js : jsonNodeArray) {
+                String key = js.get("dataId").asText();
+                String value = js.get("content").asText();
+                result.put(key, value);
+            }
+        }
+        return result;
+    }
+
     @Override
     public void updateMetaData(Map<String, String> metadataMap) {
         String protocol = metadataMap.get(EventMeshMetaConfig.EVENT_MESH_PROTO);

eventmesh-meta/eventmesh-meta-nacos/src/test/java/org/apache/eventmesh/registry/nacos/service/NacosMetaServiceTest.java

@@ -19,7 +19,6 @@
 
 import static org.apache.eventmesh.common.Constants.HTTP;
 
-import org.apache.eventmesh.api.exception.MetaException;
 import org.apache.eventmesh.api.meta.dto.EventMeshRegisterInfo;
 import org.apache.eventmesh.api.meta.dto.EventMeshUnRegisterInfo;
 import org.apache.eventmesh.common.config.CommonConfiguration;
@@ -106,23 +105,4 @@ public void testShutdown() throws NoSuchFieldException, IllegalAccessException {
         Assertions.assertFalse((Boolean.parseBoolean(initStatusField.toString())));
         Assertions.assertFalse((Boolean.parseBoolean(startStatusField.toString())));
     }
-
-    @Test
-    public void testRegister() {
-        Assertions.assertThrows(MetaException.class, () -> {
-            nacosMetaService.init();
-            nacosMetaService.start();
-            nacosMetaService.register(eventMeshRegisterInfo);
-        });
-    }
-
-    @Test
-    public void testUnRegister() {
-        Assertions.assertThrows(MetaException.class, () -> {
-            nacosMetaService.init();
-            nacosMetaService.start();
-            nacosMetaService.unRegister(eventMeshUnRegisterInfo);
-        });
-    }
-
 }