Refactor(rust-sdk): rewrite and align with java sdk#5271
Draft
daflyinbed wants to merge 19 commits into
Draft
Conversation
…d broker response codes - codec: to_event skips response attrs using canonical ProtocolKey constants (statuscode/responsemessage/time) instead of stale spellings that leaked them as cloudevents extensions - codec: get_subject no longer falls back to source, matching Java EventMeshCloudEventUtils.getSubject (avoids default '/' being treated as topic) - consumer: subscribe_webhook records local subscriptions only on broker success - consumer: unsubscribe removes local subscriptions only on broker success
…-compose Add an end-to-end test suite (tests/e2e/) for the Rust SDK that exercises the gRPC producer/consumer against a live EventMesh runtime. The suite is fully self-contained: docker-compose.yml and its docker/conf/ config files move into the crate directory so the tests carry their own infrastructure. - e2e feature gate keeps plain server-free; tests auto-skip when neither Docker nor a server is reachable - runtime.rs manages the container lifecycle from Rust (no shell script): OnceLock-guarded + ctor::dtor teardown, with an EVENTMESH_E2E_EXTERNAL override to reuse an external server - parallel-safe by design: each test uses a unique topic and consumer group (default cargo test parallelism, no --test-threads=1) - harness creates topics via the admin HTTP API and warms subscriptions (standalone requires a subscriber before publishing) - capability-aware: request/reply and unsubscribe-after-publish tolerate standalone broker limitations while asserting fully on RocketMQ - covers publish, batch publish, one-way, stream subscribe+receive, unsubscribe, and request/reply round-trip
…hosts
- add :z to config bind mounts so containers can read them on
SELinux-enforcing hosts (configs are labeled user_home_t, unreadable by
the confined container process)
- shrink broker heap and cap MaxDirectMemorySize: runbroker.sh auto-sizes
the heap from total host RAM (~3.8GB here) with -XX:+AlwaysPreTouch,
which prevents the broker from starting on memory-constrained hosts
(silent exit 253)
- drop broker-logs/broker-store named volumes: the rocketmq image doesn't
ship /home/rocketmq/{logs,store}, so those volumes are created
root-owned and the broker (uid 3000) can't write to them, crashing on
store init
Exercise the SDK against a durable RocketMQ backend (namesrv + broker + eventmesh) instead of the in-memory standalone store. Update the profile selection in the lifecycle harness and the matching doc comments.
The EventMesh admin /topic endpoint parses the POST body with Netty's
HttpPostRequestDecoder (application/x-www-form-urlencoded / multipart),
not JSON. Sending {"name":...} yielded a blank name and the broker threw
"Topic name can not be blank", so the topic was never created. The
consumer then subscribed to a non-existent topic, its rebalance failed
("topic not exist"), and no messages were ever pulled/delivered.
Use reqwest's .form() so the name is sent correctly; this makes topic
creation actually succeed for both the standalone and rocketmq profiles.
The channel was built with .timeout(config.timeout) (default 5s), which is a hard per-RPC cap applied to every call on the channel. That wrongly terminated two cases: - request_reply: the caller may pass a much longer round-trip timeout (e.g. 15s), but the channel killed the RPC at 5s with a transport TimeoutExpired before the reply arrived. - subscribe_stream: the bidirectional subscription stream is long-lived, so a channel-wide request timeout is incorrect by construction. Drop the channel-wide timeout and apply the config default only to the short unary RPCs (publish / batch publish / one-way / webhook subscribe / unsubscribe) via a small `timed` wrapper. request_reply keeps its caller-supplied timeout and subscribe_stream stays unbounded.
…class wins The published apache/eventmesh image ships rocketmq-*.jar in lib/ (system classpath). With them there, parent-first classloading always resolves rocketmq-client's own ConsumeMessageConcurrentlyService over EventMesh's shadow copy that lives in the storage plugin, so the consume listener throws ClassCastException and silently drops every message before it can be delivered over gRPC (apache#5213). The plugin classloader (JarExtensionClassLoader) sorts its URL list alphabetically, so once the rocketmq jars sit next to eventmesh-storage-rocketmq.jar under plugin/storage/rocketmq/, the EventMesh shadow class loads first and consumption works. Override the eventmesh-rocketmq container command to move the jars there before start.sh runs, so the e2e suite works against the unmodified image.
… request Mirror the Java SDK's SubStreamHandler.buildReplyMessage so the consumer's stream reply carries what the broker needs to match it back to the pending request: - build_reply now merges the incoming request's CloudEvent attributes into the reply (reply attrs win, request fills the gaps), propagating correlation99id / reply99to99client that RocketMQ's RequestFutureHolder pairs on. Previously the reply dropped these and the producer always timed out with CODE 10006. - mark_as_reply no longer clears the reply data. EventMesh's ReplyMessageProcessor runs ServiceUtils.validateCloudEventData, which for text content requires non-empty textData; an empty body failed validation so producer.reply() was never invoked. The Java SDK does not strip the reply payload. Also tighten the e2e request_reply skip so only a genuine 'not supported' (standalone) reply is skipped; a request timeout (10006) now surfaces as a real failure instead of being masked. Verified: cargo test --features e2e passes 7/7; server logs show the full round-trip (request -> consumer reply ~17ms -> producer receives reply ~34ms).
The background heartbeat task spawned in GrpcConsumer::new was never cancellable — no Drop, no shutdown signal, JoinHandle discarded — so every dropped consumer leaked a permanently-running task that kept sending heartbeats against the broker. Introduce a CancellationToken shared by the heartbeat loop and the stream receive loop, with three shutdown entry points: - StreamServe::with_graceful_shutdown(signal) — bind an external trigger (Ctrl-C, oneshot) that cancels the shared token - GrpcConsumer::shutdown() — imperative, awaitable heartbeat exit - Drop — unconditional safety net (cancel + abort) subscribe_stream is now synchronous and returns a StreamServe driver (impl IntoFuture). The gRPC stream is opened lazily on the first poll, mirroring axum's Server::serve: a single .await drives subscription and delivery, with optional .with_graceful_shutdown() chained before it. Also: mark_as_reply now forces datacontenttype=application/json to match the Java SubStreamHandler.buildReplyMessage, and the unused SubscriptionReply type is removed.
feat(rust-sdk): http
* feat(rust-sdk): tcp
* fix(rust sdk): resolve TCP hostnames, sync unsubscribe state, harden e2e RR
Address three PR review findings in the TCP transport:
- connection.rs: pass "host:port" to TcpStream::connect instead of
pre-parsing into SocketAddr, so DNS names like "localhost" (the
default server_addr) resolve. Previously the default config and any
hostname-based deployment failed with InvalidArgument.
- consumer.rs: clear the entire local subscriptions map on a successful
unsubscribe. The runtime's UnSubscribeProcessor ignores the request
body and drops all session topics (matching Java's bodyless
MessageUtils.unsubscribe()), so retaining un-passed topics left the
SDK believing they were still subscribed. Added a loopback unit test
verifying subscribe A+B then unsubscribe([A]) yields empty local state.
- tcp_request_reply.rs: only skip the RR assertion for an externally
provided server (Mode::External, the sole standalone-capable path);
panic on Mode::Started where the harness-launched rocketmq broker
must support request/reply. Stops timeouts, codec regressions, bad
ACKs, and connection failures from being silently swallowed.
* fix(rust sdk): preserve request metadata in TCP replies
Merge the inbound request's wire properties into the listener's reply
before serializing RESPONSE_TO_SERVER, mirroring the gRPC consumer's
build_reply merge. A hand-built reply previously dropped the request's
correlation extensions (RocketMQ reply-to / correlation-id), so the
broker could not match the reply and TcpProducer::request_reply timed
out.
* fix(rust sdk): make TCP header seq optional for server-initiated frames
The Java runtime sends SERVER_GOODBYE_REQUEST and REDIRECT_TO_CLIENT with
seq = null (omitted by JsonUtils). A required seq field made serde reject
those valid frames before handle_inbound could send SERVER_GOODBYE_RESPONSE,
so clients dropped the connection with a codec error during graceful
shutdown/rebalance. Change Header.seq to Option<String> and adjust
correlation/ACK helpers accordingly.
* fix(rust sdk): handle REDIRECT_TO_CLIENT and preserve redirect target fields
The REDIRECT_TO_CLIENT feature was wired into the codec dispatch but
broken in two places, leaving TCP consumers unable to follow a rebalance:
1. RedirectInfo only had a defaulted `redirect_to` field, which does not
match the Java wire shape (`RedirectInfo{ip,port}`). Serde silently
discarded the target address when decoding a redirect frame, so even a
handler could not reconnect. Align the struct with
org.apache.eventmesh.common.protocol.tcp.RedirectInfo (ip: String,
port: u16).
2. Command::RedirectToClient fell through to the default branch in
handle_inbound and was silently ignored. The runtime then closes the
session unconditionally after a 30s grace period
(EventMeshTcp2Client.redirectClient2NewEventMesh +
closeSessionIfTimeout), so delivery froze until the forced disconnect.
handle_inbound now returns bool: the new RedirectToClient arm logs the
advertised ip/port and signals the receive loop to stop so the caller
can reconnect immediately, while ServerGoodbyeRequest and the default
arm keep the previous loop semantics.
TCP auto-reconnect: the background I/O task now re-establishes the TCP
connection + HELLO handshake after I/O errors or server-side close, with
exponential backoff (1s–30s) and configurable retry cap (default infinite).
Consumers automatically replay all subscriptions + LISTEN_REQUEST on
reconnect via an internal notification channel. Mirrors the Java SDK's
heartbeat-driven reconnect but with backoff and prompt pending-request
cleanup.
- ReconnectConfig in config/tcp.rs (enabled, max_retries, backoff range)
- connection.rs refactored: establish() + outer reconnect loop +
inner io_loop()
- consumer.rs receive loop: select! on reconnect events, re-subscribe
all topics + re-LISTEN
TCP CloudEvents: native cloudevents::Event interop over TCP, using
protocoltype=cloudevents and application/cloudevents+json body format
(matching the Java runtime's codec path). The producer gains
publish_cloud_event / broadcast_cloud_event / request_reply_cloud_event.
Inbound CloudEvents are transparently converted to EventMeshMessage so
existing listeners handle them without changes.
- message.rs: build_cloud_event_package, parse_cloud_event,
cloud_event_to_message, message_to_cloud_event (behind cloud_events)
- producer.rs: three new inherent methods
- consumer.rs: handle_inbound detects cloudevents protocol and converts
- examples/tcp/producer_cloud_events.rs
- tests/e2e/tcp_cloud_events.rs (publish + broadcast)
75 unit tests pass; clippy -D warnings clean; e2e compiles.
…tatus codes - publish_one_way now calls the dedicated PublisherService/publishOneWay RPC instead of the two-way publish, honoring fire-and-forget semantics - publish_cloud_event and request_reply check is_success() before returning Ok, mirroring publish/publish_batch so ? no longer treats broker-rejected (ACL/validation/send-failure) responses as success
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Welcome to the Apache EventMesh community!!
This is your first PR in our project. We're very excited to have you onboard contributing. Your contributions are greatly appreciated!
Please make sure that the changes are covered by tests.
We will be here shortly.
Let us know if you need any help!
Want to get closer to the community?
| WeChat Assistant | WeChat Public Account | Slack |
|---|---|---|
 |
 |
Join Slack Chat |
Mailing Lists:
| Name | Description | Subscribe | Unsubscribe | Archive |
|---|---|---|---|---|
| Users | User support and questions mailing list | Subscribe | Unsubscribe | Mail Archives |
| Development | Development related discussions | Subscribe | Unsubscribe | Mail Archives |
| Commits | All commits to repositories | Subscribe | Unsubscribe | Mail Archives |
| Issues | Issues or PRs comments and reviews | Subscribe | Unsubscribe | Mail Archives |
Contributor
|
@daflyinbed please create an issue associate with this pr,and fix the pr title, you can accord to other prs. thx |
…ion lifecycle Remove the Subscriber trait and split subscribe/unsubscribe (RPC) from the receive-loop driver (connection lifecycle). Each transport now exposes transport-specific consumer types with a background receive loop spawned at construction time: - GrpcStreamConsumer: eager-open bidi stream + spawned driver + heartbeat. subscribe() sends over the active stream; unsubscribe() is a unary RPC. - GrpcWebhookConsumer: lightweight RPC-only client (no listener, no driver). - TcpConsumer: connect() does TCP+HELLO+LISTEN+spawn driver in one step. subscribe/unsubscribe via conn.io() on the already-open connection. - HttpConsumer: subscribe_webhook/unsubscribe as inherent methods; wait_for_shutdown added for clean exit. Shutdown signal is optional, passed at construction. wait_for_shutdown() blocks until the signal fires or the connection closes (clean exit); drop does best-effort cancel+abort. Deletes: Subscriber trait, StreamServe/ListenServe public types, with_graceful_shutdown on drivers, TCP listen()/add_subscription().
- docker-compose.yml: default both EventMesh services to the locally-built eventmesh:jdk11-test image (IMAGE override still honored) - runtime.rs: open perms on the bind-mounted docker/conf/ dir + files before . Containers run as a different uid (rocketmq = 3000) and on hosts with a restrictive ACL (other::---) the broker couldn't read its own config and crashed on boot with FileNotFoundException - subscribe.rs / tcp_subscribe.rs: the unsubscribe assertions only treated a timeout as success, but the server tears down the stream on unsubscribe so the channel returns Ok(None). Accept Ok(None) too; only a real delivered message (Ok(Some)) is a leak
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This is a rewrite of rust sdk. I wrote old rust sdk it 2 years ago and stuck in setup a environment to debug it.
There is some anti pattern to make rust sdk compitable with java sdk.
Todo: