[ISSUE #4621] Implement TransformerEngine for EventMesh Transformer

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHTTPServer.java

@@ -92,6 +92,8 @@
 
     private FilterEngine filterEngine;
 
+    private TransformerEngine transformerEngine;
+
     private HttpRetryer httpRetryer;
 
     private transient RateLimiter msgRateLimiter;
@@ -137,6 +139,8 @@
 
         filterEngine = new FilterEngine(metaStorage, producerManager, consumerManager);
 
+        transformerEngine = new TransformerEngine(metaStorage, producerManager, consumerManager);
+
         super.setHandlerService(new HandlerService());
         super.getHandlerService().setMetrics(this.getMetrics());
 
@@ -180,6 +184,8 @@
 
         filterEngine.shutdown();
 
+        transformerEngine.shutdown();
+
         consumerManager.shutdown();
 
         httpClientPool.shutdown();
@@ -354,6 +360,10 @@
         return filterEngine;
     }
 
+    public TransformerEngine getTransformerEngine() {
+        return transformerEngine;
+    }
+
     public MetaStorage getMetaStorage() {
         return metaStorage;
     }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/TransformerEngine.java

@@ -0,0 +1,140 @@
+/*
+ * 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.runtime.boot;
+
+import org.apache.eventmesh.api.meta.MetaServiceListener;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.LogUtils;
+import org.apache.eventmesh.runtime.core.protocol.http.consumer.ConsumerGroupManager;
+import org.apache.eventmesh.runtime.core.protocol.http.consumer.ConsumerManager;
+import org.apache.eventmesh.runtime.core.protocol.producer.EventMeshProducer;
+import org.apache.eventmesh.runtime.core.protocol.producer.ProducerManager;
+import org.apache.eventmesh.runtime.meta.MetaStorage;
+import org.apache.eventmesh.transformer.Transformer;
+import org.apache.eventmesh.transformer.TransformerBuilder;
+import org.apache.eventmesh.transformer.TransformerParam;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TransformerEngine {
+
+    /**
+     * key:group-topic
+     **/
+    private final Map<String, Transformer> transformerMap = new HashMap<>();
+
+    private final String transformerPrefix = "transformer-";
+
+    private final MetaStorage metaStorage;
+
+    private MetaServiceListener metaServiceListener;
+
+    private final ProducerManager producerManager;
+
+    private final ConsumerManager consumerManager;
+
+    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
+
+    public TransformerEngine(MetaStorage metaStorage, ProducerManager producerManager, ConsumerManager consumerManager) {
+        this.metaStorage = metaStorage;
+        this.producerManager = producerManager;
+        this.consumerManager = consumerManager;
+    }
+
+    public void start() {
+        Map<String, String> transformerMetaData = metaStorage.getMetaData(transformerPrefix, true);
+        for (Entry<String, String> transformerDataEntry : transformerMetaData.entrySet()) {
+            // transformer-group
+            String key = transformerDataEntry.getKey();
+            // topic-transformerParam list
+            String value = transformerDataEntry.getValue();
+            updateTransformerMap(key, value);
+        }
+        metaServiceListener = this::updateTransformerMap;
+
+        // addListeners for producerManager & consumerManager
+        scheduledExecutorService.scheduleAtFixedRate(() -> {
+            ConcurrentHashMap<String, EventMeshProducer> producerMap = producerManager.getProducerTable();
+            for (String producerGroup : producerMap.keySet()) {
+                for (String transformerKey : transformerMap.keySet()) {
+                    if (!StringUtils.contains(transformerKey, producerGroup)) {
+                        addTransformerListener(producerGroup);
+                        LogUtils.info(log, "addTransformerListener for producer group: " + producerGroup);
+                    }
+                }
+            }
+            ConcurrentHashMap<String, ConsumerGroupManager> consumerMap = consumerManager.getClientTable();
+            for (String consumerGroup : consumerMap.keySet()) {
+                for (String transformerKey : transformerMap.keySet()) {
+                    if (!StringUtils.contains(transformerKey, consumerGroup)) {
+                        addTransformerListener(consumerGroup);
+                        LogUtils.info(log, "addTransformerListener for consumer group: " + consumerGroup);
+                    }
+                }
+            }
+        }, 10_000, 5_000, TimeUnit.MILLISECONDS);
+    }
+
+    private void updateTransformerMap(String key, String value) {
+        String group = StringUtils.substringAfter(key, transformerPrefix);
+
+        JsonNode transformerJsonNodeArray = JsonUtils.getJsonNode(value);
+
+        if (transformerJsonNodeArray != null) {
+            for (JsonNode transformerJsonNode : transformerJsonNodeArray) {
+                String topic = transformerJsonNode.get("topic").asText();
+                String transformerParam = transformerJsonNode.get("transformerParam").toString();
+                TransformerParam tfp = JsonUtils.parseObject(transformerParam, TransformerParam.class);
+                Transformer transformer = TransformerBuilder.buildTransformer(tfp);
+                transformerMap.put(group + "-" + topic, transformer);
+            }
+        }
+        addTransformerListener(group);
+    }
+
+    public void addTransformerListener(String group) {
+        String transformerKey = transformerPrefix + group;
+        try {
+            metaStorage.getMetaDataWithListener(metaServiceListener, transformerKey);
+        } catch (Exception e) {
+            throw new RuntimeException("addTransformerListener exception", e);
+        }
+    }
+
+    public void shutdown() {
+        scheduledExecutorService.shutdown();
+    }
+
+    public Transformer getTransformer(String key) {
+        return transformerMap.get(key);
+    }
+
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncEventProcessor.java

