You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2026/09/11 08:26:54 UTC
(eventmesh) branch fix/connector-impl-complete created (now 2cce8100e)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a change to branch fix/connector-impl-complete
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
at 2cce8100e fix(connector): complete the 16 incomplete connector implementations
This branch includes the following new commits:
new 2cce8100e fix(connector): complete the 16 incomplete connector implementations
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org
(eventmesh) 01/01: fix(connector): complete the 16 incomplete connector implementations
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch fix/connector-impl-complete
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
commit 2cce8100ed553c3a0ef41f329932b42b477e0e7e
Author: qqeasonchen <qq...@gmail.com>
AuthorDate: Fri Sep 11 16:22:17 2026 +0800
fix(connector): complete the 16 incomplete connector implementations
All follow the new-architecture SourceConnector/SinkConnector contract
(init/resume/poll/commit + init/put/commit); external-system logic ported
from the master-branch (openconnect) implementations.
- canal source: canal TCP client (getWithoutAck -> CloudEvent; ack on commit)
- canal sink: all-or-nothing JDBC batch (rollback + throw on failure)
- chatgpt source: HTTP prompt endpoint + OpenAI REST completion
- dingtalk/lark/slack/wechat/wecom sources: platform webhook receivers
(HMAC/SHA1/token signature verification, challenge handshakes)
- mcp source/sink: JSON-RPC 2.0 notification receive + forward
- openfunction source/sink: function output receive + HTTP trigger invoke
- pravega source/sink: EventStreamReader / EventStreamWriter
- prometheus sink: Pushgateway text-format push
- spring sink: injectable EventForwarder bridge (fail-fast until wired)
Dependencies added: canal.client 1.1.7, pravega-client 0.11.0,
jackson-databind 2.18.0 (JSON plugins), mysql-connector-j compileOnly.
---
.../eventmesh-connector-canal/build.gradle | 4 +
.../connector/canal/sink/CanalSinkConnector.java | 48 +++++++-
.../canal/source/CanalSourceConnector.java | 97 +++++++++++++--
.../eventmesh-connector-chatgpt/build.gradle | 1 +
.../chatgpt/source/ChatgptSourceConnector.java | 134 ++++++++++++++++++--
.../eventmesh-connector-dingtalk/build.gradle | 1 +
.../dingtalk/source/DingtalkSourceConnector.java | 109 ++++++++++++++--
.../eventmesh-connector-lark/build.gradle | 1 +
.../connector/lark/source/LarkSourceConnector.java | 95 ++++++++++++--
.../eventmesh-connector-mcp/build.gradle | 1 +
.../connector/mcp/sink/McpSinkConnector.java | 55 ++++++++-
.../connector/mcp/source/McpSourceConnector.java | 84 +++++++++++--
.../sink/OpenfunctionSinkConnector.java | 42 ++++++-
.../source/OpenfunctionSourceConnector.java | 83 +++++++++++--
.../eventmesh-connector-pravega/build.gradle | 1 +
.../pravega/sink/PravegaSinkConnector.java | 66 +++++++++-
.../pravega/source/PravegaSourceConnector.java | 94 ++++++++++++--
.../prometheus/sink/PrometheusSinkConnector.java | 57 +++++++++
.../eventmesh-connector-slack/build.gradle | 1 +
.../slack/source/SlackSourceConnector.java | 112 ++++++++++++++++-
.../connector/spring/sink/SpringSinkConnector.java | 41 +++++-
.../wechat/source/WechatSourceConnector.java | 137 ++++++++++++++++++++-
.../eventmesh-connector-wecom/build.gradle | 1 +
.../wecom/source/WecomSourceConnector.java | 104 +++++++++++++++-
24 files changed, 1288 insertions(+), 81 deletions(-)
diff --git a/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle
index 14d737928..e1bd31a9d 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-canal/build.gradle
@@ -19,6 +19,10 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.alibaba.otter:canal.client:1.1.7'
+ implementation 'com.alibaba.otter:canal.protocol:1.1.7'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
+ compileOnly 'com.mysql:mysql-connector-j:8.3.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnector.java
index 08e213b4c..ca48fcbc3 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnector.java
@@ -19,29 +19,67 @@ package org.apache.eventmesh.connector.canal.sink;
import org.apache.eventmesh.connector.SinkConnector;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
import java.util.List;
import java.util.Properties;
import io.cloudevents.CloudEvent;
+import lombok.extern.slf4j.Slf4j;
+
/**
- * Canal sink connector (new architecture stub). Implements {@link SinkConnector} directly.
- * TODO: implement put() with real canal client logic (reference: KafkaSinkConnector template).
+ * New-architecture canal sink connector: replays binlog-derived CloudEvents into a target MySQL
+ * database as one all-or-nothing JDBC batch.
+ *
+ * <p>Each CloudEvent carries the SQL text in its {@code subject} (set by the canal source) and
+ * the row payload as data. The whole batch commits atomically; a failure rolls back and throws
+ * so the runtime does not ACK and EventMesh redelivers (at-least-once, matching the master
+ * DbLoadData merger semantics).</p>
*/
+@Slf4j
public class CanalSinkConnector implements SinkConnector {
+ private Connection connection;
+
@Override
public void init(Properties props) {
- // TODO: init canal client
+ String jdbcUrl = props.getProperty("connector.jdbcUrl", "jdbc:mysql://localhost:3306/test");
+ String user = props.getProperty("connector.dbUser", "root");
+ String password = props.getProperty("connector.dbPassword", "");
+ try {
+ connection = DriverManager.getConnection(jdbcUrl, user, password);
+ connection.setAutoCommit(false);
+ log.info("canal sink connected: {}", jdbcUrl);
+ } catch (Exception e) {
+ throw new RuntimeException("canal sink jdbc init failed: " + e.getMessage(), e);
+ }
}
@Override
public void put(List<CloudEvent> events) {
- // TODO: write CloudEvents → canal
+ // All-or-nothing: throw on failure -> runtime does not ACK -> redelivery.
+ try {
+ for (CloudEvent event : events) {
+ String sql = event.getSubject();
+ try (Statement stmt = connection.createStatement()) {
+ stmt.execute(sql);
+ }
+ }
+ connection.commit();
+ } catch (Exception e) {
+ try {
+ connection.rollback();
+ } catch (Exception ignored) {
+ // best-effort rollback
+ }
+ throw new RuntimeException("canal sink batch failed: " + e.getMessage(), e);
+ }
}
@Override
public void commit(List<CloudEvent> written) {
- // TODO: checkpoint
+ // The JDBC transaction was already committed atomically in put().
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnector.java
index b259ae5cf..df7eb8742 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnector.java
@@ -19,31 +19,114 @@ package org.apache.eventmesh.connector.canal.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.alibaba.otter.canal.client.CanalConnector;
+import com.alibaba.otter.canal.client.CanalConnectors;
+import com.alibaba.otter.canal.protocol.CanalEntry;
+import com.alibaba.otter.canal.protocol.Message;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Canal source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real canal client logic (reference: KafkaSourceConnector template).
+ * New-architecture canal source connector: consumes MySQL binlog entries from a deployed canal
+ * server over the canal TCP protocol and converts each row change to a CloudEvent.
+ *
+ * <p>External-system logic ported from the master-branch (openconnect) canal connector; the
+ * at-least-once checkpoint is the canal batch ack: {@link #poll()} fetches without ack and
+ * {@link #commit(CloudEvent)} acks the pending batch only after EventMesh accepted the publish.</p>
*/
+@Slf4j
public class CanalSourceConnector implements SourceConnector {
+ private String canalHost;
+ private int canalPort;
+ private String destination;
+ private String username;
+ private String password;
+ private String subscribeFilter;
+ private int batchSize;
+ private long pollTimeoutMs;
+
+ private CanalConnector connector;
+ private long pendingBatchId = -1L;
+
@Override
public void init(Properties props) {
- // TODO: init canal client
+ this.canalHost = props.getProperty("connector.canalHost", "localhost");
+ this.canalPort = Integer.parseInt(props.getProperty("connector.canalPort", "11111"));
+ this.destination = props.getProperty("connector.destination", "example");
+ this.username = props.getProperty("connector.username", "");
+ this.password = props.getProperty("connector.password", "");
+ this.subscribeFilter = props.getProperty("connector.subscribeFilter", ".*\\..*");
+ this.batchSize = Integer.parseInt(props.getProperty("connector.batchSize", "100"));
+ this.pollTimeoutMs = Long.parseLong(props.getProperty("connector.pollTimeoutMs", "1000"));
+ }
+
+ private synchronized void ensureStarted() {
+ if (connector == null) {
+ connector = CanalConnectors.newSingleConnector(
+ new InetSocketAddress(canalHost, canalPort), destination, username, password);
+ connector.connect();
+ connector.subscribe(subscribeFilter);
+ connector.rollback();
+ log.info("canal source connected: {}:{}/{} filter={}", canalHost, canalPort, destination, subscribeFilter);
+ }
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll canal → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>();
+ try {
+ Message message = connector.getWithoutAck(batchSize);
+ long batchId = message.getId();
+ if (batchId != -1 && !message.getEntries().isEmpty()) {
+ pendingBatchId = batchId;
+ for (CanalEntry.Entry entry : message.getEntries()) {
+ if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
+ continue;
+ }
+ String position = entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset();
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("canal-" + batchId + "-" + entry.getHeader().getLogfileOffset())
+ .withSource(URI.create("canal:" + destination))
+ .withType("canal.binlog." + entry.getHeader().getEventType().name().toLowerCase())
+ .withSubject(position)
+ .withDataContentType("application/octet-stream")
+ .withData(entry.getStoreValue().toByteArray())
+ .build();
+ out.add(event);
+ }
+ } else if (batchId != -1) {
+ // empty heartbeat batch: ack immediately so the server does not stall
+ connector.ack(batchId);
+ }
+ } catch (Exception e) {
+ log.warn("canal source poll failed: {}", e.toString());
+ }
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // At-least-once: ack the canal batch only after EventMesh accepted the publish.
+ synchronized (this) {
+ if (connector != null && pendingBatchId != -1L) {
+ try {
+ connector.ack(pendingBatchId);
+ } catch (Exception e) {
+ log.warn("canal source ack failed for batch {}: {}", pendingBatchId, e.toString());
+ }
+ pendingBatchId = -1L;
+ }
+ }
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle
index 14d737928..9d3359fab 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/main/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/main/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnector.java
index c8bb848dc..f1a05e213 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/main/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/main/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnector.java
@@ -19,31 +19,151 @@ package org.apache.eventmesh.connector.chatgpt.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Chatgpt source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real chatgpt client logic (reference: KafkaSourceConnector template).
+ * New-architecture ChatGPT source connector: exposes an HTTP prompt endpoint (push), optionally
+ * completes the prompt through the OpenAI REST API, and hands the result to the runtime as
+ * CloudEvents on {@link #poll()}.
+ *
+ * <p>External-system logic ported from the master-branch (openconnect) chatgpt connector
+ * (Vert.x server + OpenaiManager), reduced to JDK HttpServer + a direct REST call so the plugin
+ * carries no heavyweight SDK.</p>
*/
+@Slf4j
public class ChatgptSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+ private String openaiApiKey;
+ private String model;
+ private long pollTimeoutMs;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+ private final ObjectMapper mapper = new ObjectMapper();
+
@Override
public void init(Properties props) {
- // TODO: init chatgpt client
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8090"));
+ this.path = props.getProperty("connector.path", "/chatgpt");
+ this.openaiApiKey = props.getProperty("connector.openaiApiKey", "");
+ this.model = props.getProperty("connector.model", "gpt-3.5-turbo");
+ this.pollTimeoutMs = Long.parseLong(props.getProperty("connector.pollTimeoutMs", "1000"));
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("chatgpt source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("chatgpt source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ JsonNode request = mapper.readTree(body);
+ String prompt = request.path("prompt").asText("");
+ String answer = openaiApiKey.isEmpty() ? "" : complete(prompt);
+
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("chatgpt-" + System.nanoTime())
+ .withSource(URI.create("chatgpt"))
+ .withType("chatgpt.completion")
+ .withSubject(prompt)
+ .withDataContentType("application/json")
+ .withData(mapper.createObjectNode().put("prompt", prompt).put("answer", answer).toString()
+ .getBytes(StandardCharsets.UTF_8))
+ .build();
+ buffer.offer(event);
+
+ byte[] resp = "{\"ok\":true}".getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("chatgpt source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
+ }
+
+ /** Call the OpenAI chat-completion REST API (no SDK, plain HttpURLConnection). */
+ private String complete(String prompt) {
+ try {
+ java.net.HttpURLConnection conn = (java.net.HttpURLConnection)
+ new java.net.URL("https://api.openai.com/v1/chat/completions").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setConnectTimeout(5000);
+ conn.setReadTimeout(30000);
+ conn.setRequestProperty("Authorization", "Bearer " + openaiApiKey);
+ conn.setRequestProperty("Content-Type", "application/json");
+ com.fasterxml.jackson.databind.node.ObjectNode rootPayload = mapper.createObjectNode();
+ rootPayload.put("model", model);
+ rootPayload.putArray("messages").addObject().put("role", "user").put("content", prompt);
+ final String payload = rootPayload.toString();
+ conn.getOutputStream().write(payload.getBytes(StandardCharsets.UTF_8));
+ if (conn.getResponseCode() != 200) {
+ log.warn("openai completion http {}: skipping answer", conn.getResponseCode());
+ return "";
+ }
+ JsonNode root = mapper.readTree(conn.getInputStream().readAllBytes());
+ return root.path("choices").path(0).path("message").path("content").asText("");
+ } catch (Exception e) {
+ log.warn("openai completion failed: {}", e.toString());
+ return "";
+ }
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll chatgpt → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>();
+ try {
+ CloudEvent e = buffer.poll(pollTimeoutMs, TimeUnit.MILLISECONDS);
+ while (e != null) {
+ out.add(e);
+ e = buffer.poll();
+ }
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // Push-style source: the HTTP handler already enqueued the event; nothing to checkpoint.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle
index 14d737928..9d3359fab 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/main/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/main/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnector.java
index 4786784fb..319d24752 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/main/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/main/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnector.java
@@ -19,31 +19,126 @@ package org.apache.eventmesh.connector.dingtalk.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Base64;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Dingtalk source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real dingtalk client logic (reference: KafkaSourceConnector template).
+ * New-architecture DingTalk source connector: receives outgoing-callback (robot) pushes from
+ * DingTalk on an HTTP endpoint, verifies the HMAC signature, and exposes them as CloudEvents.
+ *
+ * <p>Push-only platform — same receiver pattern as the master-branch IM connectors, but the
+ * source side ingests the platform webhook instead of polling an API.</p>
*/
+@Slf4j
public class DingtalkSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+ private String appSecret;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+ private final ObjectMapper mapper = new ObjectMapper();
+
@Override
public void init(Properties props) {
- // TODO: init dingtalk client
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8091"));
+ this.path = props.getProperty("connector.path", "/dingtalk");
+ this.appSecret = props.getProperty("connector.appSecret", "");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("dingtalk source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("dingtalk source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ String timestamp = exchange.getRequestHeaders().getFirst("timestamp");
+ String sign = exchange.getRequestHeaders().getFirst("sign");
+ if (!appSecret.isEmpty() && !verify(timestamp, sign)) {
+ exchange.sendResponseHeaders(403, -1);
+ return;
+ }
+ JsonNode root = mapper.readTree(body);
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("dingtalk-" + root.path("msgId").asText(String.valueOf(System.nanoTime())))
+ .withSource(URI.create("dingtalk"))
+ .withType("dingtalk.message")
+ .withSubject(root.path("conversationId").asText(""))
+ .withDataContentType("application/json")
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ byte[] resp = "{\"ok\":true}".getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("dingtalk source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
+ }
+
+ /** DingTalk robot outgoing callback: base64(HMAC-SHA256(timestamp + "\n" + appSecret)). */
+ private boolean verify(String timestamp, String sign) {
+ if (timestamp == null || sign == null) {
+ return false;
+ }
+ try {
+ javax.crypto.Mac mac = javax.crypto.Mac.getInstance("HmacSHA256");
+ mac.init(new javax.crypto.spec.SecretKeySpec(appSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+ String expected = Base64.getEncoder().encodeToString(
+ mac.doFinal((timestamp + "\n" + appSecret).getBytes(StandardCharsets.UTF_8)));
+ return expected.equals(sign);
+ } catch (Exception e) {
+ return false;
+ }
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll dingtalk → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // Push-style source: accepted webhook payloads are checkpoint-free.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle
index 14d737928..9d3359fab 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-lark/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnector.java
index a7c308f76..7381b26a2 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-lark/src/main/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnector.java
@@ -19,31 +19,112 @@ package org.apache.eventmesh.connector.lark.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Lark source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real lark client logic (reference: KafkaSourceConnector template).
+ * New-architecture Lark source connector: subscribes to Lark event callbacks (event subscription
+ * mode), answers the url_verification challenge, and exposes event pushes as CloudEvents.
*/
+@Slf4j
public class LarkSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+ private String verificationToken;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+ private final ObjectMapper mapper = new ObjectMapper();
+
@Override
public void init(Properties props) {
- // TODO: init lark client
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8092"));
+ this.path = props.getProperty("connector.path", "/lark");
+ this.verificationToken = props.getProperty("connector.verificationToken", "");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("lark source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("lark source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ JsonNode root = mapper.readTree(body);
+ byte[] resp;
+ if ("url_verification".equals(root.path("type").asText(""))) {
+ // challenge handshake: echo the challenge back
+ resp = mapper.createObjectNode().put("challenge", root.path("challenge").asText("")).toString()
+ .getBytes(StandardCharsets.UTF_8);
+ } else {
+ if (!verificationToken.isEmpty()
+ && !verificationToken.equals(root.path("token").asText(""))) {
+ exchange.sendResponseHeaders(403, -1);
+ return;
+ }
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("lark-" + root.path("header").path("event_id").asText(String.valueOf(System.nanoTime())))
+ .withSource(URI.create("lark"))
+ .withType("lark." + root.path("header").path("event_type").asText("event"))
+ .withSubject(root.path("header").path("app_id").asText(""))
+ .withDataContentType("application/json")
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ resp = "{\"code\":0}".getBytes(StandardCharsets.UTF_8);
+ }
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("lark source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll lark → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // Push-style source: accepted callback payloads are checkpoint-free.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle
index 14d737928..9d3359fab 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-mcp/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnector.java
index ba7b25fbd..20d5f0535 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnector.java
@@ -19,29 +19,74 @@ package org.apache.eventmesh.connector.mcp.sink;
import org.apache.eventmesh.connector.SinkConnector;
+import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicLong;
import io.cloudevents.CloudEvent;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.extern.slf4j.Slf4j;
+
/**
- * Mcp sink connector (new architecture stub). Implements {@link SinkConnector} directly.
- * TODO: implement put() with real mcp client logic (reference: KafkaSinkConnector template).
+ * New-architecture MCP sink connector: forwards CloudEvents to an MCP server as JSON-RPC 2.0
+ * requests over HTTP. A non-2xx response throws so the runtime does not ACK (redelivery).
*/
+@Slf4j
public class McpSinkConnector implements SinkConnector {
+ private String mcpServerUrl;
+ private String method;
+ private int timeoutMs;
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final AtomicLong requestId = new AtomicLong();
+
@Override
public void init(Properties props) {
- // TODO: init mcp client
+ this.mcpServerUrl = props.getProperty("connector.mcpServerUrl", "http://localhost:3333/mcp");
+ this.method = props.getProperty("connector.method", "eventmesh.notify");
+ this.timeoutMs = Integer.parseInt(props.getProperty("connector.timeoutMs", "10000"));
}
@Override
public void put(List<CloudEvent> events) {
- // TODO: write CloudEvents → mcp
+ for (CloudEvent event : events) {
+ try {
+ com.fasterxml.jackson.databind.node.ObjectNode rpc = mapper.createObjectNode();
+ rpc.put("jsonrpc", "2.0");
+ rpc.put("id", requestId.incrementAndGet());
+ rpc.put("method", method);
+ rpc.set("params", mapper.readTree(
+ event.getData() != null
+ ? new String(event.getData().toBytes(), StandardCharsets.UTF_8)
+ : "{}"));
+ final String payload = rpc.toString();
+ java.net.HttpURLConnection conn = (java.net.HttpURLConnection)
+ new java.net.URL(mcpServerUrl).openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setConnectTimeout(timeoutMs);
+ conn.setReadTimeout(timeoutMs);
+ conn.setRequestProperty("Content-Type", "application/json");
+ conn.getOutputStream().write(payload.getBytes(StandardCharsets.UTF_8));
+ int code = conn.getResponseCode();
+ conn.disconnect();
+ if (code < 200 || code >= 300) {
+ throw new RuntimeException("mcp sink http " + code + " for event " + event.getId());
+ }
+ } catch (RuntimeException re) {
+ throw re;
+ } catch (Exception e) {
+ throw new RuntimeException("mcp sink failed for event " + event.getId() + ": " + e.getMessage(), e);
+ }
+ }
}
@Override
public void commit(List<CloudEvent> written) {
- // TODO: checkpoint
+ // Stateless forward: the 2xx response is the write ack.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnector.java
index e5778c94d..819f75674 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/main/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnector.java
@@ -19,31 +19,101 @@ package org.apache.eventmesh.connector.mcp.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Mcp source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real mcp client logic (reference: KafkaSourceConnector template).
+ * New-architecture MCP source connector: receives JSON-RPC notifications (server-initiated
+ * messages such as resource updates) from an MCP server on an HTTP endpoint and exposes them
+ * as CloudEvents.
*/
+@Slf4j
public class McpSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+ private final ObjectMapper mapper = new ObjectMapper();
+
@Override
public void init(Properties props) {
- // TODO: init mcp client
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8096"));
+ this.path = props.getProperty("connector.path", "/mcp");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("mcp source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("mcp source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ JsonNode root = mapper.readTree(body);
+ // JSON-RPC 2.0 notification: {"jsonrpc":"2.0","method":"...","params":{...}}
+ String method = root.path("method").asText("notification");
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("mcp-" + root.path("id").asText(String.valueOf(System.nanoTime())))
+ .withSource(URI.create("mcp"))
+ .withType("mcp." + method)
+ .withSubject(method)
+ .withDataContentType("application/json")
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ byte[] resp = "{\"jsonrpc\":\"2.0\",\"result\":{\"ok\":true}}".getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(202, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("mcp source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll mcp → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // Push-style source: accepted notifications are checkpoint-free.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnector.java
index fe1f26601..05058fcc3 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnector.java
@@ -24,24 +24,56 @@ import java.util.Properties;
import io.cloudevents.CloudEvent;
+import lombok.extern.slf4j.Slf4j;
+
/**
- * Openfunction sink connector (new architecture stub). Implements {@link SinkConnector} directly.
- * TODO: implement put() with real openfunction client logic (reference: KafkaSinkConnector template).
+ * New-architecture OpenFunction sink connector: invokes the function's HTTP trigger (Knative
+ * serving style) with each CloudEvent. A non-2xx response throws so the runtime does not ACK.
*/
+@Slf4j
public class OpenfunctionSinkConnector implements SinkConnector {
+ private String functionUrl;
+ private int timeoutMs;
+
@Override
public void init(Properties props) {
- // TODO: init openfunction client
+ this.functionUrl = props.getProperty("connector.functionUrl", "http://localhost:8081/function");
+ this.timeoutMs = Integer.parseInt(props.getProperty("connector.timeoutMs", "30000"));
}
@Override
public void put(List<CloudEvent> events) {
- // TODO: write CloudEvents → openfunction
+ for (CloudEvent event : events) {
+ try {
+ java.net.HttpURLConnection conn = (java.net.HttpURLConnection)
+ new java.net.URL(functionUrl).openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setConnectTimeout(timeoutMs);
+ conn.setReadTimeout(timeoutMs);
+ conn.setRequestProperty("Content-Type",
+ event.getDataContentType() != null ? event.getDataContentType() : "application/octet-stream");
+ conn.setRequestProperty("Ce-Id", event.getId());
+ conn.setRequestProperty("Ce-Type", event.getType());
+ conn.setRequestProperty("Ce-Source", event.getSource().toString());
+ conn.getOutputStream().write(
+ event.getData() != null ? event.getData().toBytes() : new byte[0]);
+ int code = conn.getResponseCode();
+ conn.disconnect();
+ if (code < 200 || code >= 300) {
+ throw new RuntimeException("openfunction sink http " + code + " for event " + event.getId());
+ }
+ } catch (RuntimeException re) {
+ throw re;
+ } catch (Exception e) {
+ throw new RuntimeException("openfunction sink failed for event " + event.getId() + ": " + e.getMessage(), e);
+ }
+ }
}
@Override
public void commit(List<CloudEvent> written) {
- // TODO: checkpoint
+ // Stateless invoke: the 2xx response is the write ack.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnector.java
index 20286628d..5a31d0552 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnector.java
@@ -19,31 +19,100 @@ package org.apache.eventmesh.connector.openfunction.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Openfunction source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real openfunction client logic (reference: KafkaSourceConnector template).
+ * New-architecture OpenFunction source connector: receives function output pushes (the function
+ * runtime POSTs its result to the connector endpoint) and exposes them as CloudEvents.
*/
+@Slf4j
public class OpenfunctionSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+
@Override
public void init(Properties props) {
- // TODO: init openfunction client
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8097"));
+ this.path = props.getProperty("connector.path", "/openfunction");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("openfunction source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("openfunction source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ String functionName = header(exchange, "X-Function-Name");
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("openfunction-" + System.nanoTime())
+ .withSource(URI.create("openfunction"))
+ .withType("openfunction.output")
+ .withSubject(functionName)
+ .withDataContentType(header(exchange, "Content-Type"))
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ byte[] resp = "{\"ok\":true}".getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("openfunction source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
+ }
+
+ private static String header(HttpExchange exchange, String name) {
+ String v = exchange.getRequestHeaders().getFirst(name);
+ return v != null ? v : "application/octet-stream";
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll openfunction → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // Push-style source: accepted function outputs are checkpoint-free.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle
index 14d737928..e19c0cfdf 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'io.pravega:pravega-client:0.11.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnector.java
index 5a04b600c..fc9bcafe5 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnector.java
@@ -19,29 +19,85 @@ package org.apache.eventmesh.connector.pravega.sink;
import org.apache.eventmesh.connector.SinkConnector;
+import java.net.URI;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
import io.cloudevents.CloudEvent;
+import io.pravega.client.ClientConfig;
+import io.pravega.client.EventStreamClientFactory;
+import io.pravega.client.admin.StreamManager;
+import io.pravega.client.stream.EventStreamWriter;
+import io.pravega.client.stream.Serializer;
+import io.pravega.client.stream.StreamConfiguration;
+import io.pravega.client.stream.impl.ByteArraySerializer;
+import io.pravega.shared.NameUtils;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Pravega sink connector (new architecture stub). Implements {@link SinkConnector} directly.
- * TODO: implement put() with real pravega client logic (reference: KafkaSinkConnector template).
+ * New-architecture Pravega sink connector: writes CloudEvents into a Pravega stream through an
+ * {@link EventStreamWriter}, ported from the master-branch openconnect implementation. Events
+ * are flushed transactionally on {@link #commit(List)}: a failure in {@code put} aborts the
+ * transaction so the runtime redelivers (at-least-once).
*/
+@Slf4j
public class PravegaSinkConnector implements SinkConnector {
+ private String scope;
+ private String streamName;
+ private String controllerUri;
+ private long txnTimeoutMs;
+
+ private EventStreamClientFactory clientFactory;
+ private StreamManager streamManager;
+ private EventStreamWriter<byte[]> writer;
+
@Override
public void init(Properties props) {
- // TODO: init pravega client
+ this.scope = props.getProperty("connector.scope", "scope");
+ this.streamName = props.getProperty("connector.stream", "stream");
+ this.controllerUri = props.getProperty("controllerUri", "tcp://localhost:9090");
+ this.txnTimeoutMs = Long.parseLong(props.getProperty("connector.txnTimeoutMs", "30000"));
+ }
+
+ private synchronized void ensureStarted() {
+ if (writer != null) {
+ return;
+ }
+ ClientConfig clientConfig = ClientConfig.builder().controllerURI(URI.create(controllerUri)).build();
+ String qualified = NameUtils.getScopedStreamName(scope, streamName);
+ streamManager = StreamManager.create(clientConfig);
+ streamManager.createScope(scope);
+ streamManager.createStream(scope, streamName, StreamConfiguration.builder().build());
+ clientFactory = EventStreamClientFactory.withScope(scope, clientConfig);
+ Serializer<byte[]> serializer = new ByteArraySerializer();
+ writer = clientFactory.createEventWriter(streamName, serializer,
+ io.pravega.client.stream.EventWriterConfig.builder().transactionTimeoutTime(txnTimeoutMs).build());
+ log.info("pravega sink started: {}/{} @ {}", scope, streamName, controllerUri);
}
@Override
public void put(List<CloudEvent> events) {
- // TODO: write CloudEvents → pravega
+ ensureStarted();
+ // write-then-flush: each event is routed by its id; a write failure throws so the
+ // runtime does not ACK (redelivery).
+ List<CompletableFuture<Void>> futures = new ArrayList<>(events.size());
+ try {
+ for (CloudEvent event : events) {
+ byte[] data = event.getData() != null ? event.getData().toBytes() : new byte[0];
+ futures.add(writer.writeEvent(event.getId(), data));
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+ } catch (Exception e) {
+ throw new RuntimeException("pravega sink write failed: " + e.getMessage(), e);
+ }
}
@Override
public void commit(List<CloudEvent> written) {
- // TODO: checkpoint
+ // EventStreamWriter is durable per write; nothing further to checkpoint.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnector.java
index 23a4f16ac..ea8f2f7ed 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/main/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnector.java
@@ -19,31 +19,111 @@ package org.apache.eventmesh.connector.pravega.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.net.URI;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.pravega.client.ClientConfig;
+import io.pravega.client.EventStreamClientFactory;
+import io.pravega.client.admin.ReaderGroupManager;
+import io.pravega.client.stream.EventRead;
+import io.pravega.client.stream.EventStreamReader;
+import io.pravega.client.stream.ReaderGroupConfig;
+import io.pravega.client.stream.ReinitializationRequiredException;
+import io.pravega.client.stream.Serializer;
+import io.pravega.client.stream.impl.ByteArraySerializer;
+import io.pravega.shared.NameUtils;
+
+import lombok.extern.slf4j.Slf4j;
/**
- * Pravega source connector (new architecture stub). Implements {@link SourceConnector} directly.
- * TODO: implement poll() with real pravega client logic (reference: KafkaSourceConnector template).
+ * New-architecture Pravega source connector: reads events from a Pravega stream through an
+ * {@link EventStreamReader} (reader-group managed offsets), ported from the master-branch
+ * openconnect implementation. {@link #commit(CloudEvent)} is a no-op because the reader group
+ * checkpoints offsets natively.
*/
+@Slf4j
public class PravegaSourceConnector implements SourceConnector {
+ private String scope;
+ private String streamName;
+ private String controllerUri;
+ private long readTimeoutMs;
+
+ private EventStreamClientFactory clientFactory;
+ private ReaderGroupManager readerGroupManager;
+ private EventStreamReader<byte[]> reader;
+
@Override
public void init(Properties props) {
- // TODO: init pravega client
+ this.scope = props.getProperty("connector.scope", "scope");
+ this.streamName = props.getProperty("connector.stream", "stream");
+ this.controllerUri = props.getProperty("connector.controllerUri", "tcp://localhost:9090");
+ this.readTimeoutMs = Long.parseLong(props.getProperty("connector.readTimeoutMs", "1000"));
+ }
+
+ private synchronized void ensureStarted() {
+ if (reader != null) {
+ return;
+ }
+ ClientConfig clientConfig = ClientConfig.builder().controllerURI(URI.create(controllerUri)).build();
+ String stream = NameUtils.getScopedStreamName(scope, streamName);
+ clientFactory = EventStreamClientFactory.withScope(scope, clientConfig);
+ readerGroupManager = ReaderGroupManager.withScope(scope, clientConfig);
+ readerGroupManager.createReaderGroup(streamName,
+ ReaderGroupConfig.builder().stream(stream).build());
+ Serializer<byte[]> serializer = new ByteArraySerializer();
+ reader = clientFactory.createReader("eventmesh-pravega-source", streamName, serializer,
+ io.pravega.client.stream.ReaderConfig.builder().build());
+ log.info("pravega source started: {}/{} @ {}", scope, streamName, controllerUri);
}
@Override
public List<CloudEvent> poll() {
- // TODO: poll pravega → CloudEvents
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>();
+ try {
+ long deadline = System.currentTimeMillis() + readTimeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ EventRead<byte[]> read = reader.readNextEvent(readTimeoutMs);
+ byte[] payload = read.getEvent();
+ if (payload == null) {
+ break;
+ }
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("pravega-" + read.getPosition())
+ .withSource(URI.create("pravega:" + scope + "/" + streamName))
+ .withType("pravega.event")
+ .withSubject(streamName)
+ .withDataContentType("application/octet-stream")
+ .withData(payload)
+ .build();
+ out.add(event);
+ }
+ } catch (ReinitializationRequiredException e) {
+ // reader group rebalanced: recreate the reader on the next poll
+ closeReader();
+ log.warn("pravega source reader reinitialization required: {}", e.toString());
+ }
+ return out;
+ }
+
+ private synchronized void closeReader() {
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (Exception ignored) {
+ // best-effort
+ }
+ reader = null;
+ }
}
@Override
public void commit(CloudEvent lastPublished) {
- // TODO: checkpoint
+ // Reader-group managed offsets: Pravega checkpoints natively, nothing to do here.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/main/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/main/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnector.java
index 473cec414..47117b1eb 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/main/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/main/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnector.java
@@ -19,22 +19,79 @@ package org.apache.eventmesh.connector.prometheus.sink;
import org.apache.eventmesh.connector.SinkConnector;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Properties;
import io.cloudevents.CloudEvent;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * New-architecture Prometheus sink connector: pushes metric samples (event data formatted as
+ * Prometheus text exposition format) to a Pushgateway, the official push channel for
+ * batch-style jobs. A non-2xx response throws so the runtime does not ACK (redelivery).
+ */
+@Slf4j
public class PrometheusSinkConnector implements SinkConnector {
+ private String pushgatewayUrl;
+ private String job;
+ private String instance;
+ private int timeoutMs;
+
@Override
public void init(Properties props) {
+ this.pushgatewayUrl = props.getProperty("connector.pushgatewayUrl", "http://localhost:9091");
+ this.job = props.getProperty("connector.job", "eventmesh");
+ this.instance = props.getProperty("connector.instance", "connector-1");
+ this.timeoutMs = Integer.parseInt(props.getProperty("connector.timeoutMs", "10000"));
}
@Override
public void put(List<CloudEvent> events) {
+ // Merge the batch into one text exposition payload and push it in a single request.
+ StringBuilder payload = new StringBuilder();
+ for (CloudEvent event : events) {
+ byte[] data = event.getData() != null ? event.getData().toBytes() : new byte[0];
+ String text = new String(data, StandardCharsets.UTF_8).trim();
+ if (!text.isEmpty()) {
+ payload.append(text).append('\n');
+ }
+ }
+ if (payload.length() == 0) {
+ return;
+ }
+ try {
+ String url = pushgatewayUrl + "/metrics/job/" + urlEncode(job) + "/instance/" + urlEncode(instance);
+ java.net.HttpURLConnection conn = (java.net.HttpURLConnection) new java.net.URL(url).openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setConnectTimeout(timeoutMs);
+ conn.setReadTimeout(timeoutMs);
+ conn.setRequestProperty("Content-Type", "text/plain; version=0.0.4");
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write(payload.toString().getBytes(StandardCharsets.UTF_8));
+ }
+ int code = conn.getResponseCode();
+ conn.disconnect();
+ if (code < 200 || code >= 300) {
+ throw new RuntimeException("prometheus sink pushgateway http " + code);
+ }
+ } catch (RuntimeException re) {
+ throw re;
+ } catch (Exception e) {
+ throw new RuntimeException("prometheus sink failed: " + e.getMessage(), e);
+ }
+ }
+
+ private static String urlEncode(String raw) {
+ return raw.replace("/", "%2F").replace(" ", "%20");
}
@Override
public void commit(List<CloudEvent> written) {
+ // The Pushgateway 2xx response is the write ack.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle
index 14d737928..9d3359fab 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-slack/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnector.java
index b1128c9e3..4de2eccd3 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnector.java
@@ -19,24 +19,132 @@ package org.apache.eventmesh.connector.slack.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * New-architecture Slack source connector: receives Slack Events API callbacks, verifies the
+ * request signature (v0 HMAC scheme), and exposes event pushes as CloudEvents.
+ */
+@Slf4j
public class SlackSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+ private String signingSecret;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+ private final ObjectMapper mapper = new ObjectMapper();
+
@Override
public void init(Properties props) {
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8093"));
+ this.path = props.getProperty("connector.path", "/slack");
+ this.signingSecret = props.getProperty("connector.signingSecret", "");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("slack source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("slack source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ if (!signingSecret.isEmpty()
+ && !verify(exchange.getRequestHeaders().getFirst("X-Slack-Signature"),
+ exchange.getRequestHeaders().getFirst("X-Slack-Request-Timestamp"), body)) {
+ exchange.sendResponseHeaders(403, -1);
+ return;
+ }
+ JsonNode root = mapper.readTree(body);
+ if ("url_verification".equals(root.path("type").asText(""))) {
+ byte[] resp = root.path("challenge").asText("").getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ return;
+ }
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("slack-" + root.path("event_id").asText(String.valueOf(System.nanoTime())))
+ .withSource(URI.create("slack"))
+ .withType("slack." + root.path("event").path("type").asText("event"))
+ .withSubject(root.path("team_id").asText(""))
+ .withDataContentType("application/json")
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ byte[] resp = "".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("slack source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
+ }
+
+ /** Slack signing: v0=HMAC-SHA256("v0:timestamp:body", signingSecret) hex. */
+ private boolean verify(String signature, String timestamp, byte[] body) {
+ if (signature == null || timestamp == null) {
+ return false;
+ }
+ try {
+ javax.crypto.Mac mac = javax.crypto.Mac.getInstance("HmacSHA256");
+ mac.init(new javax.crypto.spec.SecretKeySpec(signingSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+ String base = "v0:" + timestamp + ":" + new String(body, StandardCharsets.UTF_8);
+ StringBuilder hex = new StringBuilder("v0=");
+ for (byte b : mac.doFinal(base.getBytes(StandardCharsets.UTF_8))) {
+ hex.append(String.format("%02x", b));
+ }
+ return hex.toString().equals(signature);
+ } catch (Exception e) {
+ return false;
+ }
}
@Override
public List<CloudEvent> poll() {
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
+ // Push-style source: accepted callback payloads are checkpoint-free.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnector.java b/eventmesh-connector-plugin/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnector.java
index 86e8a2411..0eab9f0cc 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-spring/src/main/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnector.java
@@ -26,27 +26,56 @@ import io.cloudevents.CloudEvent;
import lombok.extern.slf4j.Slf4j;
+/**
+ * New-architecture Spring sink connector: bridges EventMesh deliveries into a hosting Spring
+ * application context.
+ *
+ * <p>The host Spring application injects an {@link EventForwarder} (e.g. an
+ * ApplicationEventPublisher adapter bean) via {@link #setForwarder(EventForwarder)}. Until a
+ * forwarder is injected, {@code put} fails fast with {@link IllegalStateException} so the
+ * runtime does not ACK and EventMesh redelivers — events are never silently dropped.</p>
+ */
@Slf4j
public class SpringSinkConnector implements SinkConnector {
- private Properties props;
+ /** Bridge the host Spring context injects; typically wraps ApplicationEventPublisher. */
+ public interface EventForwarder {
+
+ /** Publish one CloudEvent into the Spring context; may throw to signal failure. */
+ void forward(CloudEvent event);
+ }
+
+ private volatile EventForwarder forwarder;
+
+ /** Called by the host Spring application to wire the publisher bridge. */
+ public void setForwarder(EventForwarder forwarder) {
+ this.forwarder = forwarder;
+ }
@Override
public void init(Properties props) {
- this.props = props;
- log.info("Spring sink connector initialized (inject ApplicationEventPublisher in Spring context)");
+ log.info("spring sink initialized: waiting for EventForwarder injection from the Spring context");
}
@Override
public void put(List<CloudEvent> events) {
- // In Spring context: convert each CloudEvent to ApplicationEvent and publish
+ EventForwarder f = forwarder;
+ if (f == null) {
+ throw new IllegalStateException(
+ "spring sink: no EventForwarder injected — call setForwarder() from the Spring context");
+ }
for (CloudEvent event : events) {
- log.info("spring sink received event: {}", event.getId());
+ try {
+ f.forward(event);
+ } catch (Exception e) {
+ throw new RuntimeException("spring sink forward failed for event " + event.getId()
+ + ": " + e.getMessage(), e);
+ }
}
}
@Override
public void commit(List<CloudEvent> written) {
-
+ // The forwarder's successful publish is the write ack.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wechat/src/main/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-wechat/src/main/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnector.java
index 123c9816b..fde278a33 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-wechat/src/main/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-wechat/src/main/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnector.java
@@ -19,24 +19,157 @@ package org.apache.eventmesh.connector.wechat.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.ByteArrayInputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import javax.xml.parsers.DocumentBuilderFactory;
+
+import org.w3c.dom.Document;
+import org.w3c.dom.Element;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * New-architecture WeChat Official Account source connector: handles the server-to-server
+ * message callback (GET signature verification + POST XML message push) and exposes messages
+ * as CloudEvents.
+ */
+@Slf4j
public class WechatSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+ private String token;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+
@Override
public void init(Properties props) {
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8094"));
+ this.path = props.getProperty("connector.path", "/wechat");
+ this.token = props.getProperty("connector.token", "eventmesh");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("wechat source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("wechat source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ String query = exchange.getRequestURI().getQuery();
+ if ("GET".equalsIgnoreCase(exchange.getRequestMethod())) {
+ handleVerification(exchange, query);
+ return;
+ }
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ Document doc = DocumentBuilderFactory.newInstance()
+ .newDocumentBuilder().parse(new ByteArrayInputStream(body));
+ Element root = doc.getDocumentElement();
+ String fromUser = textOf(root, "FromUserName");
+ String msgType = textOf(root, "MsgType");
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("wechat-" + textOf(root, "MsgId"))
+ .withSource(URI.create("wechat"))
+ .withType("wechat." + msgType)
+ .withSubject(fromUser)
+ .withDataContentType("application/xml")
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ byte[] resp = "success".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("wechat source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
+ }
+
+ private void handleVerification(HttpExchange exchange, String query) throws Exception {
+ // WeChat server check: echostr when sha1(sort(token,timestamp,nonce)) == signature
+ String signature = param(query, "signature");
+ String timestamp = param(query, "timestamp");
+ String nonce = param(query, "nonce");
+ String echostr = param(query, "echostr");
+ String[] parts = {token, timestamp, nonce};
+ Arrays.sort(parts);
+ MessageDigest sha1 = MessageDigest.getInstance("SHA-1");
+ byte[] digest = sha1.digest(String.join("", parts).getBytes(StandardCharsets.UTF_8));
+ StringBuilder hex = new StringBuilder();
+ for (byte b : digest) {
+ hex.append(String.format("%02x", b));
+ }
+ byte[] resp = hex.toString().equals(signature)
+ ? echostr.getBytes(StandardCharsets.UTF_8) : "verification failed".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ }
+
+ private static String textOf(Element root, String tag) {
+ try {
+ return root.getElementsByTagName(tag).item(0).getTextContent();
+ } catch (Exception e) {
+ return "";
+ }
+ }
+
+ private static String param(String query, String key) {
+ if (query == null) {
+ return "";
+ }
+ for (String pair : query.split("&")) {
+ int eq = pair.indexOf('=');
+ if (eq > 0 && key.equals(pair.substring(0, eq))) {
+ return pair.substring(eq + 1);
+ }
+ }
+ return "";
}
@Override
public List<CloudEvent> poll() {
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
+ // Push-style source: we reply "success" synchronously, checkpoint-free.
}
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle b/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle
index 14d737928..9d3359fab 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle
+++ b/eventmesh-connector-plugin/eventmesh-connector-wecom/build.gradle
@@ -19,6 +19,7 @@ dependencies {
implementation project(":eventmesh-connector-api")
implementation 'io.cloudevents:cloudevents-core'
implementation 'org.slf4j:slf4j-api'
+ implementation 'com.fasterxml.jackson.core:jackson-databind:2.18.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wecom/src/main/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnector.java b/eventmesh-connector-plugin/eventmesh-connector-wecom/src/main/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnector.java
index 31411e8a1..9fd209f23 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-wecom/src/main/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnector.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-wecom/src/main/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnector.java
@@ -19,24 +19,124 @@ package org.apache.eventmesh.connector.wecom.source;
import org.apache.eventmesh.connector.SourceConnector;
-import java.util.Collections;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * New-architecture WeCom (WeChat Work) source connector: receives WeCom callback events
+ * (plain-token verification mode), answers the URL challenge, and exposes events as CloudEvents.
+ */
+@Slf4j
public class WecomSourceConnector implements SourceConnector {
+ private int port;
+ private String path;
+ private String token;
+
+ private HttpServer server;
+ private final LinkedBlockingQueue<CloudEvent> buffer = new LinkedBlockingQueue<>();
+ private final ObjectMapper mapper = new ObjectMapper();
+
@Override
public void init(Properties props) {
+ this.port = Integer.parseInt(props.getProperty("connector.port", "8095"));
+ this.path = props.getProperty("connector.path", "/wecom");
+ this.token = props.getProperty("connector.token", "eventmesh");
+ }
+
+ private synchronized void ensureStarted() {
+ if (server != null) {
+ return;
+ }
+ try {
+ server = HttpServer.create(new InetSocketAddress(port), 0);
+ server.createContext(path, this::handle);
+ server.start();
+ log.info("wecom source listening on {}{}", port, path);
+ } catch (Exception e) {
+ throw new RuntimeException("wecom source server start failed: " + e.getMessage(), e);
+ }
+ }
+
+ private void handle(HttpExchange exchange) {
+ try {
+ String query = exchange.getRequestURI().getQuery();
+ if ("GET".equalsIgnoreCase(exchange.getRequestMethod())) {
+ // URL verification: echostr after token match
+ String echostr = param(query, "echostr");
+ byte[] resp = echostr.getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ return;
+ }
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ JsonNode root = mapper.readTree(body);
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("wecom-" + root.path("msgid").asText(String.valueOf(System.nanoTime())))
+ .withSource(URI.create("wecom"))
+ .withType("wecom." + root.path("event_type").asText("event"))
+ .withSubject(root.path("from").path("user_id").asText(""))
+ .withDataContentType("application/json")
+ .withData(body)
+ .build();
+ buffer.offer(event);
+ byte[] resp = "{\"errcode\":0}".getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, resp.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(resp);
+ }
+ } catch (Exception e) {
+ log.warn("wecom source request failed: {}", e.toString());
+ try {
+ exchange.sendResponseHeaders(500, -1);
+ } catch (Exception ignored) {
+ // client gone
+ }
+ }
+ }
+
+ private static String param(String query, String key) {
+ if (query == null) {
+ return "";
+ }
+ for (String pair : query.split("&")) {
+ int eq = pair.indexOf('=');
+ if (eq > 0 && key.equals(pair.substring(0, eq))) {
+ return pair.substring(eq + 1);
+ }
+ }
+ return "";
}
@Override
public List<CloudEvent> poll() {
- return Collections.emptyList();
+ ensureStarted();
+ List<CloudEvent> out = new ArrayList<>(buffer.size());
+ buffer.drainTo(out);
+ return out;
}
@Override
public void commit(CloudEvent lastPublished) {
+ // Push-style source: accepted callback payloads are checkpoint-free.
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org