Integration with Apache EventMesh

functions-framework-api/pom.xml

@@ -28,6 +28,11 @@
             <artifactId>dapr-sdk</artifactId>
             <version>1.8.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.eventmesh</groupId>
+            <artifactId>eventmesh-sdk-java</artifactId>
+            <version>1.9.0-release</version>
+        </dependency>
     </dependencies>
 
     <parent>

functions-framework-api/src/main/java/dev/openfunction/functions/Context.java

@@ -16,6 +16,8 @@
 
 package dev.openfunction.functions;
 
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+
 import io.cloudevents.CloudEvent;
 import io.dapr.client.DaprClient;
 
@@ -121,5 +123,7 @@ public interface Context {
      */
     DaprClient getDaprClient();
 
+    EventMeshGrpcProducer getEventMeshProducer();
+
     byte[] packageAsCloudevent(String payload);
 }

functions-framework-invoker/pom.xml

@@ -63,6 +63,16 @@
             <artifactId>functions-framework-api</artifactId>
             <version>1.3.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.eventmesh</groupId>
+            <artifactId>eventmesh-connector-openfunction</artifactId>
+            <version>1.9.0-release</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.eventmesh</groupId>
+            <artifactId>eventmesh-sdk-java</artifactId>
+            <version>1.9.0-release</version>
+        </dependency>
         <dependency>
             <groupId>io.cloudevents</groupId>
             <artifactId>cloudevents-api</artifactId>
