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 09:36:09 UTC
(eventmesh) branch develop updated: fix(connector): complete the 16 incomplete connector implementations (#5394)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new 62d51db48 fix(connector): complete the 16 incomplete connector implementations (#5394)
62d51db48 is described below
commit 62d51db4857ab0568630a0c1e3512a7be0e95fcb
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Fri Sep 11 17:36:03 2026 +0800
fix(connector): complete the 16 incomplete connector implementations (#5394)
* 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.
* docs(connector): add per-plugin guides for all 23 connectors
Overview (model, running, catalog, matrix) + one page per plugin
(classes, source/sink behavior, config key tables from the actual
props.getProperty defaults, run examples). Replaces the master-era
website 03-connect docs (9 plugins) with full 23-plugin coverage
matching the new-architecture contract.
---
docs/feature/connectors/README.md | 140 +++++++++++++++++++++
docs/feature/connectors/_category_.json | 5 +
docs/feature/connectors/canal.md | 39 ++++++
docs/feature/connectors/chatgpt.md | 34 +++++
docs/feature/connectors/dingtalk.md | 32 +++++
docs/feature/connectors/file.md | 30 +++++
docs/feature/connectors/http.md | 31 +++++
docs/feature/connectors/jdbc.md | 33 +++++
docs/feature/connectors/kafka.md | 34 +++++
docs/feature/connectors/knative.md | 31 +++++
docs/feature/connectors/lark.md | 32 +++++
docs/feature/connectors/mcp.md | 33 +++++
docs/feature/connectors/mongodb.md | 34 +++++
docs/feature/connectors/openfunction.md | 32 +++++
docs/feature/connectors/pravega.md | 36 ++++++
docs/feature/connectors/prometheus.md | 33 +++++
docs/feature/connectors/pulsar.md | 32 +++++
docs/feature/connectors/rabbitmq.md | 35 ++++++
docs/feature/connectors/redis.md | 32 +++++
docs/feature/connectors/rocketmq.md | 34 +++++
docs/feature/connectors/s3.md | 31 +++++
docs/feature/connectors/slack.md | 32 +++++
docs/feature/connectors/spring.md | 30 +++++
docs/feature/connectors/wechat.md | 32 +++++
docs/feature/connectors/wecom.md | 32 +++++
docs/index.md | 4 +
.../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 ++++++++++++++-
50 files changed, 2191 insertions(+), 81 deletions(-)
diff --git a/docs/feature/connectors/README.md b/docs/feature/connectors/README.md
new file mode 100644
index 000000000..7f7e4d760
--- /dev/null
+++ b/docs/feature/connectors/README.md
@@ -0,0 +1,140 @@
+# Connectors
+
+**Audience:** operators and developers wiring EventMesh to external systems —
+message queues, databases, chat platforms, AI services. Covers the connector
+model, the 23 shipped plugins, how to configure and run them. For the API a
+plugin implements see [Connector API split plan](../connector-api-split.md);
+for runtime configuration of the connector scheduler see
+[Configuration reference](../../quickstart/configuration.md).
+
+---
+
+## The connector model
+
+A **connector** moves CloudEvents between EventMesh and an external system on
+one direction:
+
+- a **Source** pulls (or receives) data from the external system and the
+ connector runtime publishes it into EventMesh over HTTP;
+- a **Sink** long-polls EventMesh for deliveries and writes them into the
+ external system.
+
+The contract lives in `eventmesh-connector-api`:
+
+| Method | Source | Sink |
+| --- | --- | --- |
+| `init(Properties)` | configure from `-D` flags | configure from `-D` flags |
+| `resume(String)` / `poll()` | resume from an offset marker; pull the next batch | — |
+| `put(List<CloudEvent>)` | — | write a batch (throw on failure → no ACK → redelivery) |
+| `commit(...)` | checkpoint after EventMesh accepted the publish | checkpoint after the write |
+
+Delivery semantics are **at-least-once**: sources checkpoint only after the
+runtime accepted the publish, and a failing sink write leaves the delivery
+un-ACKed so EventMesh redelivers (dedup by event id downstream).
+
+## Running a connector
+
+Each plugin ships inside the connector-runtime image; `bin/start-connector.sh`
+starts a worker:
+
+```bash
+# CONNECTOR_OPTS carries the plugin class and every -D flag the plugin reads
+CONNECTOR_OPTS="-Dconnector.class=org.apache.eventmesh.connector.kafka.source.KafkaSourceConnector \
+ -Dconnector.mode=source \
+ -Dconnector.topic=my-topic \
+ -DbootstrapServers=broker:9092" \
+bin/start-connector.sh
+```
+
+Common flags: `eventmesh.runtime.url` (default `http://localhost:8080`),
+`connector.offset.mode` (`remote` | `rocksdb` | `inmemory`), and for multiple
+connectors per process the numbered form `-Dconnector.1.class=...`,
+`-Dconnector.2.class=...` (any `-Dconnector.N.*` key is passed through to the
+plugin's `init`). The runtime can also schedule connectors dynamically via
+`/admin/connectors` — see the [Admin API](../admin-api.md).
+
+## Plugin catalog
+
+**Message queues**
+
+- [Kafka connector](kafka.md)
+- [RocketMQ 4.x connector](rocketmq.md)
+- [RabbitMQ connector](rabbitmq.md)
+- [Pulsar connector](pulsar.md)
+- [Pravega connector](pravega.md)
+
+**Web & serverless**
+
+- [HTTP connector](http.md)
+- [Knative connector](knative.md)
+- [OpenFunction connector](openfunction.md)
+
+**Framework bridges**
+
+- [Spring connector](spring.md)
+
+**Databases**
+
+- [JDBC connector](jdbc.md)
+- [Canal (MySQL CDC) connector](canal.md)
+- [MongoDB connector](mongodb.md)
+
+**Storage**
+
+- [Amazon S3 connector](s3.md)
+- [File connector](file.md)
+- [Redis connector](redis.md)
+
+**Observability**
+
+- [Prometheus connector](prometheus.md)
+
+**AI & chat**
+
+- [ChatGPT (OpenAI) connector](chatgpt.md)
+- [MCP (Model Context Protocol) connector](mcp.md)
+
+**Chat & IM**
+
+- [DingTalk connector](dingtalk.md)
+- [Lark / Feishu connector](lark.md)
+- [Slack connector](slack.md)
+- [WeChat Official Account connector](wechat.md)
+- [WeCom (WeChat Work) connector](wecom.md)
+
+
+## Plugin matrix
+
+| Plugin | Source | Sink | Client dependency |
+| --- | :-: | :-: | --- |
+| [Kafka](kafka.md) | ✓ | ✓ | kafka-clients 3.9.0 |
+| [RocketMQ 4.x](rocketmq.md) | ✓ | ✓ | rocketmq-client |
+| [RabbitMQ](rabbitmq.md) | ✓ | ✓ | amqp-client 5.22.0 |
+| [Pulsar](pulsar.md) | ✓ | ✓ | pulsar-client |
+| [Pravega](pravega.md) | ✓ | ✓ | pravega-client 0.11.0 |
+| [HTTP](http.md) | ✓ | ✓ | — |
+| [Knative](knative.md) | ✓ | ✓ | — |
+| [OpenFunction](openfunction.md) | ✓ | ✓ | — |
+| [Spring](spring.md) | ✓ | ✓ | — |
+| [JDBC](jdbc.md) | ✓ | ✓ | — |
+| [Canal (MySQL CDC)](canal.md) | ✓ | ✓ | canal.client 1.1.7 |
+| [MongoDB](mongodb.md) | ✓ | ✓ | mongodb-driver-sync 4.11.0 |
+| [Amazon S3](s3.md) | ✓ | ✓ | aws-sdk-s3 |
+| [File](file.md) | ✓ | ✓ | — |
+| [Redis](redis.md) | ✓ | ✓ | redisson |
+| [Prometheus](prometheus.md) | ✓ | ✓ | — |
+| [ChatGPT (OpenAI)](chatgpt.md) | ✓ | ✓ | — |
+| [MCP (Model Context Protocol)](mcp.md) | ✓ | ✓ | — |
+| [DingTalk](dingtalk.md) | ✓ | ✓ | — |
+| [Lark / Feishu](lark.md) | ✓ | ✓ | — |
+| [Slack](slack.md) | ✓ | ✓ | — |
+| [WeChat Official Account](wechat.md) | ✓ | ✓ | — |
+| [WeCom (WeChat Work)](wecom.md) | ✓ | ✓ | — |
+
+## Adding a plugin
+
+Implement `SourceConnector` / `SinkConnector` from `eventmesh-connector-api`
+in a new `eventmesh-connector-plugin/eventmesh-connector-<name>` module,
+register the module in `settings.gradle`, and model the implementation on an
+existing plugin of the same style (pull-based like kafka/jdbc, push-receiver
+like http/dingtalk, or writer like rabbitmq/mongodb).
diff --git a/docs/feature/connectors/_category_.json b/docs/feature/connectors/_category_.json
new file mode 100644
index 000000000..e75d7f89d
--- /dev/null
+++ b/docs/feature/connectors/_category_.json
@@ -0,0 +1,5 @@
+{
+ "label": "Connectors",
+ "position": 9,
+ "collapsed": true
+}
diff --git a/docs/feature/connectors/canal.md b/docs/feature/connectors/canal.md
new file mode 100644
index 000000000..cd18a9bf6
--- /dev/null
+++ b/docs/feature/connectors/canal.md
@@ -0,0 +1,39 @@
+# Canal (MySQL CDC) connector
+
+**Audience:** operators bridging EventMesh with Canal (MySQL CDC). MySQL change-data-capture via a deployed canal server. Source consumes binlog entries over the canal TCP protocol (batch ack only after EventMesh accepted the publish); sink replays row-change CloudEvents into a target MySQL as one all-or-nothing JDBC batch.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.canal.source.CanalSourceConnector` | `CanalConnector.getWithoutAck(batchSize)` each poll; ROWDATA entries become CloudEvents (`subject` = `logfile:offset` position). `commit()` acks the pending batch — at-least-once. |
+| Sink | `org.apache.eventmesh.connector.canal.sink.CanalSinkConnector` | Executes each event's SQL (`subject`) in a single JDBC transaction; any failure rolls back and throws → no ACK → redelivery. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.canalHost` | `localhost` | Canal server host |
+| `connector.canalPort` | `11111` | Canal server port |
+| `connector.destination` | `example` | Canal destination (instance) |
+| `connector.username` | `` | Canal auth username |
+| `connector.password` | `` | Canal auth password |
+| `connector.subscribeFilter` | `.*\..*` | Binlog subscription filter (db.table regex) |
+| `connector.batchSize` | `100` | Max entries per canal batch |
+| `connector.pollTimeoutMs` | `1000` | Poll interval in ms |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.jdbcUrl` | `jdbc:mysql://localhost:3306/test` | Target MySQL JDBC URL |
+| `connector.dbUser` | `root` | Database user |
+| `connector.dbPassword` | `` | Database password |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...CanalSourceConnector -Dconnector.mode=source -Dconnector.canalHost=canal -Dconnector.destination=example -Dconnector.subscribeFilter=mydb\\..*
+```
diff --git a/docs/feature/connectors/chatgpt.md b/docs/feature/connectors/chatgpt.md
new file mode 100644
index 000000000..ac325c66d
--- /dev/null
+++ b/docs/feature/connectors/chatgpt.md
@@ -0,0 +1,34 @@
+# ChatGPT (OpenAI) connector
+
+**Audience:** operators bridging EventMesh with ChatGPT (OpenAI). OpenAI bridge. Source exposes an HTTP prompt endpoint: a client POSTs a prompt, the connector optionally completes it through the OpenAI REST API and emits both as a CloudEvent. Sink forwards events to a webhook (generic).
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.chatgpt.source.ChatgptSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path` accepts `{"prompt": ...}`; with `connector.openaiApiKey` set it calls chat-completions (`connector.model`) and emits `{prompt, answer}`; without a key it emits the prompt only. `poll()` drains the buffer. |
+| Sink | `org.apache.eventmesh.connector.chatgpt.sink.ChatgptSinkConnector` | POSTs each CloudEvent's payload to `connector.webhookUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8090` | HTTP listen port |
+| `connector.path` | `/chatgpt` | HTTP listen path |
+| `connector.openaiApiKey` | `` | OpenAI API key (empty = no completion) |
+| `connector.model` | `gpt-3.5-turbo` | Chat-completion model |
+| `connector.pollTimeoutMs` | `1000` | Buffer drain wait in ms |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.webhookUrl` | `` | Webhook to POST events into |
+
+## Running
+
+```bash
+curl -X POST http://worker:8090/chatgpt -d '{"prompt":"hi"}' then bin/start-connector.sh with -Dconnector.class=...ChatgptSourceConnector -Dconnector.mode=source -Dconnector.openaiApiKey=sk-...
+```
diff --git a/docs/feature/connectors/dingtalk.md b/docs/feature/connectors/dingtalk.md
new file mode 100644
index 000000000..9e2ba5a33
--- /dev/null
+++ b/docs/feature/connectors/dingtalk.md
@@ -0,0 +1,32 @@
+# DingTalk connector
+
+**Audience:** operators bridging EventMesh with DingTalk. DingTalk bridge. Source receives robot outgoing-callback messages (with HMAC signature verification); sink posts events to a DingTalk group-robot webhook.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.dingtalk.source.DingtalkSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path`; verifies `timestamp`+`sign` headers as base64(HMAC-SHA256(`timestamp + "\n" + appSecret`)) when `connector.appSecret` is set; accepted pushes become CloudEvents and `poll()` drains them. |
+| Sink | `org.apache.eventmesh.connector.dingtalk.sink.DingtalkSinkConnector` | POSTs each CloudEvent's payload to `connector.webhookUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8091` | HTTP listen port |
+| `connector.path` | `/dingtalk` | HTTP listen path |
+| `connector.appSecret` | `` | Robot app secret for signature verification (empty = skip verify) |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.webhookUrl` | `` | DingTalk robot webhook URL |
+
+## Running
+
+```bash
+Configure the robot's outgoing callback to http://worker:8091/dingtalk; bin/start-connector.sh with -Dconnector.class=...DingtalkSourceConnector -Dconnector.mode=source -Dconnector.appSecret=...
+```
diff --git a/docs/feature/connectors/file.md b/docs/feature/connectors/file.md
new file mode 100644
index 000000000..da7aab162
--- /dev/null
+++ b/docs/feature/connectors/file.md
@@ -0,0 +1,30 @@
+# File connector
+
+**Audience:** operators bridging EventMesh with File. Local file bridge — the simplest way to try connectors. Source reads a text file line by line; sink appends each CloudEvent as a line to a file.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.file.source.FileSourceConnector` | `BufferedReader.readLine()` per poll (a small batch each tick), tracking the read position in memory. |
+| Sink | `org.apache.eventmesh.connector.file.sink.FileSinkConnector` | Appends each CloudEvent (`id` + payload text) as a line via `PrintStream` (append mode). |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.filePath` | `/tmp/source.txt` | File to tail |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.filePath` | `/tmp/sink.txt` | File to append into |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...FileSourceConnector -Dconnector.mode=source -Dconnector.topic=lines -Dconnector.filePath=/var/log/app.log
+```
diff --git a/docs/feature/connectors/http.md b/docs/feature/connectors/http.md
new file mode 100644
index 000000000..8dd804559
--- /dev/null
+++ b/docs/feature/connectors/http.md
@@ -0,0 +1,31 @@
+# HTTP connector
+
+**Audience:** operators bridging EventMesh with HTTP. Generic HTTP bridge. Source is a webhook receiver: any system can POST events into EventMesh without an SDK. Sink forwards each CloudEvent to an HTTP endpoint.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.http.source.HttpSourceConnector` | JDK `HttpServer` (virtual threads) accepts POSTs on `connector.port`+`connector.path` into a buffer; `poll()` drains it as CloudEvents. |
+| Sink | `org.apache.eventmesh.connector.http.sink.HttpSinkConnector` | POSTs each CloudEvent's data to `connector.url`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8082` | Listen port for the webhook |
+| `connector.path` | `/webhook` | Listen path |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.url` | `http://localhost:9090/sink` | Target endpoint to POST into |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...HttpSourceConnector -Dconnector.mode=source -Dconnector.topic=incoming -Dconnector.port=8082
+```
diff --git a/docs/feature/connectors/jdbc.md b/docs/feature/connectors/jdbc.md
new file mode 100644
index 000000000..32705cbb9
--- /dev/null
+++ b/docs/feature/connectors/jdbc.md
@@ -0,0 +1,33 @@
+# JDBC connector
+
+**Audience:** operators bridging EventMesh with JDBC. Database bridge over plain JDBC. Source tails a table by a monotonically increasing id column; sink inserts each CloudEvent into a table via a parameterized INSERT.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.jdbc.source.JdbcSourceConnector` | Runs `connector.query` (with the last-seen id substituted for the `?`) each poll; each row (`id`, `data` columns) becomes a CloudEvent and advances the cursor. |
+| Sink | `org.apache.eventmesh.connector.jdbc.sink.JdbcSinkConnector` | Executes `connector.insertSql` per CloudEvent (id = event id, data = payload bytes). |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.jdbcUrl` | `jdbc:mysql://localhost:3306/test` | JDBC URL |
+| `connector.query` | `SELECT * FROM events WHERE id > ? ORDER BY id LIMIT 100` | Tail query (`?` = last id) |
+| `connector.lastId` | `0` | Initial cursor |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.jdbcUrl` | `jdbc:mysql://localhost:3306/test` | JDBC URL |
+| `connector.insertSql` | `INSERT INTO events (id, data) VALUES (?, ?)` | Insert statement |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...JdbcSourceConnector -Dconnector.mode=source -Dconnector.jdbcUrl=jdbc:mysql://db:3306/mydb
+```
diff --git a/docs/feature/connectors/kafka.md b/docs/feature/connectors/kafka.md
new file mode 100644
index 000000000..2dc8cca03
--- /dev/null
+++ b/docs/feature/connectors/kafka.md
@@ -0,0 +1,34 @@
+# Kafka connector
+
+**Audience:** operators bridging EventMesh with Kafka. Move events between EventMesh topics and an Apache Kafka cluster. The source consumes a Kafka topic with a consumer group and commits offsets only after EventMesh accepted the publish; the sink writes CloudEvent payloads to a target topic.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.kafka.source.KafkaSourceConnector` | Polls a source topic via `KafkaConsumer` (manual offset commit); each record becomes a CloudEvent (`id = topic-partition-offset`). `commit()` runs `commitSync` after the runtime accepted the batch. |
+| Sink | `org.apache.eventmesh.connector.kafka.sink.KafkaSinkConnector` | Produces each CloudEvent's data bytes to the target topic via `KafkaProducer`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `bootstrapServers` | `localhost:9092` | Kafka bootstrap servers |
+| `topic` | `source-topic` | Topic to consume |
+| `groupId` | `eventmesh-connector-source` | Consumer group id |
+| `pollTimeoutMs` | `1000` | Kafka poll timeout in ms |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `bootstrapServers` | `localhost:9092` | Kafka bootstrap servers |
+| `topic` | `sink-topic` | Target topic to produce into |
+
+## Running
+
+```bash
+bin/start-connector.sh with CONNECTOR_OPTS="-Dconnector.class=org.apache.eventmesh.connector.kafka.source.KafkaSourceConnector -Dconnector.mode=source -Dconnector.topic=my-topic -DbootstrapServers=broker:9092"
+```
diff --git a/docs/feature/connectors/knative.md b/docs/feature/connectors/knative.md
new file mode 100644
index 000000000..6706ed82e
--- /dev/null
+++ b/docs/feature/connectors/knative.md
@@ -0,0 +1,31 @@
+# Knative connector
+
+**Audience:** operators bridging EventMesh with Knative. Knative eventing bridge. Source receives CloudEvents pushed by a Knative broker (the source acts as the Knative subscriber); sink forwards events to a Knative service/endpoint.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.knative.source.KnativeSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path` buffers Knative-delivered CloudEvents; `poll()` drains the buffer. |
+| Sink | `org.apache.eventmesh.connector.knative.sink.KnativeSinkConnector` | POSTs each CloudEvent to `connector.sinkUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8080` | Listen port for broker deliveries |
+| `connector.path` | `/` | Listen path |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.sinkUrl` | `http://localhost:8080/sink` | Knative service endpoint |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...KnativeSourceConnector -Dconnector.mode=source -Dconnector.port=8080
+```
diff --git a/docs/feature/connectors/lark.md b/docs/feature/connectors/lark.md
new file mode 100644
index 000000000..55bece06a
--- /dev/null
+++ b/docs/feature/connectors/lark.md
@@ -0,0 +1,32 @@
+# Lark / Feishu connector
+
+**Audience:** operators bridging EventMesh with Lark / Feishu. Lark (Feishu) bridge. Source receives event-subscription callbacks (answering the `url_verification` challenge automatically); sink posts events to a Lark custom-bot webhook.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.lark.source.LarkSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path`; answers `url_verification` by echoing the challenge, verifies the callback `token` against `connector.verificationToken` when set, and emits event pushes as CloudEvents. |
+| Sink | `org.apache.eventmesh.connector.lark.sink.LarkSinkConnector` | POSTs each CloudEvent's payload to `connector.webhookUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8092` | HTTP listen port |
+| `connector.path` | `/lark` | HTTP listen path |
+| `connector.verificationToken` | `` | Event-subscription verification token (empty = skip verify) |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.webhookUrl` | `` | Lark custom-bot webhook URL |
+
+## Running
+
+```bash
+Set the app event-subscription request URL to http://worker:8092/lark; bin/start-connector.sh with -Dconnector.class=...LarkSourceConnector -Dconnector.mode=source
+```
diff --git a/docs/feature/connectors/mcp.md b/docs/feature/connectors/mcp.md
new file mode 100644
index 000000000..a17880bff
--- /dev/null
+++ b/docs/feature/connectors/mcp.md
@@ -0,0 +1,33 @@
+# MCP (Model Context Protocol) connector
+
+**Audience:** operators bridging EventMesh with MCP (Model Context Protocol). MCP server bridge over JSON-RPC 2.0. Source receives server-initiated notifications (e.g. resource updates) on an HTTP endpoint; sink forwards each CloudEvent to an MCP server as a JSON-RPC request.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.mcp.source.McpSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path` accepts JSON-RPC notifications; each becomes a CloudEvent (`type` = `mcp.<method>`) and `poll()` drains them. |
+| Sink | `org.apache.eventmesh.connector.mcp.sink.McpSinkConnector` | POSTs a JSON-RPC 2.0 request per CloudEvent (`method` = `connector.method`, `params` = event payload) to `connector.mcpServerUrl`; non-2xx throws → no ACK → redelivery. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8096` | HTTP listen port |
+| `connector.path` | `/mcp` | HTTP listen path |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.mcpServerUrl` | `http://localhost:3333/mcp` | MCP server endpoint |
+| `connector.method` | `eventmesh.notify` | JSON-RPC method to invoke |
+| `connector.timeoutMs` | `10000` | Request timeout in ms |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...McpSinkConnector -Dconnector.mode=sink -Dconnector.mcpServerUrl=http://mcp:3333/mcp
+```
diff --git a/docs/feature/connectors/mongodb.md b/docs/feature/connectors/mongodb.md
new file mode 100644
index 000000000..47761c98b
--- /dev/null
+++ b/docs/feature/connectors/mongodb.md
@@ -0,0 +1,34 @@
+# MongoDB connector
+
+**Audience:** operators bridging EventMesh with MongoDB. MongoDB bridge. Source tails a collection's insert stream; sink inserts each CloudEvent as a document.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.mongodb.source.MongodbSourceConnector` | Watches the collection (`watch()` cursor) and buffers change documents for `poll()`. |
+| Sink | `org.apache.eventmesh.connector.mongodb.sink.MongodbSinkConnector` | Inserts a document per CloudEvent (`_id` = event id, `data` = payload). |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.mongoUri` | `mongodb://localhost:27017` | MongoDB connection URI |
+| `connector.database` | `test` | Database |
+| `connector.collection` | `events` | Collection to watch |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.mongoUri` | `mongodb://localhost:27017` | MongoDB connection URI |
+| `connector.database` | `test` | Database |
+| `connector.collection` | `sink` | Target collection |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...MongodbSourceConnector -Dconnector.mode=source -Dconnector.mongoUri=mongodb://mongo:27017 -Dconnector.database=events
+```
diff --git a/docs/feature/connectors/openfunction.md b/docs/feature/connectors/openfunction.md
new file mode 100644
index 000000000..29cb24689
--- /dev/null
+++ b/docs/feature/connectors/openfunction.md
@@ -0,0 +1,32 @@
+# OpenFunction connector
+
+**Audience:** operators bridging EventMesh with OpenFunction. OpenFunction bridge. Source receives function output pushes (the function runtime POSTs its result to the connector endpoint); sink invokes a function's HTTP trigger with CloudEvent headers (`Ce-Id`/`Ce-Type`/`Ce-Source`).
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.openfunction.source.OpenfunctionSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path` receives function outputs (`X-Function-Name` header → CloudEvent subject); `poll()` drains. |
+| Sink | `org.apache.eventmesh.connector.openfunction.sink.OpenfunctionSinkConnector` | POSTs each CloudEvent (data + Ce-* headers) to `connector.functionUrl`; non-2xx throws → no ACK → redelivery. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8097` | Listen port for function outputs |
+| `connector.path` | `/openfunction` | Listen path |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.functionUrl` | `http://localhost:8081/function` | Function HTTP trigger URL |
+| `connector.timeoutMs` | `30000` | Connect/read timeout in ms |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...OpenfunctionSinkConnector -Dconnector.mode=sink -Dconnector.functionUrl=http://fn.default.svc.cluster.local:8080
+```
diff --git a/docs/feature/connectors/pravega.md b/docs/feature/connectors/pravega.md
new file mode 100644
index 000000000..e86fcbc87
--- /dev/null
+++ b/docs/feature/connectors/pravega.md
@@ -0,0 +1,36 @@
+# Pravega connector
+
+**Audience:** operators bridging EventMesh with Pravega. Move events between EventMesh and a Pravega stream. The source reads through a reader group (offsets managed natively by Pravega); the sink appends events to a stream, creating the scope/stream on first start.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.pravega.source.PravegaSourceConnector` | `EventStreamReader` reads up to the read-timeout each poll; a `ReinitializationRequiredException` (reader-group rebalance) closes the reader and recreates it on the next poll. |
+| Sink | `org.apache.eventmesh.connector.pravega.sink.PravegaSinkConnector` | `EventStreamWriter.writeEvent(routingKey=id, bytes)` for each CloudEvent; the batch is joined before returning (throw on failure → no ACK → redelivery). |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.scope` | `scope` | Pravega scope |
+| `connector.stream` | `stream` | Stream to read |
+| `connector.controllerUri` | `tcp://localhost:9090` | Controller URI |
+| `connector.readTimeoutMs` | `1000` | Per-poll read window in ms |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.scope` | `scope` | Pravega scope (created on start) |
+| `connector.stream` | `stream` | Target stream (created on start) |
+| `connector.controllerUri` | `tcp://localhost:9090` | Controller URI |
+| `connector.txnTimeoutMs` | `30000` | Writer transaction timeout in ms |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...PravegaSourceConnector -Dconnector.mode=source -Dconnector.scope=my-scope -Dconnector.stream=my-stream
+```
diff --git a/docs/feature/connectors/prometheus.md b/docs/feature/connectors/prometheus.md
new file mode 100644
index 000000000..29690b5a8
--- /dev/null
+++ b/docs/feature/connectors/prometheus.md
@@ -0,0 +1,33 @@
+# Prometheus connector
+
+**Audience:** operators bridging EventMesh with Prometheus. Metrics bridge. Source scrapes a Prometheus `/metrics` endpoint on each poll and emits the exposition text as a CloudEvent; sink pushes metric samples (text exposition format) to a Pushgateway.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.prometheus.source.PrometheusSourceConnector` | GETs `connector.metricsUrl` each poll; the response body becomes one CloudEvent. |
+| Sink | `org.apache.eventmesh.connector.prometheus.sink.PrometheusSinkConnector` | Merges the batch's payloads into one text-exposition document and POSTs it to the Pushgateway; non-2xx throws → no ACK → redelivery. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.metricsUrl` | `http://localhost:9090/metrics` | Metrics endpoint to scrape |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.pushgatewayUrl` | `http://localhost:9091` | Pushgateway base URL |
+| `connector.job` | `eventmesh` | Pushgateway job label |
+| `connector.instance` | `connector-1` | Pushgateway instance label |
+| `connector.timeoutMs` | `10000` | Push timeout in ms |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...PrometheusSinkConnector -Dconnector.mode=sink -Dconnector.pushgatewayUrl=http://pg:9091 -Dconnector.job=eventmesh
+```
diff --git a/docs/feature/connectors/pulsar.md b/docs/feature/connectors/pulsar.md
new file mode 100644
index 000000000..15af19db8
--- /dev/null
+++ b/docs/feature/connectors/pulsar.md
@@ -0,0 +1,32 @@
+# Pulsar connector
+
+**Audience:** operators bridging EventMesh with Pulsar. Bridge EventMesh topics with an Apache Pulsar cluster. Source subscribes to a topic; sink produces to a persistent topic.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.pulsar.source.PulsarSourceConnector` | Subscribes via `PulsarClient` consumer (bytes schema) and buffers messages for `poll()`. |
+| Sink | `org.apache.eventmesh.connector.pulsar.sink.PulsarSinkConnector` | Produces each CloudEvent's data bytes via a Pulsar bytes producer. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.serviceUrl` | `pulsar://localhost:6650` | Pulsar service URL |
+| `connector.topic` | `persistent://public/default/source` | Topic to subscribe |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.serviceUrl` | `pulsar://localhost:6650` | Pulsar service URL |
+| `connector.topic` | `persistent://public/default/sink` | Target topic |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...PulsarSourceConnector -Dconnector.mode=source -Dconnector.topic=persistent://public/default/my-topic
+```
diff --git a/docs/feature/connectors/rabbitmq.md b/docs/feature/connectors/rabbitmq.md
new file mode 100644
index 000000000..ea91cddaa
--- /dev/null
+++ b/docs/feature/connectors/rabbitmq.md
@@ -0,0 +1,35 @@
+# RabbitMQ connector
+
+**Audience:** operators bridging EventMesh with RabbitMQ. Move events between EventMesh and a RabbitMQ broker over AMQP 0-9-1. Source consumes a queue; sink publishes to an exchange with an optional routing key.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.rabbitmq.source.RabbitmqSourceConnector` | `basicConsume` on the configured queue (auto-ack) into an internal buffer drained by `poll()`. |
+| Sink | `org.apache.eventmesh.connector.rabbitmq.sink.RabbitmqSinkConnector` | `basicPublish` each CloudEvent's data to the exchange with the routing key. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.host` | `localhost` | AMQP host |
+| `connector.port` | `5672` | AMQP port |
+| `connector.queue` | `source` | Queue to consume |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.host` | `localhost` | AMQP host |
+| `connector.port` | `5672` | AMQP port |
+| `connector.exchange` | `` | Target exchange (empty = default) |
+| `connector.routingKey` | `` | Routing key |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...RabbitmqSinkConnector -Dconnector.mode=sink -Dconnector.clientId=c1 -Dconnector.exchange=amq.topic -Dconnector.routingKey=events
+```
diff --git a/docs/feature/connectors/redis.md b/docs/feature/connectors/redis.md
new file mode 100644
index 000000000..27fbbcd3f
--- /dev/null
+++ b/docs/feature/connectors/redis.md
@@ -0,0 +1,32 @@
+# Redis connector
+
+**Audience:** operators bridging EventMesh with Redis. Redis pub/sub bridge over Redisson. Source listens on a channel; sink publishes each CloudEvent to a channel.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.redis.source.RedisSourceConnector` | Redisson topic listener buffers channel messages; `poll()` drains them as CloudEvents. |
+| Sink | `org.apache.eventmesh.connector.redis.sink.RedisSinkConnector` | Publishes each CloudEvent's payload to the Redis topic. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.redisUrl` | `redis://localhost:6379` | Redis address |
+| `connector.topic` | `source` | Channel to subscribe |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.redisUrl` | `redis://localhost:6379` | Redis address |
+| `connector.topic` | `sink` | Channel to publish |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...RedisSourceConnector -Dconnector.mode=source -Dconnector.redisUrl=redis://redis:6379 -Dconnector.topic=events
+```
diff --git a/docs/feature/connectors/rocketmq.md b/docs/feature/connectors/rocketmq.md
new file mode 100644
index 000000000..94bb71b8c
--- /dev/null
+++ b/docs/feature/connectors/rocketmq.md
@@ -0,0 +1,34 @@
+# RocketMQ 4.x connector
+
+**Audience:** operators bridging EventMesh with RocketMQ 4.x. Bridge EventMesh topics with an Apache RocketMQ 4.x cluster (nameserver-based). Source pulls from a consumer group; sink produces with a producer group.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.rocketmq.source.RocketmqSourceConnector` | Consumes a topic via `DefaultMQPushConsumer`; messages buffer into an internal queue that `poll()` drains. |
+| Sink | `org.apache.eventmesh.connector.rocketmq.sink.RocketmqSinkConnector` | Sends each CloudEvent via `DefaultMQProducer` to the configured topic. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.namesrvAddr` | `localhost:9876` | Name server address |
+| `connector.topic` | `source-topic` | Topic to consume |
+| `connector.group` | `connector-source` | Consumer group name |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.namesrvAddr` | `localhost:9876` | Name server address |
+| `connector.group` | `connector-sink` | Producer group name |
+| `connector.topic` | `sink-topic` | Target topic (set via producer topic config) |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...RocketmqSourceConnector -Dconnector.mode=source -Dconnector.topic=my-topic -Dconnector.namesrvAddr=ns:9876
+```
diff --git a/docs/feature/connectors/s3.md b/docs/feature/connectors/s3.md
new file mode 100644
index 000000000..63ec22d37
--- /dev/null
+++ b/docs/feature/connectors/s3.md
@@ -0,0 +1,31 @@
+# Amazon S3 connector
+
+**Audience:** operators bridging EventMesh with Amazon S3. S3 object bridge. Source lists new objects in a bucket (cursor = last key); sink writes each CloudEvent as an object keyed by the event id.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.s3.source.S3SourceConnector` | `ListObjectsV2` with `startAfter(lastKey)` each poll; new objects are fetched and emitted as CloudEvents, advancing the cursor. |
+| Sink | `org.apache.eventmesh.connector.s3.sink.S3SinkConnector` | Puts each CloudEvent's payload as an object (`id` = key) into the target bucket. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.bucket` | `events` | Source bucket |
+| `connector.prefix` | `` | Key prefix filter |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.bucket` | `sink` | Target bucket |
+
+## Running
+
+```bash
+bin/start-connector.sh with -Dconnector.class=...S3SourceConnector -Dconnector.mode=source -Dconnector.bucket=incoming -Dconnector.prefix=2026/
+```
diff --git a/docs/feature/connectors/slack.md b/docs/feature/connectors/slack.md
new file mode 100644
index 000000000..98fdce331
--- /dev/null
+++ b/docs/feature/connectors/slack.md
@@ -0,0 +1,32 @@
+# Slack connector
+
+**Audience:** operators bridging EventMesh with Slack. Slack bridge. Source receives Events API callbacks (verifying the v0 HMAC request signature and answering the challenge); sink posts events to a Slack incoming webhook.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.slack.source.SlackSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path`; when `connector.signingSecret` is set, verifies `X-Slack-Signature` (v0 = HMAC-SHA256 of `v0:timestamp:body`) + timestamp; answers `url_verification` challenges; event pushes become CloudEvents. |
+| Sink | `org.apache.eventmesh.connector.slack.sink.SlackSinkConnector` | POSTs each CloudEvent's payload to `connector.webhookUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8093` | HTTP listen port |
+| `connector.path` | `/slack` | HTTP listen path |
+| `connector.signingSecret` | `` | Slack app signing secret (empty = skip verify) |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.webhookUrl` | `` | Slack incoming webhook URL |
+
+## Running
+
+```bash
+Point the Slack app's Events API request URL at http://worker:8093/slack; bin/start-connector.sh with -Dconnector.class=...SlackSourceConnector -Dconnector.mode=source -Dconnector.signingSecret=...
+```
diff --git a/docs/feature/connectors/spring.md b/docs/feature/connectors/spring.md
new file mode 100644
index 000000000..9b9d8c387
--- /dev/null
+++ b/docs/feature/connectors/spring.md
@@ -0,0 +1,30 @@
+# Spring connector
+
+**Audience:** operators bridging EventMesh with Spring. Spring application bridge. Source buffers events published by Spring code (register a producer that feeds the buffer); sink exposes an injectable `EventForwarder` so the hosting Spring context receives EventMesh deliveries as native events.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.spring.source.SpringSourceConnector` | A `LinkedBlockingQueue` feeds `poll()`; host Spring code (e.g. an `@EventListener` adapter) enqueues CloudEvents. |
+| Sink | `org.apache.eventmesh.connector.spring.sink.SpringSinkConnector` | Calls the injected `EventForwarder.forward(event)` per CloudEvent. **Fail-fast**: until the Spring context wires a forwarder via `setForwarder(...)`, `put()` throws so the runtime redelivers instead of silently dropping events. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `(none)` | `—` | The source needs no external config |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `(none)` | `—` | Wire the forwarder programmatically from the Spring context |
+
+## Running
+
+```bash
+Host inside a Spring Boot app; call SpringSinkConnector.setForwarder(event -> publisher.publishEvent(...)) after init
+```
diff --git a/docs/feature/connectors/wechat.md b/docs/feature/connectors/wechat.md
new file mode 100644
index 000000000..64a772100
--- /dev/null
+++ b/docs/feature/connectors/wechat.md
@@ -0,0 +1,32 @@
+# WeChat Official Account connector
+
+**Audience:** operators bridging EventMesh with WeChat Official Account. WeChat Official Account bridge. Source handles the platform's server-to-server callback — the GET echostr verification (SHA-1 of token/timestamp/nonce) and POST XML message pushes; sink posts events to a webhook.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.wechat.source.WechatSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path`; GET answers the server verification (echostr when `sha1(sort(token,timestamp,nonce))` matches); POST parses the XML message (FromUserName/MsgType/MsgId) into a CloudEvent and replies `success`. |
+| Sink | `org.apache.eventmesh.connector.wechat.sink.WechatSinkConnector` | POSTs each CloudEvent's payload to `connector.webhookUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8094` | HTTP listen port |
+| `connector.path` | `/wechat` | HTTP listen path |
+| `connector.token` | `eventmesh` | Server-verification token configured in the WeChat admin console |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.webhookUrl` | `` | Webhook to POST events into |
+
+## Running
+
+```bash
+Set the account's server URL to http://worker:8094/wechat with the matching token; bin/start-connector.sh with -Dconnector.class=...WechatSourceConnector -Dconnector.mode=source
+```
diff --git a/docs/feature/connectors/wecom.md b/docs/feature/connectors/wecom.md
new file mode 100644
index 000000000..dd2deda84
--- /dev/null
+++ b/docs/feature/connectors/wecom.md
@@ -0,0 +1,32 @@
+# WeCom (WeChat Work) connector
+
+**Audience:** operators bridging EventMesh with WeCom (WeChat Work). WeCom bridge. Source receives WeCom callback events (plain-token verification mode: answers the GET echostr challenge, accepts POST JSON events); sink posts events to a WeCom group-robot webhook.
+
+---
+
+## Classes
+
+| Direction | Class | Behavior |
+| --- | --- | --- |
+| Source | `org.apache.eventmesh.connector.wecom.source.WecomSourceConnector` | JDK `HttpServer` on `connector.port`+`connector.path`; GET echoes the `echostr` (URL challenge); POST parses the JSON event (`event_type`, `from.user_id`, `msgid`) into a CloudEvent and replies `{"errcode":0}`. |
+| Sink | `org.apache.eventmesh.connector.wecom.sink.WecomSinkConnector` | POSTs each CloudEvent's payload to `connector.webhookUrl`. |
+
+## Source configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.port` | `8095` | HTTP listen port |
+| `connector.path` | `/wecom` | HTTP listen path |
+| `connector.token` | `eventmesh` | Callback token configured in the WeCom admin console |
+
+## Sink configuration
+
+| Key | Default | Description |
+| --- | --- | --- |
+| `connector.webhookUrl` | `` | WeCom group-robot webhook URL |
+
+## Running
+
+```bash
+Set the app's receive-message URL to http://worker:8095/wecom; bin/start-connector.sh with -Dconnector.class=...WecomSourceConnector -Dconnector.mode=source
+```
diff --git a/docs/index.md b/docs/index.md
index 8b42d6e69..427836005 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -50,6 +50,10 @@ too.
task lifecycle, JSON-RPC/MCP bridge, REST + SDK usage
*(Experimental)*; [readiness decision](feature/a2a-readiness.md).
+- [Connectors](feature/connectors/README.md) — bridge EventMesh with external
+ systems (Kafka, RocketMQ, RabbitMQ, databases, chat platforms): the connector
+ model, running a worker, and a per-plugin guide for all 23 shipped plugins.
+
**Internals & mechanisms**
- [Offset management](feature/offset-management.md) — the two offset layers,
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