zhaohai666 opened a new pull request, #1381:
URL: https://github.com/apache/rocketmq-clients/pull/1381
# feat(rust): implement LiteSimpleConsumer
## Summary
This PR ports the **LiteSimpleConsumer** from the reference Java client to
the Rust client
(RocketMQ 5.x gRPC protocol), branch `feat/rust-lite-simple-consumer` based
on `master`
(`fc088d4a`). It lands **1 commit** touching **12 files (+1,283 / −32)**.
The lite consumer model lets one consumer group consume many short-lived
"lite topics" that all map
onto a single parent topic. The consumer binds to the parent topic, then
dynamically subscribes /
unsubscribes individual lite topics through the `SyncLiteSubscription` RPC,
and pulls their messages
explicitly with `receive` + `ack`.
**What is included**
- `src/lite_simple_consumer.rs`: `LiteSimpleConsumer` +
`LiteSimpleConsumerTrait` with
`subscribe_lite`, `subscribe_lite_with_offset`, `unsubscribe_lite`,
`get_lite_topic_set`,
`get_consumer_group`, `get_bind_topic`, `receive(max_message_num,
invisible_duration)`, `ack`,
`change_invisible_duration` and `shutdown`. It reuses the existing
dual-client design of
`LitePushConsumer`: the owning `SimpleConsumer` plus a lightweight cloned
`Client` shared with
`LiteSubscriptionManager`, so both talk over the same telemetry session.
- Telemetry: `NotifyUnsubscribeLiteCommand` and `Settings` are forwarded to
`LiteSubscriptionManager` (the lite push consumer's handler loop was wired
to a channel that never
received anything; the new consumer uses the channel that `Client::start`
actually feeds).
- `ClientType::LiteSimpleConsumer` (=6) plus
`Client::clone_for_lite_simple_consumer()`.
- `util::build_lite_simple_consumer_settings`: simple-consumer settings
reporting
`LITE_SIMPLE_CONSUMER` with the bind topic subscribed as `SUB_ALL`.
- `util::prune_lite_route`: keeps only the first readable master queue of a
route, mirroring
`LiteSimpleConsumerImpl#updateSubscriptionLoadBalancer` in Java (a lite
consumer only needs a
route to *a* broker of the parent topic; the server resolves the lite
queues).
- `SimpleConsumer`: adds `new_with_client`, `start_with_telemetry`,
`check_started`,
`shutdown_ref` and `receive_lite` so the lite consumer can reuse the
simple-consumer plumbing
without duplicating it.
- Two defect fixes on the lite path (see below).
---
## Defect fix 1: missing `lite_topic` in ack / change-invisible-duration
requests
`Client::ack_message` and `Client::change_invisible_duration` always sent
`lite_topic: None`. The
proxy resolves LMQ receipt handles through that field, so acking a lite
message fails with
`50001 INTERNAL_SERVER_ERROR`.
`AckMessageEntry` now exposes `lite_topic()` (default `None`, implemented by
`MessageView` from the
message system properties), the entry construction is factored into
`util::build_ack_message_entry`,
and both requests carry it. Since the fix lives in `Client`, it benefits
every consumer that
receives lite messages, not only the new one.
---
## Defect fix 2: `unsubscribe_lite` always failed with "client is not
started"
`LiteSubscriptionManager::unsubscribe_lite` validated the running state on
the client it holds,
which is the lightweight clone created by `clone_for_lite_consumer()`. That
clone intentionally has
`shutdown_tx = None` (only the owning client may trigger shutdown), so
`check_started` always
reported "client is not started" and every unsubscribe failed — for the push
consumer as well.
The state check is now the responsibility of the public API of the owning
consumer, exactly like
`subscribe_lite` already did. This was found by the real-cluster test, not
by unit tests.
---
## Tests
**Offline tests**
- `cargo test --lib`: 93 passed (4 of them added by this PR): settings carry
`LITE_SIMPLE_CONSUMER` with the bind topic subscribed as `SUB_ALL`; route
pruning keeps the first
readable master queue and errors on an empty/unreadable route; the ack
entry carries `lite_topic`
for lite messages and `None` for regular ones.
- `tests/lite_simple_consumer_test.rs`: 4 passed — consumer group / bind
topic validation,
`get_consumer_group` / `get_bind_topic` / `get_lite_topic_set`, and
`subscribe_lite` / `unsubscribe_lite` before `start` returning
`ClientIsNotRunning`.
**Real-cluster integration tests** — 2/2 passing against a local RocketMQ
5.5.1 deployment
(namesrv + broker with `enableLmq=true` and `enableMultiDispatch=true`,
standalone **proxy in
CLUSTER mode**, parent topic created with `message.type=LITE`, consumer
group pre-created with
`+lite.bind.topic=<parentTopic>`):
- `test_receive_and_ack`: send 5 lite messages →
`subscribe_lite_with_offset` with
`OffsetPolicy::Min` → receive → assert `lite_topic` on every message → ack
all →
`unsubscribe_lite` removes it from the local set → shutdown.
- `test_receive_after_subscribe`: messages produced after the subscription
are delivered and acked.
The suite is gated by `ROCKETMQ_RUST_LITE_ENDPOINTS`; without it the tests
print a skip message and
pass, so CI without a cluster stays green.
**Example verified end to end**
`examples/lite_simple_consumer.rs` was run against the same cluster:
subscribe → receive 3 lite
messages (`lite_topic=Some("example-lite-topic")`) → ack (no errors) →
unsubscribe → clean
shutdown, with the broker logging the consumer as `CONSUME_ACTIVELY
LITE_SELECTIVE`.
---
## Notes
- Like `lite_push_consumer`, the new module is compiled under
`#[cfg(not(test))]`: in test builds
`Client` resolves to `MockClient` (via `mockall_double`), so lite
consumers cannot be constructed
in unit tests. Its assertions therefore live in `tests/`, while pure
helpers (settings, route
pruning, ack entry) are unit tested in `src/util.rs`.
- `cargo fmt --check` and `cargo clippy --all-features -- -D warnings` (the
CI commands) are clean.
---
## Checklist
- [x] Reference implementation (`java/client/.../LiteSimpleConsumerImpl`)
reviewed and mirrored
- [x] `cargo build` / `cargo clippy --all-features -- -D warnings` / `cargo
fmt --check` clean
- [x] Offline tests pass (`cargo test --lib` 93, `lite_simple_consumer_test`
4)
- [x] Real-cluster integration tests pass (2/2)
- [x] Example added, executed end to end against a local cluster, and
documented in `rust/README.md`
- [x] No changes to generated protobuf bindings required
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
