[WIP] Refactor(rust-sdk): rewrite and align with java sdk

eventmesh-sdks/eventmesh-sdk-rust/AGENTS.md

@@ -0,0 +1,152 @@
+# AGENTS.md — EventMesh Rust SDK
+
+Cargo crate `eventmesh` (`edition = "2021"`, **MSRV 1.75.0**). Speaks the
+EventMesh **gRPC**, **HTTP**, and **TCP** protocols. gRPC wire format is
+CloudEvents-protobuf; the simple `EventMeshMessage` model is converted at the
+gRPC boundary by `codec.rs`. HTTP wire format is
+`application/x-www-form-urlencoded` with JSON payloads in the `content` field,
+mirroring the Java SDK. TCP uses the EventMesh binary wire protocol
+(length-prefixed frames with `"EventMesh"` magic) and supports automatic
+reconnect with exponential backoff plus CloudEvents JSON interop.
+
+This crate is **not** part of the Gradle build and has **no GitHub Actions
+CI**. All verification is local. The parent repo `AGENTS.md` covers the
+docker-compose profiles and runtime ports; this file covers the Rust workflow.
+
+## Build prerequisite: `protoc`
+
+`build.rs` invokes `tonic-build`, which shells out to the `protoc` compiler at
+**build time**. `prost-build` finds `protoc` automatically when it is on
+`PATH`; only set the `PROTOC` env var to point at a binary that is **not** on
+`PATH`. (`brew install protobuf`, `apt-get install protobuf-compiler`, etc. all
+put `protoc` on `PATH`, so no prefix is needed.)
+
+```bash
+cargo build --features full
+```
+
+## Feature matrix
+
+| Feature | Notes |
+|---|---|
+| `grpc` (default) | gRPC transport — publish, batch, request-reply, stream/webhook subscribe. |
+| `http` | HTTP transport — `HttpProducer`, `HttpConsumer`, webhook middleware + built-in `WebhookServer`. Uses reqwest + axum. |
+| `tcp` | TCP transport — `TcpProducer`, `TcpConsumer`, native binary wire protocol. Auto-reconnect with exponential backoff. CloudEvents interop behind `cloud_events`. |
+| `cloud_events` | Native `cloudevents::Event` interop (gRPC, HTTP, and TCP). |
+| `tls` | TLS on the gRPC channel. |
+| `full` | `grpc` + `http` + `tcp` + `cloud_events` + `tls`. Use this for clippy/test so every code path compiles. |
+| `e2e` | Gates the live-server integration suite (`tests/e2e/`). A plain `cargo test` never touches Docker. |
+
+## Verification (the order the README mandates)
+
+```bash
+cargo fmt
+cargo clippy --features full --all-targets -- -D warnings
+cargo test --features full
+```
+
+Clippy runs with **`-D warnings`** (warnings are errors here). There is no
+`rustfmt.toml`/`clippy.toml` — defaults apply. To run a single test binary:
+`cargo test --features full --test codec_test`.
+
+## Generated proto code (do not hand-edit)
+
+`build.rs` compiles `proto/eventmesh-{service,cloudevents}.proto` into `OUT_DIR`
+at build time, generating **client stubs only** (`build_server(false)`), with
+`--experimental_allow_proto3_optional`. The generated tree is pulled in via
+`src/proto_gen.rs` → `tonic::include_proto!("...")`; it is **not** checked in.
+Add convenience aliases in `proto_gen.rs`, not in the generated module.
+
+## Architecture notes
+
+- `src/lib.rs` is `#![deny(unsafe_code)]` — no `unsafe` anywhere.
+- `src/transport/mod.rs` defines `Publisher` as an **async-fn-in-trait**
+  (Rust 1.75). It is therefore **not object-safe** — use the concrete
+  `GrpcProducer` / `TcpProducer` / `HttpProducer` directly, never `dyn`.
+  The subscribe side has **no trait** — each transport exposes its own
+  consumer type (`GrpcStreamConsumer`, `GrpcWebhookConsumer`, `TcpConsumer`,
+  `HttpConsumer`) with transport-specific `subscribe` / `unsubscribe` /
+  `subscribe_webhook` methods, a background receive loop (where applicable),
+  and `wait_for_shutdown()` for clean exit.
+- `src/transport/grpc/codec.rs` is the `EventMeshMessage` ↔ CloudEvents-protobuf
+  bridge for the gRPC transport.
+- `src/transport/http/codec.rs` is the `EventMeshMessage` ↔ form-urlencoded +
+  JSON bridge for the HTTP transport. The wire format is
+  `application/x-www-form-urlencoded` (not JSON bodies); payloads go in the
+  `content` form field as JSON strings.
+- `src/config/grpc.rs` — `GrpcClientConfig` + fluent builder.
+- `src/config/http.rs` — `HttpClientConfig` + fluent builder. Accepts
+  semicolon/comma-separated `host:port[:weight]` server lists; uses the shared
+  `LoadBalanceSelector` from `common/loadbalance.rs`.
+- `src/transport/http/webhook.rs` — **internal** axum handler (`WebhookHandler`
+  + `WebhookState`) used only by `WebhookServer`. Not part of the public API.
+- `src/transport/http/server.rs` — built-in `WebhookServer` (axum) implementing
+  `IntoFuture` for one-liner `.await` startup with optional graceful shutdown.
+- `src/transport/tcp/connection.rs` — TCP engine: `establish()` does TCP +
+  HELLO handshake; `run()` is an outer reconnect loop that calls `io_loop()`
+  (the inner select! over read/write/heartbeat). When `ReconnectConfig::enabled`
+  (default `true`), the outer loop re-establishes the connection with
+  exponential backoff after I/O errors. `take_reconnect_rx()` delivers a
+  notification on each successful reconnect so consumers can replay
+  subscriptions.
+- `src/transport/tcp/message.rs` — package builders for `EventMeshMessage`
+  (`build_message_package`) and CloudEvents (`build_cloud_event_package`,
+  behind `cloud_events`). CloudEvents bodies use `protocoltype=cloudevents` and
+  are serialized as `application/cloudevents+json` raw bytes (matching the Java
+  runtime's codec path).
+- `src/config/tcp.rs` — `TcpClientConfig` + `ReconnectConfig` + fluent builders.
+- `src/common/` — `ProtocolKey`, status codes, constants, `LoadBalanceSelector`
+  shared across transports.
+
+### HTTP transport specifics
+
+- The HTTP consumer is **client-only** (like the Java SDK): it registers a
+  webhook URL with the runtime and sends heartbeats. The runtime POSTs messages
+  to that URL. The SDK provides two ways to receive those pushes:
+  1. **`WebhookServer`** — a built-in axum server (`IntoFuture` + graceful
+     shutdown) for users who don't want to manage their own HTTP server.
+  2. **Custom endpoint** — the user hosts any HTTP server (axum, actix, hyper,
+     …) and decodes pushes with the **public codec helpers** in
+     `src/transport/http/codec.rs`: `parse_push_body`,
+     `PushMessageRequestBody::to_event_mesh_message`, and `WebhookReply`
+     (the `{"retCode": 0}` ack). See the `http_consumer_custom` example.
+- There is intentionally **no public `WebhookHandler`/`WebhookLayer`/`WebhookState`**
+  type. The old "register the SDK's handler on your own Router" mode was
+  removed; users who embed the webhook in their own app write the handler
+  themselves on top of the codec utilities.
+- Runtime routing: the EventMesh HTTP server has two routing mechanisms —
+  path-based (new-style handlers, checked first by URI prefix match) and
+  code-header-based (old-style, checked by the `code` header when no path
+  matches). Because this SDK sends `application/x-www-form-urlencoded` bodies,
+  **all** operations (publish, subscribe, unsubscribe, heartbeat) use
+  code-header-based routing via the root path `/` (`uri::ROOT`). Posting to a
+  path-based handler (e.g. `/eventmesh/subscribe/local`) with a form body breaks
+  body decoding: the `topic` form field is parsed as a string and cannot be
+  deserialized as `List<SubscriptionItem>`.
+- Heartbeat interval: 30s (mirrors the Java SDK), spawned as a background
+  tokio task tied to a `CancellationToken`.
+
+## End-to-end tests (`tests/e2e/`)
+
+Run with: `cargo test --features e2e`.
+
+- The harness (`tests/e2e/runtime.rs`) **auto-starts the `rocketmq` docker-compose
+  profile** and tears it down at process exit. Set `EVENTMESH_E2E_EXTERNAL=1` to
+  point at a server you started yourself.
+- If neither Docker nor a reachable server is found, **tests skip, not fail**.
+- Each test generates a unique topic + consumer group (monotonic counter + nanos
+  timestamp), so the suite is parallel-safe on the shared broker.
+- **Standalone (in-memory) broker limitations:** a topic must be created via the
+  admin API **and** a consumer subscribed *before* any publish, and it does not
+  implement request/reply. The harness warms topics automatically and the
+  request/reply test self-skips its assertion on standalone — use the `rocketmq`
+  profile (the e2e default) to exercise the full path.
+- Topic creation hits the admin API at `POST /topic`, which expects
+  **`application/x-www-form-urlencoded`** (not JSON).
+
+## Conventions
+
+- **Apache license header is required on every `.rs` file** — copy the header
+  block from a neighbor (e.g. `build.rs`). Consistent with the rest of the repo.
+- Builders use the `Option<T> field + fluent setter + consuming build()` idiom
+  (see `GrpcClientConfigBuilder`); mirror it for new config types.

eventmesh-sdks/eventmesh-sdk-rust/Cargo.toml

@@ -1,86 +1,138 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
 #
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
+#   http://www.apache.org/licenses/LICENSE-2.0
 #
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
 
 [package]
 name = "eventmesh"
 version = "1.9.0"
 edition = "2021"
-authors = [
-    "mxsm <mxsm@apache.org>"
-]
-msrv = "1.75.0"
-description = "Rust client for Apache EventMesh"
+rust-version = "1.75.0"
+authors = ["Apache EventMesh <dev@eventmesh.apache.org>"]
 license = "Apache-2.0"
-keywords = ["EventMesh", "SDK", "rust-client", "rust", "eventmesh-rust-sdk"]
-readme = "./README.md"
-homepage = "https://github.com/apache/eventmesh"
+description = "Apache EventMesh Rust SDK"
+homepage = "https://eventmesh.apache.org"
 repository = "https://github.com/apache/eventmesh"
+keywords = ["eventmesh", "messaging", "cloud-events", "grpc"]
+categories = ["api-bindings", "network-programming"]
+
+[lib]
+name = "eventmesh"
+path = "src/lib.rs"
 
 [features]
-default = ["grpc", "eventmesh_message"]
-full = ["grpc", "eventmesh_message","cloud_events"]
-eventmesh_message = []
-cloud_events = []
-tls = []
-grpc = []
+default = ["grpc"]
+# Transport protocols (implemented incrementally).
+grpc = ["dep:tonic", "dep:prost", "dep:prost-types"]
+# HTTP transport (producer, consumer, webhook middleware + built-in server).
+http = ["dep:reqwest", "dep:serde_urlencoded", "dep:axum", "dep:tower", "dep:tower-http"]
+# TCP transport (producer, consumer, native TCP wire protocol).
+# `tokio-util/codec` is required by the framed codec in
+# `src/transport/tcp/{codec,connection}.rs`; without it the crate fails to
+# compile when `tcp` is enabled without `grpc` (tonic pulls in `codec` only as
+# a side effect of the `grpc` feature).
+tcp = ["tokio-util/codec"]
+# Message models.
+cloud_events = ["dep:cloudevents"]
+# TLS support.
+tls = ["tonic/tls", "tonic/tls-native-roots"]
+# Convenience aggregate.
+full = ["grpc", "http", "tcp", "cloud_events", "tls"]
+# End-to-end tests (tests/e2e/*). Gated off by default so plain `cargo test`
+# never needs a live server. Run with `cargo test --features e2e`. Implies all
+# transports + cloud_events so the full suite (gRPC + HTTP + TCP + CE) compiles
+# from a single flag.
+e2e = ["grpc", "http", "tcp", "cloud_events"]
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [dependencies]
-# common
-anyhow = "1.0"
+tokio = { version = "1", features = ["full"] }
+tokio-stream = "0.1"
+tokio-util = "0.7"
+futures = "0.3"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+thiserror = "1"
+bytes = "1"
+rand = "0.8"
+uuid = { version = "1", features = ["v4"] }
+tracing = "0.1"
+http = "1"
+# gRPC transport.
+tonic = { version = "0.12", optional = true }
+prost = { version = "0.13", optional = true }
+prost-types = { version = "0.13", optional = true }
+# CloudEvents interop.
+cloudevents = { package = "cloudevents-sdk", version = "0.7", optional = true }
+# HTTP transport.
+reqwest = { version = "0.12", optional = true, features = ["json"] }
+serde_urlencoded = { version = "0.7", optional = true }
+tower = { version = "0.5", optional = true }
+tower-http = { version = "0.6", optional = true }
+axum = { version = "0.7", optional = true }
 
-#Rust grpc
-tonic = "0.10"
-prost = "0.12"
-prost-types = "0.12"
+[build-dependencies]
+tonic-build = "0.12"
 
-#tokio
-tokio = { version = "1.32.0", features = ["full"] }
+[dev-dependencies]
+tokio = { version = "1", features = ["full", "test-util"] }
+tracing-subscriber = { version = "0.3", features = ["fmt"] }
+# e2e test harness (only referenced under the `e2e` feature):
+reqwest = { version = "0.12", default-features = false, features = ["json"] }
+ctor = "0.2"
 
-#serde
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
+[[example]]
+name = "grpc_producer"
+path = "examples/grpc/producer.rs"
+required-features = ["grpc"]
 
-#log
-tracing = "0.1"
-tracing-subscriber = "0.3"
+[[example]]
+name = "grpc_consumer"
+path = "examples/grpc/consumer.rs"
+required-features = ["grpc"]
 
-#cloudEvents
-cloudevents-sdk = "0.7.0"
+[[example]]
+name = "grpc_producer_cloud_events"
+path = "examples/grpc/producer_cloud_events.rs"
+required-features = ["grpc", "cloud_events"]
 
-# tools crate
-thiserror = "1.0"
-bytes = "1"
-rand = "0.8"
-uuid = { version = "1.4.1", features = ["v4"] }
-local-ip-address = "0.5.6"
-futures = "0.3"
-log = "0.4.20"
-chrono = "0.4"
+[[example]]
+name = "http_producer"
+path = "examples/http/producer.rs"
+required-features = ["http"]
 
-[build-dependencies]
-tonic-build = "0.10"
+[[example]]
+name = "http_consumer_server"
+path = "examples/http/consumer_server.rs"
+required-features = ["http"]
+
+[[example]]
+name = "http_consumer_custom"
+path = "examples/http/consumer_custom.rs"
+required-features = ["http"]
+
+[[example]]
+name = "tcp_producer"
+path = "examples/tcp/producer.rs"
+required-features = ["tcp"]
 
 [[example]]
-name = "producer_example"
-path = "examples/grpc/producer_example.rs"
-required-features = ["grpc", "eventmesh_message","cloud_events"]
+name = "tcp_consumer"
+path = "examples/tcp/consumer.rs"
+required-features = ["tcp"]
 
 [[example]]
-name = "consumer_example"
-path = "examples/grpc/consumer_example.rs"
-required-features = ["grpc", "eventmesh_message"]
+name = "tcp_producer_cloud_events"
+path = "examples/tcp/producer_cloud_events.rs"
+required-features = ["tcp", "cloud_events"]