@@ -45,6 +45,7 @@
 import org.apache.eventmesh.runtime.util.EventMeshUtil;
 import org.apache.eventmesh.runtime.util.RemotingHelper;
 import org.apache.eventmesh.trace.api.common.EventMeshTraceConstants;
+import org.apache.eventmesh.transformer.Transformer;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -161,6 +162,7 @@
         final String topic = event.getSubject();
 
         Pattern filterPattern = eventMeshHTTPServer.getFilterEngine().getFilterPattern(producerGroup + "-" + topic);
+        Transformer transformer = eventMeshHTTPServer.getTransformerEngine().getTransformer(producerGroup + "-" + topic);
 
         // validate body
         if (StringUtils.isAnyBlank(bizNo, uniqueId, producerGroup, topic)
@@ -252,6 +254,14 @@
                 isFiltered = filterPattern.filter(JsonUtils.toJSONString(event));
             }
 
+            // apply transformer
+            if (isFiltered && transformer != null) {
+                String data = transformer.transform(JsonUtils.toJSONString(event));
+                event = CloudEventBuilder.from(event).withData(Objects.requireNonNull(JsonUtils.toJSONString(data))
+                    .getBytes(StandardCharsets.UTF_8)).build();
+                sendMessageContext.setEvent(event);
+            }
+
             if (isFiltered) {
                 eventMeshProducer.send(sendMessageContext, new SendCallback() {
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/AsyncHTTPPushRequest.java

@@ -38,6 +38,7 @@
 import org.apache.eventmesh.runtime.core.protocol.http.consumer.HandleMsgContext;
 import org.apache.eventmesh.runtime.util.EventMeshUtil;
 import org.apache.eventmesh.runtime.util.WebhookUtil;
+import org.apache.eventmesh.transformer.Transformer;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.time.DateFormatUtils;
@@ -53,6 +54,7 @@
 
 import java.io.IOException;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -138,6 +140,20 @@
                 return;
             }
         }
+        Transformer transformer = eventMeshHTTPServer.getTransformerEngine()
+            .getTransformer(handleMsgContext.getConsumerGroup() + "-" + handleMsgContext.getTopic());
+        if (transformer != null) {
+            try {
+                String data = transformer.transform(JsonUtils.toJSONString(event));
+                event = CloudEventBuilder.from(event).withData(Objects.requireNonNull(JsonUtils.toJSONString(data))
+                    .getBytes(StandardCharsets.UTF_8)).build();
+            } catch (Exception exception) {
+                LOGGER.warn("apply transformer to cloudevents error, group:{}, topic:{}, bizSeqNo={}, uniqueId={}",
+                    this.handleMsgContext.getConsumerGroup(),
+                    this.handleMsgContext.getTopic(), this.handleMsgContext.getBizSeqNo(), this.handleMsgContext.getUniqueId(), exception);
+                return;
+            }
+        }
         handleMsgContext.setEvent(event);
         super.setEvent(event);
 

eventmesh-transformer/src/main/java/org/apache/eventmesh/transformer/JsonPathParser.java

@@ -72,8 +72,8 @@ public List<Variable> match(String json) throws JsonProcessingException {
 
         List<Variable> variableList = new ArrayList<>(variablesList.size());
         for (Variable element : variablesList) {
-            if (JsonPathUtils.isValidAndDefinite(element.getJsonPath())) {
-                String res = JsonPathUtils.matchJsonPathValueWithString(json, element.getJsonPath());
+            if (JsonPathUtils.isValidAndDefinite(element.getValue())) {
+                String res = JsonPathUtils.matchJsonPathValueWithString(json, element.getValue());
                 Variable variable = new Variable(element.getName(), res);
                 variableList.add(variable);
             } else {

eventmesh-transformer/src/main/java/org/apache/eventmesh/transformer/Template.java

@@ -34,8 +34,8 @@ public Template(String template) {
     public String substitute(List<Variable> variables) throws TransformException {
 
         Map<String, String> valuesMap = variables.stream()
-            .filter(variable -> variable.getJsonPath() != null)
-            .collect(Collectors.toMap(Variable::getName, Variable::getJsonPath));
+            .filter(variable -> variable.getValue() != null)
+            .collect(Collectors.toMap(Variable::getName, Variable::getValue));
         StringSubstitutor sub = new StringSubstitutor(valuesMap);
 
         return sub.replace(template);

eventmesh-transformer/src/main/java/org/apache/eventmesh/transformer/TemplateTransformer.java

@@ -37,8 +37,7 @@ public String transform(String json) throws JsonProcessingException {
         // 1: get variable match results
         List<Variable> variableList = jsonPathParser.match(json);
         // 2: use results replace template
-        String res = template.substitute(variableList);
-        return res;
+        return template.substitute(variableList);
     }
 
 }

eventmesh-transformer/src/main/java/org/apache/eventmesh/transformer/TransformerBuilder.java

@@ -19,41 +19,20 @@
 
 public class TransformerBuilder {
 
-    public static final class Builder {
-
-        private final TransformerType transformerType;
-        private String template;
-        private String content;
-
-        public Builder(TransformerType transformerType) {
-            this.transformerType = transformerType;
-        }
-
-        public Builder setContent(String content) {
-            this.content = content;
-            return this;
-        }
-
-        public Builder setTemplate(String template) {
-            this.template = template;
-            return this;
-        }
-
-        public Transformer build() {
-            switch (this.transformerType) {
-                case CONSTANT:
-                    return buildConstantTransformer(this.content);
-                case ORIGINAL:
-                    return buildOriginalTransformer();
-                case TEMPLATE:
-                    return buildTemplateTransFormer(this.content, this.template);
-                default:
-                    throw new TransformException("invalid config");
-            }
+    public static Transformer buildTransformer(TransformerParam transformerParam) {
+        switch (transformerParam.getTransformerType()) {
+            case ORIGINAL:
+                return buildOriginalTransformer();
+            case CONSTANT:
+                return buildConstantTransformer(transformerParam.getValue());
+            case TEMPLATE:
+                return buildTemplateTransFormer(transformerParam.getValue(), transformerParam.getTemplate());
+            default:
+                throw new TransformException("invalid config");
         }
-
     }
 
+
     public static Transformer buildTemplateTransFormer(String jsonContent, String template) {
         JsonPathParser jsonPathParser = new JsonPathParser(jsonContent);
         Template templateEntry = new Template(template);

eventmesh-transformer/src/main/java/org/apache/eventmesh/transformer/TransformerParam.java

@@ -0,0 +1,63 @@
+/*
+ * 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.transformer;
+
+public class TransformerParam {
+
+    private TransformerType transformerType;
+    private String value;
+    private String template;
+
+    public TransformerParam() {
+    }
+
+    public TransformerParam(TransformerType transformerType, String value, String template) {
+        this.transformerType = transformerType;
+        this.value = value;
+        this.template = template;
+    }
+
+    public TransformerParam(TransformerType transformerType, String value) {
+        this(transformerType, value, null);
+    }
+
+    public TransformerType getTransformerType() {
+        return transformerType;
+    }
+
+    public void setTransformerType(TransformerType transformerType) {
+        this.transformerType = transformerType;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    public String getTemplate() {
+        return template;
+    }
+
+    public void setTemplate(String template) {
+        this.template = template;
+    }
+
+}