@@ -73,6 +83,11 @@
             <artifactId>cloudevents-core</artifactId>
             <version>${cloudevents.sdk.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.cloudevents</groupId>
+            <artifactId>cloudevents-protobuf</artifactId>
+            <version>${cloudevents.sdk.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.google.auto.value</groupId>
             <artifactId>auto-value</artifactId>

functions-framework-invoker/src/main/java/dev/openfunction/invoker/Runner.java

@@ -19,6 +19,7 @@
 
 import dev.openfunction.invoker.context.RuntimeContext;
 import dev.openfunction.invoker.trigger.DaprTrigger;
+import dev.openfunction.invoker.trigger.EventMeshTrigger;
 import dev.openfunction.invoker.trigger.HttpTrigger;
 import dev.openfunction.invoker.trigger.Trigger;
 import org.apache.commons.lang3.StringUtils;
@@ -82,6 +83,10 @@ public static void main(String[] args) {
                 triggers.add(new DaprTrigger(runtimeContext, functionClasses));
             }
 
+            if (runtimeContext.hasEventMeshTrigger()) {
+                triggers.add(new EventMeshTrigger(runtimeContext, functionClasses));
+            }
+
             for (Trigger trigger : triggers) {
                 trigger.start();
             }

functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/FunctionContext.java

@@ -172,6 +172,8 @@ static class Triggers {
         private HttpTrigger http;
         private DaprTrigger[] dapr;
 
+        private EventMeshTrigger eventMesh;
+
         public HttpTrigger getHttp() {
             return http;
         }
@@ -187,6 +189,14 @@ public DaprTrigger[] getDapr() {
         public void setDapr(DaprTrigger[] dapr) {
             this.dapr = dapr;
         }
+
+        public EventMeshTrigger getEventMesh() {
+            return eventMesh;
+        }
+
+        public void setEventMesh(EventMeshTrigger eventMesh) {
+            this.eventMesh = eventMesh;
+        }
     }
 
     static class HttpTrigger {
@@ -240,4 +250,99 @@ public void setInputName(String inputName) {
             this.inputName = inputName;
         }
     }
+
+    static class EventMeshTrigger {
+        private String name;
+
+        private String type;
+
+        private String topic;
+
+        private String eventMeshConnectorAddr;
+
+        private String eventMeshConnectorPort;
+
+        private String producerGroup;
+
+        private String env;
+
+        private String idc;
+
+        private String sysId;
+
+        public String getName() {
+            return name;
+        }
+
+        public void setName(String name) {
+            this.name = name;
+        }
+
+        public String getType() {
+            return type;
+        }
+
+        public void setType(String type) {
+            this.type = type;
+        }
+
+        public String getTopic() {
+            return topic;
+        }
+
+        public void setTopic(String topic) {
+            this.topic = topic;
+        }
+
+
+        public String getEventMeshConnectorAddr() {
+            return eventMeshConnectorAddr;
+        }
+
+        public void setEventMeshConnectorAddr(String eventMeshConnectorAddr) {
+            this.eventMeshConnectorAddr = eventMeshConnectorAddr;
+        }
+
+        public String getEventMeshConnectorPort() {
+            return eventMeshConnectorPort;
+        }
+
+        public void setEventMeshConnectorPort(String eventMeshConnectorPort) {
+            this.eventMeshConnectorPort = eventMeshConnectorPort;
+        }
+
+        public String getProducerGroup() {
+            return producerGroup;
+        }
+
+        public void setProducerGroup(String producerGroup) {
+            this.producerGroup = producerGroup;
+        }
+
+        public String getEnv() {
+            return env;
+        }
+
+        public void setEnv(String env) {
+            this.env = env;
+        }
+
+        public String getIdc() {
+            return idc;
+        }
+
+        public void setIdc(String idc) {
+            this.idc = idc;
+        }
+
+        public String getSysId() {
+            return sysId;
+        }
+
+        public void setSysId(String sysId) {
+            this.sysId = sysId;
+        }
+    }
+
+
 }

functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/RuntimeContext.java

@@ -26,7 +26,7 @@
 import dev.openfunction.invoker.tracing.TracingProvider;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.provider.EventFormatProvider;
-import org.apache.commons.collections.CollectionUtils;
+
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.ArrayUtils;
 
@@ -227,6 +227,36 @@ public Map<String, Component> getDaprTrigger() {
         return null;
     }
 
+
+    public boolean hasEventMeshTrigger() {
+        return functionContext.getTriggers() != null && functionContext.getTriggers().getEventMesh() != null;
+    }
+
+    public Component getEventMeshTrigger() {
+        if (functionContext.getTriggers() != null &&
+            functionContext.getTriggers().getEventMesh() != null) {
+            FunctionContext.EventMeshTrigger trigger = functionContext.getTriggers().getEventMesh();
+
+            Component component = new Component();
+            component.setComponentName(trigger.getName());
+            component.setComponentType(trigger.getType());
+            component.setTopic(trigger.getTopic());
+            Map<String, String> metaDataMap = new HashMap<>();
+            metaDataMap.put("eventMeshConnectorAddr", trigger.getEventMeshConnectorAddr());
+            metaDataMap.put("eventMeshConnectorPort", trigger.getEventMeshConnectorPort());
+            metaDataMap.put("producerGroup", trigger.getProducerGroup());
+            metaDataMap.put("env", trigger.getEnv());
+            metaDataMap.put("idc", trigger.getIdc());
+            metaDataMap.put("sysId", trigger.getSysId());
+
+            component.setMetadata(metaDataMap);
+
+            return component;
+        }
+
+        return null;
+    }
+
     public FunctionContext getFunctionContext() {
         return functionContext;
     }

functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/UserContext.java

@@ -24,6 +24,8 @@
 import io.dapr.client.DaprClient;
 import jakarta.servlet.http.HttpServletResponse;
 
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+
 import java.net.URI;
 import java.util.*;
 import java.util.logging.Level;
@@ -44,7 +46,9 @@ public class UserContext implements Context {
     );
 
     private final RuntimeContext runtimeContext;
-    private final DaprClient daprClient;
+    private DaprClient daprClient;
+
+    private EventMeshGrpcProducer eventMeshGrpcProducer;
 
     private Out out;
 
@@ -62,6 +66,11 @@ public UserContext(RuntimeContext runtimeContext, DaprClient daprClient) {
         this.daprClient = daprClient;
     }
 
+    public UserContext(RuntimeContext runtimeContext, EventMeshGrpcProducer eventMeshGrpcProducer) {
+        this.runtimeContext = runtimeContext;
+        this.eventMeshGrpcProducer = eventMeshGrpcProducer;
+    }
+
     public UserContext withHttp(HttpRequest httpRequest, HttpResponse httpResponse) {
         this.httpRequest = httpRequest;
         this.httpResponse = httpResponse;
@@ -167,6 +176,11 @@ public DaprClient getDaprClient() {
         return daprClient;
     }
 
+    @Override
+    public EventMeshGrpcProducer getEventMeshProducer() {
+        return eventMeshGrpcProducer;
+    }
+
     @Override
     public byte[] packageAsCloudevent(String payload) {
         CloudEvent event = new CloudEventBuilder()