[WIP] Refactor(rust-sdk): rewrite and align with java sdk#5271
Draft
daflyinbed wants to merge 35 commits into
Draft
[WIP] Refactor(rust-sdk): rewrite and align with java sdk#5271daflyinbed wants to merge 35 commits into
daflyinbed wants to merge 35 commits into
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
daflyinbed
changed the title
daflyinbed
changed the title
…CP CloudEvents The server's CloudEventsProtocolAdaptor.fromCloudEvent uses datacontenttype to resolve the EventFormat serializer via EventFormatProvider.resolveFormat(). Only application/cloudevents+json is registered; other values (application/json, text/plain) return null and trigger Preconditions.checkNotNull NPE on the downlink path, silently dropping messages before they reach consumers. - Set datacontenttype to application/cloudevents+json in e2e tests and example - Document the requirement in TCP module, producer methods, and build_cloud_event_package - Increase e2e recv timeouts: 35s for publish (RocketMQ cold-start rebalance), 25s settle for broadcast (CONSUME_FROM_LAST_OFFSET skips pre-rebalance msgs)
The gRPC stream consumer previously processed messages serially: listener.handle().await blocked the receive loop, causing throughput degredation and shutdown stalls when handlers were slow. Replace the serial dispatch with bounded concurrency: - Semaphore (permits = max_concurrent_handlers, default 64) limits in-flight handler tasks; when exhausted the loop stops pulling from the gRPC stream, engaging flow-control backpressure. - JoinSet manages spawned handler tasks; completed tasks are reaped each loop iteration to prevent unbounded growth. - On shutdown the loop drains all in-flight handlers to completion (mirroring axum's graceful-shutdown behaviour). Drop still aborts. - Reply order is no longer guaranteed to match message-arrival order; each reply is self-correlating via request attributes so protocol correctness is unaffected. Set max_concurrent_handlers=1 to restore strict serial / in-order-reply semantics (matching the Java SDK). Add GrpcClientConfig::max_concurrent_handlers field + builder setter. Add e2e test verifying handler overlap under a slow listener.
…t-thread runtime All e2e tests used #[tokio::test] which defaults to a current-thread (single-worker) runtime. Tonic's bidirectional subscribe_stream RPC requires background connection-driver tasks that cannot progress on a current-thread runtime, causing an indefinite silent hang. Changes: - Switch all e2e tests to #[tokio::test(flavor = "multi_thread")] - Add a 15s timeout to GrpcClient::subscribe_stream that converts the hang into an actionable error message mentioning the runtime cause - Document the multi-threaded runtime requirement on the public API - Gracefully skip publish_one_way when the server returns UNIMPLEMENTED (EventMesh runtime does not implement the publishOneWay gRPC method)
1. gRPC consumer CPU spin: guard join_next() with pending() on empty
JoinSet; cancel shutdown token on stream end; add heartbeat RPC timeout.
2. gRPC webhook unsubscribe: pass the webhook URL to the server instead
of always sending url=None, which left ghost subscriptions.
3. HTTP webhook payload corruption: stop treating content as a serialized
EventMeshMessage when it starts with '{'. The Runtime always puts the
business payload in content; the previous logic silently lost data
when the payload happened to deserialize as EventMeshMessage. Test
updated to use a realistic business JSON payload.
4. TCP consumer ACK: disconnect without ACK on parse failure or when
CloudEvents feature is disabled; let listener panics propagate
instead of swallowing them and ACKing.
5. gRPC CloudEvent conversion: preserve standard CE id, time, dataschema,
and typed extensions (Boolean/Integer) instead of stringifying;
reverse conversion preserves type, binary data, and time.
6. TLS configuration: reject use_tls=true when the tls feature is off
instead of silently producing an unencrypted https URI; load
OS-native trust roots when tls_config is None.
7. MSRV: update declared rust-version from 1.75 to 1.86 (matching the
actual minimum imposed by dependencies).
8. e2e false positive: add EVENTMESH_E2E_STRICT env var. Without it
tests still skip (return early, counted as passed). With it, tests
panic when no runtime is available, so CI goes red instead of green.
…ibe error semantics, wire compat fixes Issue 5 — PublishResponse::is_success() now requires code == Some(0); missing/unparseable codes are treated as failure, not success. Issue 7 — Wire compatibility fixes: - gRPC application/protobuf: decode/encode content as serialized google.protobuf.Any (matches Java's Any.parseFrom/toByteArray). - HTTP extFields: filter reserved keys (topic, content, ttl, bizseqno, uniqueid, producergroup) so runtime post-processing cannot reverse-overwrite typed form fields. - TCP: merge wire headers into props on receive so protocol metadata (datacontenttype, etc.) survives the round-trip. Issue 8 — TCP CloudEvent publish paths (publish/broadcast/request_reply) now validate datacontenttype == application/cloudevents+json before sending; other values cause a server-side NPE and silent message drop. Fix the cloudevents_build_sets_protocol_headers test that used the documented bad value application/json. Issue 9 — TCP unsubscribe now returns Err(Server) on non-zero response codes, consistent with subscribe and publish paths. Local subscription state is only cleared on success.
…panic hang, and surface redirect reason Issue 1 (ghost writes): io()/send() now fail-fast via is_active() check when the connection is down; outbound queue is drained on disconnect so stale packages are never re-sent after reconnect. Orphan RESPONSE_TO_CLIENT frames are silently dropped to avoid producer-only connections accumulating late replies. Issue 2 (slow consumer): inbound channel Full no longer silently drops server pushes. Instead the I/O loop exits with SlowConsumer, triggering reconnect so the server redelivers unacked messages. Issue 3 (driver panic hang): wait_for_shutdown() now races the shutdown token against the driver JoinHandle, returning ShutdownReason::Error on panic instead of hanging forever. Issue 4 (redirect not returned): REDIRECT_TO_CLIENT now extracts the RedirectInfo and surfaces it via wait_for_shutdown()'s new return type ShutdownReason::Redirect(RedirectInfo), so the caller can reconnect to the advertised node.
…rl) subscription key, gRPC error semantics, loadbalance OOM/overflow, webhook extFields retry - apache#6: implement redacting Debug for ClientIdentity, TlsClientIdentity, TlsConfig, TlsConfigBuilder, UserAgent, and all three config builders; password/token/key_pem show as "***", cert PEMs show byte length only - apache#5: remove #[serde(default)] from EventMeshRetObj.ret_code so missing or non-numeric retCode fails deserialization instead of defaulting to 0 - #9a: change subscription map key from topic to (topic, url) in both HTTP and gRPC consumers so same-topic-different-URL subscriptions coexist without overwriting each other - #9b: gRPC subscribe_webhook_rpc / unsubscribe_*_rpc now return Err on non-zero business code instead of Ok(response), matching HTTP and TCP - #10b: WeightRandom no longer expands nodes by weight (OOM risk); adopt Java's totalWeight + linear scan approach, O(1) memory - #10c: WeightRoundRobin counters promoted from i32 to i64 with precomputed total_weight to prevent overflow on large weights - #10e: invalid extFields JSON in webhook push body now returns Err (triggers WebhookReply::retry) instead of being silently swallowed
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: