This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new d7408c74 [rust] Add LiteSimpleConsumer with tests and example (#1381)
d7408c74 is described below
commit d7408c74946aefbb890144f15005613126cc7de7
Author: zhaohai <[email protected]>
AuthorDate: Sat Oct 24 06:05:28 2026 +0800
[rust] Add LiteSimpleConsumer with tests and example (#1381)
* [rust] Add LiteSimpleConsumer with tests and example
Implement the lite-topic variant of SimpleConsumer for the Rust client,
mirroring the Java reference implementation.
Implementation
- LiteSimpleConsumer: a consumer bound to a single parent (bind) topic that
dynamically (un)subscribes lite topics through SyncLiteSubscription and
pulls
messages explicitly with receive()/ack()/change_invisible_duration(). It
wraps a SimpleConsumer plus a lightweight cloned client shared with
LiteSubscriptionManager, and forwards NotifyUnsubscribeLiteCommand /
Settings
telemetry commands to the manager.
- ClientType: add LiteSimpleConsumer (=6) and a
clone_for_lite_simple_consumer()
variant of the lite client clone.
- build_lite_simple_consumer_settings(): simple consumer settings carrying
client type LITE_SIMPLE_CONSUMER and the bind topic with a SUB_ALL filter.
- prune_lite_route(): keep only the first readable master queue of a route,
as
LiteSimpleConsumerImpl#updateSubscriptionLoadBalancer does in Java.
- SimpleConsumer: add new_with_client(), start_with_telemetry(),
check_started(), shutdown_ref() and receive_lite() so the lite consumer
can
reuse the simple consumer plumbing.
Defect fix: missing liteTopic in ack
- AckMessageEntry now exposes lite_topic() and the ack / change invisible
duration requests carry it. Without it the proxy cannot resolve the LMQ
receipt handle and answers 50001 INTERNAL_SERVER_ERROR.
Defect fix: unsubscribeLite always failed
- LiteSubscriptionManager::unsubscribe_lite() validated the running state
on the
lightweight client clone, whose shutdown_tx is intentionally None, so it
always answered "client is not started". The state check now belongs to
the
public API of the owning consumer, like subscribeLite already did.
Tests
- tests/lite_simple_consumer_test.rs: 4 offline tests (builder validation,
accessors, life cycle checks).
- src/util.rs: unit tests for the lite settings, the route pruning and the
liteTopic carried in the ack entry.
- tests/lite_simple_consumer_integration_test.rs: 2 real-cluster tests
guarded
by ROCKETMQ_RUST_LITE_ENDPOINTS. Verified against a local RocketMQ 5.5.1
cluster (broker enableLmq/enableMultiDispatch, proxy in CLUSTER mode,
message.type=LITE parent topic, group with +lite.bind.topic): 2/2 passed.
- Offline suites: cargo test --lib 93 passed, lite_simple_consumer_test 4
passed; cargo fmt --check and cargo clippy --all-features -- -D warnings
clean.
Example
- examples/lite_simple_consumer.rs plus a "Lite Topic" section in README.md.
The example was also run against the local cluster end to end: subscribe,
receive 3 lite messages, ack, unsubscribe, shutdown.
* fix(rust): harden LiteSimpleConsumer lifecycle and settings handling
- Stop the LiteSubscriptionManager periodic-sync scheduler and await its
join handle before closing client sessions. The scheduler retained an
Arc<Client> and could recreate a telemetry session and re-send the
CompleteAdd command after shutdown. A unit test keeps the runtime alive
for more than one sync interval after stopping and asserts no further
RPCs are issued.
- Check the running state before ack() and change_invisible_duration().
Both previously panicked at unwrap() inside get_session_with_endpoints()
when called after shutdown(&mut self); they now return
ErrorKind::ClientIsNotRunning. Covered by integration tests.
- Wait for the server's initial Settings (bounded, 3s) before start()
returns, so an immediate first subscribe() validates against the
server-provided max_lite_topic_size instead of the local default 64.
Falls back to local defaults with a warning on timeout. Covered by an
integration test that subscribes immediately after start.
---
rust/README.md | 54 +++
rust/examples/lite_simple_consumer.rs | 104 +++++
rust/src/client.rs | 60 +--
rust/src/lib.rs | 6 +
rust/src/lite_push_consumer.rs | 7 +
rust/src/lite_simple_consumer.rs | 442 +++++++++++++++++++++
rust/src/lite_subscription_manager.rs | 259 +++++++++++-
rust/src/model/common.rs | 1 +
rust/src/model/message.rs | 12 +
rust/src/simple_consumer.rs | 103 ++++-
rust/src/util.rs | 252 ++++++++++++
.../tests/lite_simple_consumer_integration_test.rs | 289 ++++++++++++++
rust/tests/lite_simple_consumer_test.rs | 109 +++++
13 files changed, 1652 insertions(+), 46 deletions(-)
diff --git a/rust/README.md b/rust/README.md
index dc80d132..0b9999df 100644
--- a/rust/README.md
+++ b/rust/README.md
@@ -38,6 +38,60 @@ cargo run --example producer
cargo run --example simple_consumer
```
+### Lite Topic
+
+Lite topics are sub-topics of a parent topic with reduced metadata and storage
overhead.
+
+Server side preparation:
+
+1. start the broker with `enableLmq=true` and `enableMultiDispatch=true`;
+2. create the parent topic with `message.type=LITE`:
+
+ ```sh
+ sh mqadmin updateTopic -n <namesrv> -c <cluster> -t <parentTopic> -a
+message.type=LITE
+ ```
+
+3. create the consumer group and bind it to the parent topic:
+
+ ```sh
+ sh mqadmin updateSubGroup -n <namesrv> -c <cluster> -g <group> --attributes
"+lite.bind.topic=<parentTopic>"
+ ```
+
+4. run the proxy in CLUSTER mode, the LOCAL mode does not serve lite
subscriptions.
+
+Send a lite message:
+
+```rust
+let message = MessageBuilder::lite_message_builder("parentTopic", body,
"lite-topic-1").build()?;
+let receipt = producer.send(message).await?;
+```
+
+Pull and ack lite messages with `LiteSimpleConsumer`, which is bound to one
parent topic and
+subscribes lite topics dynamically:
+
+```rust
+let mut consumer = LiteSimpleConsumer::new(client_option, option,
"parentTopic".to_string())?;
+consumer.start().await?;
+consumer
+ .subscribe_lite_with_offset("lite-topic-1".to_string(),
OffsetOption::from_policy(OffsetPolicy::Min))
+ .await?;
+
+let messages = consumer.receive(32, Duration::from_secs(15)).await?;
+for message in &messages {
+ println!("{}", message.message_id());
+ consumer.ack(message).await?;
+}
+
+consumer.unsubscribe_lite("lite-topic-1".to_string()).await?;
+consumer.shutdown().await?;
+```
+
+A runnable version lives in `examples/lite_simple_consumer.rs`:
+
+```sh
+cargo run --example lite_simple_consumer
+```
+
[codecov-rust-image]:
https://img.shields.io/codecov/c/gh/apache/rocketmq-clients/master?flag=rust&label=Rust%20Coverage&logo=codecov
[codecov-url]: https://app.codecov.io/gh/apache/rocketmq-clients
[crates-image]: https://img.shields.io/crates/v/rocketmq.svg
diff --git a/rust/examples/lite_simple_consumer.rs
b/rust/examples/lite_simple_consumer.rs
new file mode 100644
index 00000000..c66008e4
--- /dev/null
+++ b/rust/examples/lite_simple_consumer.rs
@@ -0,0 +1,104 @@
+/*
+ * 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.
+ */
+
+//! Example demonstrating how to pull and ack lite messages with a
LiteSimpleConsumer.
+//!
+//! `LiteSimpleConsumer` is bound to one parent topic and dynamically
(un)subscribes the lite
+//! topics of that parent topic. Unlike `LitePushConsumer`, messages are
pulled explicitly with
+//! `receive(...)` and committed with `ack(...)`.
+//!
+//! Server prerequisites:
+//! * broker started with `enableLmq=true` and `enableMultiDispatch=true`
+//! * a parent topic created with `message.type=LITE`
+//! * a consumer group created with `+lite.bind.topic=<parentTopic>`
+//! * the proxy running in CLUSTER mode (LOCAL mode does not serve lite
subscriptions)
+
+use rocketmq::conf::{ClientOption, SimpleConsumerOption};
+use rocketmq::model::offset_option::{OffsetOption, OffsetPolicy};
+use rocketmq::{LiteSimpleConsumer, LiteSimpleConsumerTrait};
+use std::env;
+use std::error::Error;
+use std::time::Duration;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+ tracing_subscriber::fmt::init();
+
+ let access_url =
env::var("ROCKETMQ_ENDPOINTS").unwrap_or("http://localhost:8081".to_string());
+ let bind_topic =
+
env::var("ROCKETMQ_LITE_PARENT_TOPIC").unwrap_or("lite-parent-topic".to_string());
+ let consumer_group =
+
env::var("ROCKETMQ_LITE_GROUP").unwrap_or("rust-lite-unittest-group".to_string());
+ let lite_topic =
env::var("ROCKETMQ_LITE_TOPIC").unwrap_or("lite-topic-1".to_string());
+
+ let mut client_option = ClientOption::default();
+ client_option.set_access_url(&access_url);
+
+ let mut option = SimpleConsumerOption::default();
+ option.set_consumer_group(&consumer_group);
+
+ let mut consumer = LiteSimpleConsumer::new(client_option, option,
bind_topic.clone())?;
+ consumer.start().await?;
+ println!("LiteSimpleConsumer started, bindTopic={}", bind_topic);
+
+ // Subscribe a lite topic, starting from the earliest message.
+ // The call performs a network request plus a quota check, so its result
must be handled.
+ consumer
+ .subscribe_lite_with_offset(
+ lite_topic.clone(),
+ OffsetOption::from_policy(OffsetPolicy::Min),
+ )
+ .await?;
+ println!("subscribed lite topic: {}", lite_topic);
+ println!("lite topic set: {:?}", consumer.get_lite_topic_set());
+
+ // Pull messages for at most 30 seconds.
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+ while tokio::time::Instant::now() < deadline {
+ let messages = consumer.receive(32, Duration::from_secs(15)).await?;
+ if messages.is_empty() {
+ continue;
+ }
+ for message in messages {
+ println!(
+ "received message: id={}, topic={}, lite_topic={:?}, body={}",
+ message.message_id(),
+ message.topic(),
+ message.lite_topic(),
+ String::from_utf8_lossy(message.body())
+ );
+ // Lite messages must be acked with their lite topic, which the
client does
+ // transparently from `MessageView::lite_topic()`.
+ if let Err(e) = consumer.ack(&message).await {
+ eprintln!("ack message {} failed: {:?}", message.message_id(),
e);
+ // Fall back to extending the invisible duration so that the
message can be
+ // consumed again later.
+ consumer
+ .change_invisible_duration(&message,
Duration::from_secs(10))
+ .await?;
+ }
+ }
+ }
+
+ // Unsubscribe when the lite topic is no longer needed, this releases the
quota.
+ consumer.unsubscribe_lite(lite_topic.clone()).await?;
+ println!("unsubscribed lite topic: {}", lite_topic);
+
+ consumer.shutdown().await?;
+ println!("LiteSimpleConsumer shutdown");
+ Ok(())
+}
diff --git a/rust/src/client.rs b/rust/src/client.rs
index 93820223..e0e5906b 100644
--- a/rust/src/client.rs
+++ b/rust/src/client.rs
@@ -45,7 +45,10 @@ use crate::pb::{
#[double]
use crate::session::Session;
use crate::session::{RPCClient, SharedLongPollingTimeout};
-use crate::util::{handle_receive_message_status, handle_response_status,
select_message_queue};
+use crate::util::{
+ build_ack_message_entry, handle_receive_message_status,
handle_response_status,
+ select_message_queue,
+};
#[derive(Debug)]
pub(crate) struct Client {
@@ -192,32 +195,44 @@ impl Client {
/// // Both clients share the same SessionManager and telemetry session
/// ```
pub(crate) fn clone_for_lite_consumer(&self) -> Self {
- // Update client type to LitePushConsumer
+ // FIFO is enabled for LitePushConsumer to keep the message order of a
lite topic.
+ self.clone_for_lite(
+ ClientType::LitePushConsumer,
+ pb::ClientType::LitePushConsumer,
+ true,
+ )
+ }
+
+ /// Same as [`Client::clone_for_lite_consumer`] but for
`LiteSimpleConsumer`.
+ pub(crate) fn clone_for_lite_simple_consumer(&self) -> Self {
+ self.clone_for_lite(
+ ClientType::LiteSimpleConsumer,
+ pb::ClientType::LiteSimpleConsumer,
+ false,
+ )
+ }
+
+ fn clone_for_lite(
+ &self,
+ client_type: ClientType,
+ pb_client_type: pb::ClientType,
+ fifo: bool,
+ ) -> Self {
let mut new_option = self.option.clone();
- new_option.client_type = ClientType::LitePushConsumer;
+ new_option.client_type = client_type;
- // Create new settings with LitePushConsumer type and FIFO enabled
let mut new_settings = self.settings.clone();
if let Some(Command::Settings(ref mut settings)) = &mut
new_settings.command {
- // Set client type to LitePushConsumer
- settings.client_type = Some(pb::ClientType::LitePushConsumer as
i32);
-
- // Ensure FIFO is enabled for LitePushConsumer
- // This is critical for maintaining message order in lite mode
+ settings.client_type = Some(pb_client_type as i32);
if let Some(pb::settings::PubSub::Subscription(ref mut sub)) =
settings.pub_sub {
- sub.fifo = Some(true);
+ sub.fifo = Some(fifo);
}
}
- // Share the same SessionManager via Arc::clone
- // This is the KEY feature that enables true Lite mode:
- // - Both clients use the same telemetry session
- // - No duplicate connections to the server
- // - Shared state management
+ // Share the same SessionManager via Arc::clone so that both clients
reuse the same
+ // telemetry session.
let session_manager = Arc::clone(&self.session_manager);
- // Create the lightweight clone
- // Note: shutdown_tx is set to None to avoid duplicate shutdown signals
Self {
option: new_option,
session_manager,
@@ -760,11 +775,7 @@ impl Client {
.ack_message_inner(
rpc_client,
ack_entry.topic(),
- vec![pb::AckMessageEntry {
- message_id: ack_entry.message_id(),
- receipt_handle: ack_entry.receipt_handle(),
- lite_topic: None,
- }],
+ vec![build_ack_message_entry(ack_entry)],
)
.await?;
Ok(result[0].clone())
@@ -807,6 +818,7 @@ impl Client {
ack_entry.receipt_handle(),
invisible_duration,
ack_entry.message_id(),
+ ack_entry.lite_topic().map(|topic| topic.to_string()),
)
.await?;
Ok(result)
@@ -819,6 +831,7 @@ impl Client {
receipt_handle: String,
invisible_duration: Duration,
message_id: String,
+ lite_topic: Option<String>,
) -> Result<String, ClientError> {
let request = ChangeInvisibleDurationRequest {
group: Some(Resource {
@@ -832,7 +845,7 @@ impl Client {
receipt_handle,
invisible_duration: Some(invisible_duration),
message_id,
- lite_topic: None,
+ lite_topic,
suspend: None,
};
let response = rpc_client.change_invisible_duration(request).await?;
@@ -1740,6 +1753,7 @@ pub(crate) mod tests {
"receipt_handle".to_string(),
prost_types::Duration::default(),
"message_id".to_string(),
+ None,
)
.await;
assert!(change_invisible_duration_result.is_ok());
diff --git a/rust/src/lib.rs b/rust/src/lib.rs
index b73ba6ce..8186fd27 100644
--- a/rust/src/lib.rs
+++ b/rust/src/lib.rs
@@ -121,6 +121,8 @@
// Export structs that are part of crate API.
#[cfg(not(test))]
pub use lite_push_consumer::{LitePushConsumer, LitePushConsumerTrait};
+#[cfg(not(test))]
+pub use lite_simple_consumer::{LiteSimpleConsumer, LiteSimpleConsumerTrait};
pub use model::common::ConsumeResult;
pub use model::transaction::Transaction;
pub use producer::Producer;
@@ -149,6 +151,10 @@ mod util;
#[cfg(not(test))]
mod lite_push_consumer;
#[cfg(not(test))]
+mod lite_simple_consumer;
+// Not gated behind `cfg(not(test))`: the manager carries its own unit tests.
Its `Client` is
+// doubled in test builds (`#[double]`), and the two consumers above — the
only callers of
+// `LiteSubscriptionManager::new` — are excluded from test builds, so the
types still line up.
mod lite_subscription_manager;
mod producer;
mod push_consumer;
diff --git a/rust/src/lite_push_consumer.rs b/rust/src/lite_push_consumer.rs
index b782309f..f8df28a6 100644
--- a/rust/src/lite_push_consumer.rs
+++ b/rust/src/lite_push_consumer.rs
@@ -413,6 +413,13 @@ impl LitePushConsumerTrait for LitePushConsumer {
async fn shutdown(&mut self) -> Result<(), ClientError> {
info!("Shutting down LitePushConsumer...");
+ // Stop the periodic-sync scheduler first. It retains an `Arc<Client>`
and the
+ // subscription set; if it kept running, its next tick would call
+ // `Client::get_session_for_lite_consumer()` — which deliberately
bypasses the
+ // running-state check — and could re-create a session after
+ // `SessionManager::shutdown()` cleared the session map, then send
`CompleteAdd` again.
+ self.lite_subscription_manager.stop_scheduler().await;
+
if let Some(token) = self.shutdown_token.take() {
token.cancel();
}
diff --git a/rust/src/lite_simple_consumer.rs b/rust/src/lite_simple_consumer.rs
new file mode 100644
index 00000000..65f26f99
--- /dev/null
+++ b/rust/src/lite_simple_consumer.rs
@@ -0,0 +1,442 @@
+/*
+ * 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.
+ */
+
+//! LiteSimpleConsumer - the lite topic variant of `SimpleConsumer`.
+//!
+//! Reference Java: `LiteSimpleConsumerImpl extends SimpleConsumerImpl`
+//!
+//! A `LiteSimpleConsumer` is bound to one parent topic (the *bind topic*) and
dynamically
+//! (un)subscribes lite topics of that parent topic through the
`SyncLiteSubscription` RPC.
+//! Messages are pulled explicitly with `receive()` and committed with
`ack()`, exactly like a
+//! regular `SimpleConsumer`.
+//!
+//! Key design points (aligned with Java):
+//! 1. The consumer is a thin wrapper around `SimpleConsumer` (which owns the
`Client`), plus a
+//! lightweight cloned client used by `LiteSubscriptionManager` for the
subscription RPCs.
+//! 2. Telemetry commands related to lite subscriptions
(`NotifyUnsubscribeLiteCommand`,
+//! `Settings`) are forwarded to `LiteSubscriptionManager`.
+//! 3. `ack()` / `change_invisible_duration()` carry the lite topic of the
message, otherwise the
+//! proxy cannot resolve the LMQ receipt handle and answers
`INTERNAL_SERVER_ERROR`.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use mockall_double::double;
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+use tracing::{debug, error, info, warn};
+
+#[double]
+use crate::client::Client;
+use crate::conf::{ClientOption, SimpleConsumerOption};
+use crate::error::{ClientError, ErrorKind};
+use crate::lite_subscription_manager::LiteSubscriptionManager;
+use crate::model::common::{ClientType, FilterExpression};
+use crate::model::message::MessageView;
+use crate::model::offset_option::OffsetOption;
+use crate::pb;
+use crate::simple_consumer::SimpleConsumer;
+use crate::util::build_lite_simple_consumer_settings;
+
+const OPERATION_NEW_LITE_SIMPLE_CONSUMER: &str = "lite_simple_consumer.new";
+const OPERATION_LITE_SIMPLE_CONSUMER_RECEIVE: &str =
"lite_simple_consumer.receive";
+const OPERATION_LITE_SIMPLE_CONSUMER_ACK: &str = "lite_simple_consumer.ack";
+const OPERATION_LITE_SIMPLE_CONSUMER_CHANGE_INVISIBLE_DURATION: &str =
+ "lite_simple_consumer.change_invisible_duration";
+const OPERATION_SUBSCRIBE_LITE: &str = "lite_simple_consumer.subscribe_lite";
+const OPERATION_UNSUBSCRIBE_LITE: &str =
"lite_simple_consumer.unsubscribe_lite";
+
+/// Upper bound for waiting on the server's initial `Settings` during startup.
When it elapses,
+/// startup proceeds with the built-in defaults (a warning is logged).
+const INITIAL_SETTINGS_WAIT_TIMEOUT: Duration = Duration::from_secs(3);
+
+/// LiteSimpleConsumer trait defining the interface for lite simple consumers
+#[async_trait]
+pub trait LiteSimpleConsumerTrait {
+ /// Subscribe to a lite topic
+ async fn subscribe_lite(&self, lite_topic: String) -> Result<(),
ClientError>;
+
+ /// Subscribe to a lite topic with offset option
+ async fn subscribe_lite_with_offset(
+ &self,
+ lite_topic: String,
+ offset_option: OffsetOption,
+ ) -> Result<(), ClientError>;
+
+ /// Unsubscribe from a lite topic
+ async fn unsubscribe_lite(&self, lite_topic: String) -> Result<(),
ClientError>;
+
+ /// Get the set of subscribed lite topics
+ fn get_lite_topic_set(&self) -> HashSet<String>;
+
+ /// Get the consumer group name
+ fn get_consumer_group(&self) -> String;
+
+ /// Get the parent (bind) topic name
+ fn get_bind_topic(&self) -> String;
+
+ /// Shutdown the consumer
+ async fn shutdown(&mut self) -> Result<(), ClientError>;
+}
+
+/// LiteSimpleConsumer implementation.
+///
+/// Reference Java: `LiteSimpleConsumerImpl extends SimpleConsumerImpl`
+pub struct LiteSimpleConsumer {
+ inner: SimpleConsumer,
+ lite_client: Arc<Client>,
+ lite_subscription_manager: Arc<LiteSubscriptionManager>,
+ bind_topic: String,
+ shutdown_token: Option<CancellationToken>,
+ task_tracker: Option<TaskTracker>,
+}
+
+impl LiteSimpleConsumer {
+ /// Create a new LiteSimpleConsumer
+ pub fn new(
+ client_option: ClientOption,
+ option: SimpleConsumerOption,
+ bind_topic: String,
+ ) -> Result<Self, ClientError> {
+ if option.consumer_group().is_empty() {
+ return Err(ClientError::new(
+ ErrorKind::Config,
+ "consumer group is required.",
+ OPERATION_NEW_LITE_SIMPLE_CONSUMER,
+ ));
+ }
+
+ if bind_topic.is_empty() {
+ return Err(ClientError::new(
+ ErrorKind::Config,
+ "bind topic is required.",
+ OPERATION_NEW_LITE_SIMPLE_CONSUMER,
+ ));
+ }
+
+ // Reference Java: LiteSimpleConsumerBuilderImpl sets {bindTopic:
SUB_ALL} as
+ // subscriptionExpressions and prefetches the route of the bind topic.
+ let mut option = option;
+ option.set_topics(vec![bind_topic.clone()]);
+
+ let settings = build_lite_simple_consumer_settings(&option,
&bind_topic);
+ let namespace = option.namespace().to_string();
+ let consumer_group = option.consumer_group().to_string();
+
+ let client_option = ClientOption {
+ client_type: ClientType::LiteSimpleConsumer,
+ group: Some(consumer_group.clone()),
+ namespace: namespace.clone(),
+ ..client_option
+ };
+ let client = Client::new(client_option, settings)?;
+
+ // Reference Java: new LiteSubscriptionManager(this, new
Resource(bindTopic), groupResource)
+ let lite_client = Arc::new(client.clone_for_lite_simple_consumer());
+ let lite_subscription_manager = Arc::new(LiteSubscriptionManager::new(
+ Arc::clone(&lite_client),
+ bind_topic.clone(),
+ namespace,
+ consumer_group,
+ ));
+
+ let inner = SimpleConsumer::new_with_client(client, option)?;
+
+ Ok(Self {
+ inner,
+ lite_client,
+ lite_subscription_manager,
+ bind_topic,
+ shutdown_token: None,
+ task_tracker: None,
+ })
+ }
+
+ /// Start the LiteSimpleConsumer
+ ///
+ /// Reference Java: `LiteSimpleConsumerImpl#startUp()`
+ pub async fn start(&mut self) -> Result<(), ClientError> {
+ let client_id = self.lite_client.client_id().to_string();
+ info!(
+ "Begin to start the LiteSimpleConsumer, bindTopic={}, clientId={}",
+ self.bind_topic, client_id
+ );
+
+ // Step 1: pre-fetch the route of the bind topic.
+ match self.lite_client.topic_route(&self.bind_topic, false).await {
+ Ok(route) => {
+ info!(
+ "Pre-fetched route for bindTopic={}, message_queues={}",
+ self.bind_topic,
+ route.queue.len()
+ );
+ }
+ Err(e) => {
+ warn!(
+ "Failed to pre-fetch route for bindTopic={}, error={:?}.
Will retry later.",
+ self.bind_topic, e
+ );
+ }
+ }
+
+ // Step 2: start the underlying simple consumer (reference Java:
super.startUp())
+ let (telemetry_command_tx, mut telemetry_command_rx) =
mpsc::channel(16);
+ if let Err(e) =
self.inner.start_with_telemetry(telemetry_command_tx).await {
+ error!(
+ "Failed to start LiteSimpleConsumer inner, bindTopic={},
clientId={}, error={:?}",
+ self.bind_topic, client_id, e
+ );
+ if let Err(shutdown_err) = self.shutdown().await {
+ error!(
+ "Failed to shutdown after start failure, clientId={},
error={:?}",
+ client_id, shutdown_err
+ );
+ }
+ return Err(ClientError::new(
+ ErrorKind::ClientInternal,
+ &format!("startUp err={:?}", e),
+ "lite_simple_consumer.start",
+ ));
+ }
+
+ // Step 3: start the lite subscription manager (reference Java:
liteSubscriptionManager.startUp())
+ if let Err(e) = self.lite_subscription_manager.start().await {
+ error!(
+ "Failed to start LiteSubscriptionManager, bindTopic={},
clientId={}, error={:?}",
+ self.bind_topic, client_id, e
+ );
+ if let Err(shutdown_err) = self.shutdown().await {
+ error!(
+ "Failed to shutdown after start failure, clientId={},
error={:?}",
+ client_id, shutdown_err
+ );
+ }
+ return Err(ClientError::new(
+ ErrorKind::ClientInternal,
+ &format!("startUp err={:?}", e),
+ "lite_simple_consumer.start",
+ ));
+ }
+
+ // Step 4: handle lite specific telemetry commands.
+ // Reference Java: onSettingsCommand() ->
liteSubscriptionManager.sync(settings)
+ // onNotifyUnsubscribeLiteCommand() ->
liteSubscriptionManager.onNotifyUnsubscribeLiteCommand()
+ let shutdown_token = CancellationToken::new();
+ self.shutdown_token = Some(shutdown_token.clone());
+ let task_tracker = TaskTracker::new();
+ self.task_tracker = Some(task_tracker.clone());
+
+ let manager = Arc::clone(&self.lite_subscription_manager);
+ let client_id_clone = client_id.clone();
+ task_tracker.spawn(async move {
+ loop {
+ tokio::select! {
+ command = telemetry_command_rx.recv() => {
+ match command {
+
Some(pb::telemetry_command::Command::NotifyUnsubscribeLiteCommand(ref cmd)) => {
+ info!(
+ "Received unsubscribe notification for
lite topic: {}, clientId={}",
+ cmd.lite_topic, client_id_clone
+ );
+
manager.on_notify_unsubscribe_lite_command(cmd.lite_topic.clone());
+ }
+ Some(pb::telemetry_command::Command::Settings(ref
settings_cmd)) => {
+ debug!(
+ "Received settings update from server,
clientId={}",
+ client_id_clone
+ );
+ manager.sync_settings(settings_cmd);
+ }
+ other => {
+ debug!(
+ "Command {:?} cannot be handled in lite
simple consumer, clientId={}",
+ other.is_some(),
+ client_id_clone
+ );
+ }
+ }
+ }
+ _ = shutdown_token.cancelled() => {
+ break;
+ }
+ }
+ }
+ info!(
+ "LiteSimpleConsumer telemetry handler stopped, clientId={}",
+ client_id_clone
+ );
+ });
+
+ info!(
+ "The LiteSimpleConsumer starts successfully, bindTopic={},
consumerGroup={}, clientId={}",
+ self.bind_topic,
+ self.get_consumer_group(),
+ client_id
+ );
+
+ // Step 5: wait (bounded) until the telemetry handler applied the
server's initial
+ // `Settings`, so that a `subscribe_lite` right after `start()`
validates the lite topic
+ // against the server-provided `max_lite_topic_size` instead of the
built-in default.
+ // On a current-thread runtime the handler spawned above need not have
run yet.
+ if !self
+ .lite_subscription_manager
+ .wait_for_initial_settings(INITIAL_SETTINGS_WAIT_TIMEOUT)
+ .await
+ {
+ warn!(
+ "Timed out waiting for the initial Settings from server,
bindTopic={}, \
+ clientId={}; proceeding with the local defaults,
maxLiteTopicSize=64",
+ self.bind_topic, client_id
+ );
+ }
+
+ Ok(())
+ }
+
+ /// Receive messages of the subscribed lite topics
+ ///
+ /// # Arguments
+ ///
+ /// * `max_message_num` - max message num of server returned
+ /// * `invisible_duration` - set the invisible duration of messages
returned by the server
+ pub async fn receive(
+ &self,
+ max_message_num: i32,
+ invisible_duration: Duration,
+ ) -> Result<Vec<MessageView>, ClientError> {
+ self.inner
+ .check_started(OPERATION_LITE_SIMPLE_CONSUMER_RECEIVE)?;
+ self.inner
+ .receive_lite(
+ &self.bind_topic,
+ &FilterExpression::sub_all(),
+ max_message_num,
+ invisible_duration,
+ )
+ .await
+ }
+
+ /// Ack the specified message
+ ///
+ /// It is important to acknowledge every consumed message, otherwise, they
will be received
+ /// again after the invisible duration.
+ pub async fn ack(&self, message: &MessageView) -> Result<(), ClientError> {
+ self.inner
+ .check_started(OPERATION_LITE_SIMPLE_CONSUMER_ACK)?;
+ self.inner.ack(message).await
+ }
+
+ /// Change the invisible duration of the specified message
+ pub async fn change_invisible_duration(
+ &self,
+ message: &MessageView,
+ invisible_duration: Duration,
+ ) -> Result<String, ClientError> {
+ self.inner
+
.check_started(OPERATION_LITE_SIMPLE_CONSUMER_CHANGE_INVISIBLE_DURATION)?;
+ self.inner
+ .change_invisible_duration(message, invisible_duration)
+ .await
+ }
+}
+
+#[async_trait]
+impl LiteSimpleConsumerTrait for LiteSimpleConsumer {
+ /// Subscribe to a lite topic
+ ///
+ /// Reference Java: `LiteSimpleConsumerImpl#subscribeLite(String)`
+ async fn subscribe_lite(&self, lite_topic: String) -> Result<(),
ClientError> {
+ self.inner.check_started(OPERATION_SUBSCRIBE_LITE)?;
+ self.lite_subscription_manager
+ .subscribe_lite(lite_topic, None)
+ .await
+ }
+
+ /// Subscribe to a lite topic with offset option
+ ///
+ /// Reference Java: `LiteSimpleConsumerImpl#subscribeLite(String,
OffsetOption)`
+ async fn subscribe_lite_with_offset(
+ &self,
+ lite_topic: String,
+ offset_option: OffsetOption,
+ ) -> Result<(), ClientError> {
+ self.inner.check_started(OPERATION_SUBSCRIBE_LITE)?;
+ self.lite_subscription_manager
+ .subscribe_lite(lite_topic, Some(offset_option))
+ .await
+ }
+
+ /// Unsubscribe from a lite topic
+ ///
+ /// Reference Java: `LiteSimpleConsumerImpl#unsubscribeLite(String)`
+ async fn unsubscribe_lite(&self, lite_topic: String) -> Result<(),
ClientError> {
+ self.inner.check_started(OPERATION_UNSUBSCRIBE_LITE)?;
+ self.lite_subscription_manager
+ .unsubscribe_lite(lite_topic)
+ .await
+ }
+
+ /// Get the set of subscribed lite topics
+ ///
+ /// Reference Java: `LiteSimpleConsumerImpl#getLiteTopicSet()`
+ fn get_lite_topic_set(&self) -> HashSet<String> {
+ self.lite_subscription_manager.get_lite_topic_set()
+ }
+
+ /// Get the consumer group name
+ ///
+ /// Reference Java: `LiteSimpleConsumerImpl#getConsumerGroup()`
+ fn get_consumer_group(&self) -> String {
+ self.lite_subscription_manager
+ .get_consumer_group_name()
+ .to_string()
+ }
+
+ /// Get the parent (bind) topic name
+ fn get_bind_topic(&self) -> String {
+ self.bind_topic.clone()
+ }
+
+ /// Shutdown the consumer
+ async fn shutdown(&mut self) -> Result<(), ClientError> {
+ info!("Shutting down LiteSimpleConsumer...");
+
+ // Stop the periodic-sync scheduler first. It retains an `Arc<Client>`
and the
+ // subscription set; if it kept running, its next tick would call
+ // `Client::get_session_for_lite_consumer()` — which deliberately
bypasses the
+ // running-state check — and could re-create a session after
+ // `SessionManager::shutdown()` cleared the session map, then send
`CompleteAdd` again.
+ self.lite_subscription_manager.stop_scheduler().await;
+
+ if let Some(token) = self.shutdown_token.take() {
+ token.cancel();
+ }
+
+ if let Some(tracker) = self.task_tracker.take() {
+ tracker.close();
+ tracker.wait().await;
+ }
+
+ self.inner.shutdown_ref().await?;
+
+ info!("LiteSimpleConsumer shutdown successfully");
+ Ok(())
+ }
+}
diff --git a/rust/src/lite_subscription_manager.rs
b/rust/src/lite_subscription_manager.rs
index b3e9c9eb..b6840f4b 100644
--- a/rust/src/lite_subscription_manager.rs
+++ b/rust/src/lite_subscription_manager.rs
@@ -21,9 +21,14 @@ use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
+use mockall_double::double;
use parking_lot::Mutex;
-use tracing::{error, info};
+use tokio::sync::watch;
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info, warn};
+#[double]
use crate::client::Client;
use crate::error::{ClientError, ErrorKind};
use crate::model::offset_option::OffsetOption;
@@ -34,6 +39,9 @@ use crate::util::handle_response_status;
const OPERATION_SYNC_LITE_SUBSCRIPTION: &str = "lite_subscription.sync";
+/// Default interval between two periodic `SyncLiteSubscription` rounds.
+const DEFAULT_PERIODIC_SYNC_INTERVAL: Duration = Duration::from_secs(30);
+
/// Manages lite topic subscriptions for LitePushConsumer
pub struct LiteSubscriptionManager {
client: Arc<Client>,
@@ -42,6 +50,14 @@ pub struct LiteSubscriptionManager {
lite_topic_set: Arc<Mutex<HashSet<String>>>,
lite_subscription_quota: Arc<Mutex<i32>>,
max_lite_topic_size: Arc<Mutex<i32>>,
+ /// Interval between two periodic sync rounds. Defaults to 30 seconds;
shortened by tests.
+ periodic_sync_interval: Duration,
+ /// Cancellation token of the periodic-sync scheduler.
+ scheduler_token: Mutex<Option<CancellationToken>>,
+ /// Join handle of the periodic-sync scheduler, so `stop_scheduler` can
await its exit.
+ scheduler_handle: Mutex<Option<JoinHandle<()>>>,
+ /// Flipped to `true` once the telemetry handler applied the server's
`Settings`.
+ settings_ready_tx: Arc<watch::Sender<bool>>,
}
impl LiteSubscriptionManager {
@@ -52,6 +68,7 @@ impl LiteSubscriptionManager {
namespace: String,
consumer_group: String,
) -> Self {
+ let (settings_ready_tx, _) = watch::channel(false);
Self {
client,
bind_topic: Resource {
@@ -65,6 +82,10 @@ impl LiteSubscriptionManager {
lite_topic_set: Arc::new(Mutex::new(HashSet::new())),
lite_subscription_quota: Arc::new(Mutex::new(0)),
max_lite_topic_size: Arc::new(Mutex::new(64)), // default value
+ periodic_sync_interval: DEFAULT_PERIODIC_SYNC_INTERVAL,
+ scheduler_token: Mutex::new(None),
+ scheduler_handle: Mutex::new(None),
+ settings_ready_tx: Arc::new(settings_ready_tx),
}
}
@@ -72,21 +93,57 @@ impl LiteSubscriptionManager {
pub async fn start(&self) -> Result<(), ClientError> {
self.sync_all_lite_subscription().await?;
- // Schedule periodic sync every 30 seconds
+ // Replace any scheduler left over from a previous start.
+ self.stop_scheduler().await;
+
+ // Schedule periodic sync every `periodic_sync_interval`. The task
holds an `Arc<Self>`,
+ // hence it must be cancellable: otherwise it would keep the client
alive and could
+ // re-create sessions after `Client::shutdown` cleared the session map.
+ let token = CancellationToken::new();
+ let task_token = token.clone();
let manager = self.clone_for_scheduler();
- tokio::spawn(async move {
- let mut interval = tokio::time::interval(Duration::from_secs(30));
+ let interval = self.periodic_sync_interval;
+ let handle = tokio::spawn(async move {
+ let mut interval = tokio::time::interval(interval);
loop {
- interval.tick().await;
- if let Err(e) = manager.sync_all_lite_subscription().await {
- error!("Schedule syncAllLiteSubscription error: {:?}", e);
+ tokio::select! {
+ _ = interval.tick() => {
+ if let Err(e) =
manager.sync_all_lite_subscription().await {
+ error!("Schedule syncAllLiteSubscription error:
{:?}", e);
+ }
+ }
+ _ = task_token.cancelled() => {
+ break;
+ }
}
}
+ info!("LiteSubscriptionManager periodic sync scheduler stopped");
});
+ *self.scheduler_token.lock() = Some(token);
+ *self.scheduler_handle.lock() = Some(handle);
+
Ok(())
}
+ /// Stop the periodic-sync scheduler, waiting until it exited.
+ ///
+ /// Must be called before closing the client sessions: the scheduler
retains an `Arc<Client>`
+ /// and the subscription set, and its next tick would call
+ /// `Client::get_session_for_lite_consumer()` — which bypasses the
running-state check and
+ /// could re-create a session and send another `CompleteAdd` after
shutdown.
+ pub async fn stop_scheduler(&self) {
+ if let Some(token) = self.scheduler_token.lock().take() {
+ token.cancel();
+ }
+ let handle = self.scheduler_handle.lock().take();
+ if let Some(handle) = handle {
+ if let Err(e) = handle.await {
+ warn!("Failed to join the periodic-sync scheduler: {:?}", e);
+ }
+ }
+ }
+
/// Clone necessary fields for scheduler task
fn clone_for_scheduler(&self) -> Arc<Self> {
Arc::new(Self {
@@ -96,6 +153,10 @@ impl LiteSubscriptionManager {
lite_topic_set: Arc::clone(&self.lite_topic_set),
lite_subscription_quota: Arc::clone(&self.lite_subscription_quota),
max_lite_topic_size: Arc::clone(&self.max_lite_topic_size),
+ periodic_sync_interval: self.periodic_sync_interval,
+ scheduler_token: Mutex::new(None),
+ scheduler_handle: Mutex::new(None),
+ settings_ready_tx: Arc::clone(&self.settings_ready_tx),
})
}
@@ -116,6 +177,10 @@ impl LiteSubscriptionManager {
/// Sync settings from server (reference Java: check hasSubscription first)
pub fn sync_settings(&self, settings: &pb::Settings) {
+ // The initial `Settings` command has arrived: whoever waits for it
may proceed.
+ // `send_replace` (not `send`) so the value is stored even when nobody
is waiting.
+ self.settings_ready_tx.send_replace(true);
+
// Check if settings has subscription (similar to Java's
settings.hasSubscription())
let has_subscription = matches!(
&settings.pub_sub,
@@ -139,6 +204,17 @@ impl LiteSubscriptionManager {
}
}
+ /// Wait until the telemetry handler applied the server's initial
`Settings`.
+ ///
+ /// Returns `true` when the initial settings have been applied, `false`
when `timeout`
+ /// elapsed first (e.g. the server did not push its settings in time).
Callers should log a
+ /// warning in the latter case and continue with the local defaults.
+ pub async fn wait_for_initial_settings(&self, timeout: Duration) -> bool {
+ let mut rx = self.settings_ready_tx.subscribe();
+ let ready = async { rx.wait_for(|ready| *ready).await.is_ok() };
+ tokio::time::timeout(timeout, ready).await.is_ok()
+ }
+
/// Subscribe to a lite topic (reference Java: checkRunning first)
pub async fn subscribe_lite(
&self,
@@ -188,9 +264,10 @@ impl LiteSubscriptionManager {
/// Unsubscribe from a lite topic (reference Java: checkRunning first)
pub async fn unsubscribe_lite(&self, lite_topic: String) -> Result<(),
ClientError> {
- // Check if client is running (reference Java:
consumerImpl.checkRunning() line 119)
- self.client
- .check_started(OPERATION_SYNC_LITE_SUBSCRIPTION)?;
+ // The running state is checked by the public API (LitePushConsumer /
LiteSimpleConsumer) on
+ // the owning client: the client held here is a lightweight clone
whose shutdown_tx is
+ // intentionally `None`, so `check_started` would always report "not
started".
+ // See `Client::clone_for_lite_consumer`.
// Check if subscribed
if !self.lite_topic_set.lock().contains(&lite_topic) {
@@ -313,7 +390,7 @@ impl LiteSubscriptionManager {
}
/// Check if adding delta subscriptions would exceed quota
- fn check_lite_subscription_quota(&self, delta: i32) -> Result<(),
ClientError> {
+ pub(crate) fn check_lite_subscription_quota(&self, delta: i32) ->
Result<(), ClientError> {
let current_size = self.lite_topic_set.lock().len() as i32;
let quota = *self.lite_subscription_quota.lock();
@@ -345,16 +422,25 @@ impl LiteSubscriptionManager {
#[cfg(test)]
mod tests {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
use super::*;
+ /// In test builds `Client` is doubled (`mockall_double`), so a fresh
`MockClient` is used
+ /// instead of the real one. Callers must set up the expectations they
need.
fn create_test_client() -> Arc<Client> {
- // Create a minimal client for testing
- let option = crate::conf::ClientOption::default();
- let settings = pb::TelemetryCommand {
- command: None,
- status: None,
- };
- Arc::new(Client::new(option, settings).unwrap())
+ Arc::new(Client::default())
+ }
+
+ fn test_settings(max_lite_topic_size: i32) -> pb::Settings {
+ pb::Settings {
+ pub_sub: Some(pb::settings::PubSub::Subscription(pb::Subscription {
+ lite_subscription_quota: Some(16),
+ max_lite_topic_size: Some(max_lite_topic_size),
+ ..Default::default()
+ })),
+ ..Default::default()
+ }
}
#[test]
@@ -410,4 +496,141 @@ mod tests {
assert!(manager.check_lite_subscription_quota(3).is_ok());
assert!(manager.check_lite_subscription_quota(4).is_err());
}
+
+ /// Regression test: the periodic-sync scheduler must stop when requested,
otherwise it
+ /// keeps issuing `SyncLiteSubscription` RPCs (and creating sessions)
forever after the
+ /// consumer has been shut down.
+ #[tokio::test]
+ async fn test_stop_scheduler_cancels_periodic_sync() {
+ let sync_calls = Arc::new(AtomicUsize::new(0));
+
+ // The scheduler only issues the RPC when the subscription set is not
empty; every
+ // attempt goes through `get_session_for_lite_consumer`, so counting
that call is
+ // enough. It returns an error to avoid mocking the whole session
layer.
+ let mut client = Client::default();
+ {
+ let sync_calls = Arc::clone(&sync_calls);
+ client
+ .expect_get_session_for_lite_consumer()
+ .times(1..)
+ .returning(move || {
+ sync_calls.fetch_add(1, Ordering::SeqCst);
+ Err(ClientError::new(
+ ErrorKind::ClientInternal,
+ "no session in test",
+ OPERATION_SYNC_LITE_SUBSCRIPTION,
+ ))
+ });
+ }
+ let client = Arc::new(client);
+
+ let mut manager = LiteSubscriptionManager::new(
+ client,
+ "bind_topic".to_string(),
+ "namespace".to_string(),
+ "group".to_string(),
+ );
+ manager.periodic_sync_interval = Duration::from_millis(50);
+
+ // Start with an empty subscription set so the initial sync inside
`start()` is a no-op
+ // and succeeds without any server interaction.
+ manager.start().await.expect("manager should start");
+
+ // Subscribe from "outside": the scheduler now has a topic to sync and
should start
+ // issuing RPCs.
+ manager.lite_topic_set.lock().insert("topic1".to_string());
+
+ // Give the scheduler a chance to tick, so the test proves it was
running.
+ tokio::time::sleep(Duration::from_millis(120)).await;
+ let calls_before_stop = sync_calls.load(Ordering::SeqCst);
+ assert!(
+ calls_before_stop >= 1,
+ "the scheduler should keep issuing syncs while running"
+ );
+
+ manager.stop_scheduler().await;
+
+ // Keep the runtime alive for much longer than one sync interval: no
further RPC may
+ // happen now that the scheduler has been cancelled.
+ tokio::time::sleep(Duration::from_millis(250)).await;
+ let calls_after_stop = sync_calls.load(Ordering::SeqCst);
+ assert_eq!(
+ calls_before_stop, calls_after_stop,
+ "no sync RPC may be issued after the scheduler was stopped"
+ );
+ }
+
+ /// Regression test: the server-provided `max_lite_topic_size` must be
visible before the
+ /// first `subscribe_lite`, otherwise a topic that is valid on the server
is rejected
+ /// locally against the built-in default of 64.
+ #[tokio::test]
+ async fn test_initial_settings_are_applied_before_first_subscribe() {
+ let mut client = Client::default();
+ {
+ // Validation passes, quota passes, then the RPC itself fails in
this offline test.
+ client
+ .expect_get_session_for_lite_consumer()
+ .times(1..)
+ .returning(|| {
+ Err(ClientError::new(
+ ErrorKind::ClientInternal,
+ "no session in test",
+ OPERATION_SYNC_LITE_SUBSCRIPTION,
+ ))
+ });
+ }
+ let client = Arc::new(client);
+
+ let manager = LiteSubscriptionManager::new(
+ client,
+ "bind_topic".to_string(),
+ "namespace".to_string(),
+ "group".to_string(),
+ );
+
+ // 80 ASCII bytes: longer than the built-in default (64), shorter than
the server
+ // limit used below (128).
+ let lite_topic = "a".repeat(80);
+
+ // Before any settings arrived, `wait_for_initial_settings` must time
out.
+ assert!(
+ !manager
+ .wait_for_initial_settings(Duration::from_millis(50))
+ .await,
+ "initial settings must not be reported as ready before they
arrived"
+ );
+
+ // And the topic is rejected locally against the default max size.
+ let err = manager
+ .subscribe_lite(lite_topic.clone(), None)
+ .await
+ .err()
+ .expect("subscribe must fail while the default max size is in
effect");
+ assert_eq!(*err.kind(), ErrorKind::Config);
+ assert!(err.message().contains("max length"));
+
+ // The server reports its settings (limit 128 > 80).
+ manager.sync_settings(&test_settings(128));
+
+ assert!(
+ manager
+ .wait_for_initial_settings(Duration::from_millis(50))
+ .await,
+ "initial settings must be reported as ready right after
sync_settings"
+ );
+
+ // Now validation passes; the failure (if any) must come from the RPC
layer, not from
+ // the local length check.
+ match manager.subscribe_lite(lite_topic, None).await {
+ Ok(()) => {}
+ Err(err) => {
+ assert_ne!(
+ *err.kind(),
+ ErrorKind::Config,
+ "the topic must not be rejected by local validation
anymore, got: {}",
+ err.message()
+ );
+ }
+ }
+ }
}
diff --git a/rust/src/model/common.rs b/rust/src/model/common.rs
index 4e701447..7a8669e6 100644
--- a/rust/src/model/common.rs
+++ b/rust/src/model/common.rs
@@ -37,6 +37,7 @@ pub(crate) enum ClientType {
#[allow(dead_code)]
PullConsumer = 4,
LitePushConsumer = 5,
+ LiteSimpleConsumer = 6,
}
#[derive(Debug)]
diff --git a/rust/src/model/message.rs b/rust/src/model/message.rs
index b7aaf8a4..f16056ad 100644
--- a/rust/src/model/message.rs
+++ b/rust/src/model/message.rs
@@ -525,6 +525,14 @@ pub trait AckMessageEntry {
fn message_id(&self) -> String;
fn receipt_handle(&self) -> String;
fn endpoints(&self) -> &Endpoints;
+
+ /// Lite topic the message belongs to, `None` for non-lite messages.
+ ///
+ /// Lite messages MUST be acknowledged with their lite topic, otherwise
the proxy is unable
+ /// to resolve the LMQ receipt handle and answers `INTERNAL_SERVER_ERROR`.
+ fn lite_topic(&self) -> Option<&str> {
+ None
+ }
}
/// [`MessageView`] is the data model for receive message.
@@ -567,6 +575,10 @@ impl AckMessageEntry for MessageView {
fn endpoints(&self) -> &Endpoints {
&self.endpoints
}
+
+ fn lite_topic(&self) -> Option<&str> {
+ self.lite_topic.as_deref()
+ }
}
impl MessageView {
diff --git a/rust/src/simple_consumer.rs b/rust/src/simple_consumer.rs
index 0584b414..629953bd 100644
--- a/rust/src/simple_consumer.rs
+++ b/rust/src/simple_consumer.rs
@@ -31,7 +31,8 @@ use crate::model::message::{AckMessageEntry, MessageView};
use crate::pb::telemetry_command::Command;
use crate::session::SharedLongPollingTimeout;
use crate::util::{
- build_endpoints_by_message_queue, build_simple_consumer_settings,
select_message_queue,
+ build_endpoints_by_message_queue, build_simple_consumer_settings,
prune_lite_route,
+ select_message_queue,
};
/// [`SimpleConsumer`] is a lightweight consumer to consume messages from
RocketMQ proxy.
@@ -82,8 +83,38 @@ impl SimpleConsumer {
})
}
- /// Start the simple consumer
- pub async fn start(&mut self) -> Result<(), ClientError> {
+ /// Create a simple consumer instance on top of an already built client.
+ ///
+ /// Used by [`crate::lite_simple_consumer::LiteSimpleConsumer`], which has
to pass a client
+ /// whose settings carry the lite client type.
+ pub(crate) fn new_with_client(
+ client: Client,
+ option: SimpleConsumerOption,
+ ) -> Result<Self, ClientError> {
+ if option.consumer_group().is_empty() {
+ return Err(ClientError::new(
+ ErrorKind::Config,
+ "required option is missing: consumer group is empty",
+ Self::OPERATION_NEW_SIMPLE_CONSUMER,
+ ));
+ }
+ Ok(SimpleConsumer {
+ option,
+ client,
+ shutdown_tx: None,
+ })
+ }
+
+ /// Check whether the underlying client is started.
+ pub(crate) fn check_started(&self, operation: &'static str) -> Result<(),
ClientError> {
+ self.client.check_started(operation)
+ }
+
+ /// Start the simple consumer, forwarding server telemetry commands to
`telemetry_command_tx`.
+ pub(crate) async fn start_with_telemetry(
+ &mut self,
+ telemetry_command_tx:
mpsc::Sender<crate::pb::telemetry_command::Command>,
+ ) -> Result<(), ClientError> {
if self.option.consumer_group().is_empty() {
return Err(ClientError::new(
ErrorKind::Config,
@@ -91,13 +122,23 @@ impl SimpleConsumer {
Self::OPERATION_START_SIMPLE_CONSUMER,
));
}
- let (telemetry_command_tx, mut telemetry_command_rx) =
mpsc::channel(16);
self.client.start(telemetry_command_tx).await?;
if let Some(topics) = self.option.topics() {
for topic in topics {
self.client.topic_route(topic, true).await?;
}
}
+ info!(
+ "start simple consumer success, client_id: {}",
+ self.client.client_id()
+ );
+ Ok(())
+ }
+
+ /// Start the simple consumer
+ pub async fn start(&mut self) -> Result<(), ClientError> {
+ let (telemetry_command_tx, mut telemetry_command_rx) =
mpsc::channel(16);
+ self.start_with_telemetry(telemetry_command_tx).await?;
let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
self.shutdown_tx = Some(shutdown_tx);
let long_polling_timeout = self.client.long_polling_timeout_handle();
@@ -163,6 +204,14 @@ impl SimpleConsumer {
}
}
+ /// Same as [`SimpleConsumer::shutdown`] but takes a mutable reference.
+ pub(crate) async fn shutdown_ref(&mut self) -> Result<(), ClientError> {
+ if let Some(shutdown_tx) = self.shutdown_tx.take() {
+ let _ = shutdown_tx.send(());
+ };
+ self.client.shutdown_ref().await
+ }
+
/// receive messages from the specified topic
///
/// # Arguments
@@ -193,7 +242,51 @@ impl SimpleConsumer {
batch_size: i32,
invisible_duration: Duration,
) -> Result<Vec<MessageView>, ClientError> {
- let route = self.client.topic_route(topic.as_ref(), true).await?;
+ self.receive_internal(
+ topic.as_ref(),
+ expression,
+ batch_size,
+ invisible_duration,
+ false,
+ )
+ .await
+ }
+
+ /// Receive messages of a lite consumer.
+ ///
+ /// The route is pruned to the first readable master queue because the
server resolves the
+ /// lite topic queues itself (reference Java: `LiteSimpleConsumerImpl`).
+ pub(crate) async fn receive_lite(
+ &self,
+ topic: impl AsRef<str>,
+ expression: &FilterExpression,
+ batch_size: i32,
+ invisible_duration: Duration,
+ ) -> Result<Vec<MessageView>, ClientError> {
+ self.receive_internal(
+ topic.as_ref(),
+ expression,
+ batch_size,
+ invisible_duration,
+ true,
+ )
+ .await
+ }
+
+ async fn receive_internal(
+ &self,
+ topic: &str,
+ expression: &FilterExpression,
+ batch_size: i32,
+ invisible_duration: Duration,
+ lite: bool,
+ ) -> Result<Vec<MessageView>, ClientError> {
+ let route = self.client.topic_route(topic, true).await?;
+ let route = if lite {
+ prune_lite_route(route, topic, Self::OPERATION_RECEIVE_MESSAGE)?
+ } else {
+ route
+ };
let message_queue = select_message_queue(route);
let endpoints =
build_endpoints_by_message_queue(&message_queue,
Self::OPERATION_RECEIVE_MESSAGE)?;
diff --git a/rust/src/util.rs b/rust/src/util.rs
index 1355b11e..31dcc12c 100644
--- a/rust/src/util.rs
+++ b/rust/src/util.rs
@@ -24,6 +24,7 @@ use siphasher::sip::SipHasher24;
use crate::conf::{ProducerOption, PushConsumerOption, SimpleConsumerOption};
use crate::error::{ClientError, ErrorKind};
use crate::model::common::{ClientType, Endpoints, Route};
+use crate::model::message::AckMessageEntry;
use crate::pb::settings::PubSub;
use crate::pb::telemetry_command::Command;
use crate::pb::{
@@ -55,6 +56,72 @@ pub(crate) fn select_message_queue_by_message_group(
route.queue[index as usize].clone()
}
+/// Check whether a message queue is readable and served by a master broker.
+///
+/// Reference Java:
`SubscriptionLoadBalancer.isReadableMasterQueue(MessageQueueImpl)`
+fn is_readable_master_queue(message_queue: &MessageQueue) -> bool {
+ // Queues served by a slave broker are never readable.
+ if message_queue.broker.as_ref().map(|broker| broker.id) != Some(0) {
+ return false;
+ }
+ match crate::pb::Permission::from_i32(message_queue.permission) {
+ // `Unspecified` is tolerated: some route responses omit the
permission field.
+ Some(crate::pb::Permission::Unspecified)
+ | Some(crate::pb::Permission::Read)
+ | Some(crate::pb::Permission::ReadWrite) => true,
+ _ => false,
+ }
+}
+
+/// Keep only the first readable master queue of a route.
+///
+/// Lite consumers only need a route to *a* broker of the parent topic: the
proxy resolves the
+/// real lite topic queues on the server side. Mirrors
+/// `LiteSimpleConsumerImpl#updateSubscriptionLoadBalancer` in Java.
+pub(crate) fn prune_lite_route(
+ route: Arc<Route>,
+ topic: &str,
+ operation: &'static str,
+) -> Result<Arc<Route>, ClientError> {
+ let queue = route
+ .queue
+ .iter()
+ .find(|message_queue| is_readable_master_queue(message_queue))
+ .cloned()
+ .map(|message_queue| vec![message_queue])
+ .unwrap_or_default();
+
+ if queue.is_empty() {
+ return Err(ClientError::new(
+ ErrorKind::NoBrokerAvailable,
+ &format!(
+ "no readable message queue found for lite topic route,
topic={}",
+ topic
+ ),
+ operation,
+ ));
+ }
+
+ Ok(Arc::new(Route {
+ index: Default::default(),
+ queue,
+ }))
+}
+
+/// Build the [`AckMessageEntry`] protobuf of an ack entry.
+///
+/// `lite_topic` MUST be carried for lite messages, otherwise the proxy is
unable to resolve the
+/// LMQ receipt handle and answers `INTERNAL_SERVER_ERROR`.
+pub(crate) fn build_ack_message_entry<T: AckMessageEntry + ?Sized>(
+ ack_entry: &T,
+) -> crate::pb::AckMessageEntry {
+ crate::pb::AckMessageEntry {
+ message_id: ack_entry.message_id(),
+ receipt_handle: ack_entry.receipt_handle(),
+ lite_topic: ack_entry.lite_topic().map(|topic| topic.to_string()),
+ }
+}
+
pub(crate) fn build_endpoints_by_message_queue(
message_queue: &MessageQueue,
operation: &'static str,
@@ -157,6 +224,62 @@ pub(crate) fn build_simple_consumer_settings(option:
&SimpleConsumerOption) -> T
}
}
+/// Build the settings of a `LiteSimpleConsumer`.
+///
+/// Same as [`build_simple_consumer_settings`] except that the client type is
+/// `LITE_SIMPLE_CONSUMER` and the subscription table carries the bind topic
with a `SUB_ALL`
+/// filter expression (reference Java: `LiteSimpleConsumerBuilderImpl`).
+pub(crate) fn build_lite_simple_consumer_settings(
+ option: &SimpleConsumerOption,
+ bind_topic: &str,
+) -> TelemetryCommand {
+ let platform = os_info::get();
+ TelemetryCommand {
+ command: Some(Command::Settings(Settings {
+ client_type: Some(ClientType::LiteSimpleConsumer as i32),
+ request_timeout: Some(prost_types::Duration {
+ seconds: option.timeout().as_secs() as i64,
+ nanos: option.timeout().subsec_nanos() as i32,
+ }),
+ pub_sub: Some(PubSub::Subscription(Subscription {
+ group: Some(Resource {
+ name: option.consumer_group().to_string(),
+ resource_namespace: option.namespace().to_string(),
+ }),
+ subscriptions: vec![SubscriptionEntry {
+ topic: Some(Resource {
+ name: bind_topic.to_string(),
+ resource_namespace: option.namespace().to_string(),
+ }),
+ expression: Some(FilterExpression {
+ expression:
crate::model::common::FilterExpression::sub_all()
+ .expression()
+ .to_string(),
+ r#type:
crate::model::common::FilterExpression::sub_all().filter_type()
+ as i32,
+ }),
+ }],
+ fifo: Some(false),
+ receive_batch_size: None,
+ long_polling_timeout: Some(prost_types::Duration {
+ seconds: option.long_polling_timeout().as_secs() as i64,
+ nanos: option.long_polling_timeout().subsec_nanos() as i32,
+ }),
+ lite_subscription_quota: None,
+ max_lite_topic_size: None,
+ })),
+ user_agent: Some(Ua {
+ language: SDK_LANGUAGE as i32,
+ version: SDK_VERSION.to_string(),
+ platform: format!("{} {}", platform.os_type(),
platform.version()),
+ hostname: HOST_NAME.clone(),
+ }),
+ ..Settings::default()
+ })),
+ ..TelemetryCommand::default()
+ }
+}
+
pub(crate) fn build_push_consumer_settings(option: &PushConsumerOption) ->
TelemetryCommand {
let subscriptions: Vec<SubscriptionEntry> = option
.subscription_expressions()
@@ -481,6 +604,135 @@ mod tests {
build_simple_consumer_settings(&SimpleConsumerOption::default());
}
+ #[test]
+ fn util_build_lite_simple_consumer_settings() {
+ let mut option = SimpleConsumerOption::default();
+ option.set_consumer_group("test_group");
+ let settings = build_lite_simple_consumer_settings(&option,
"parent_topic");
+ match settings.command {
+ Some(Command::Settings(settings)) => {
+ assert_eq!(
+ settings.client_type,
+ Some(ClientType::LiteSimpleConsumer as i32)
+ );
+ match settings.pub_sub {
+ Some(PubSub::Subscription(subscription)) => {
+ assert_eq!(subscription.subscriptions.len(), 1);
+ assert_eq!(
+
subscription.subscriptions[0].topic.as_ref().unwrap().name,
+ "parent_topic"
+ );
+ let expression =
subscription.subscriptions[0].expression.as_ref().unwrap();
+ assert_eq!(expression.expression, "*");
+ }
+ other => panic!("expect a subscription settings, got
{:?}", other),
+ }
+ }
+ other => panic!("expect a settings command, got {:?}", other),
+ }
+ }
+
+ fn message_queue(id: i32, broker_id: i32, permission:
crate::pb::Permission) -> MessageQueue {
+ MessageQueue {
+ topic: Some(Resource {
+ name: "parent_topic".to_string(),
+ resource_namespace: "".to_string(),
+ }),
+ id,
+ permission: permission as i32,
+ broker: Some(Broker {
+ name: "broker-a".to_string(),
+ id: broker_id,
+ endpoints: Some(pb::Endpoints {
+ scheme: pb::AddressScheme::DomainName as i32,
+ addresses: vec![],
+ }),
+ }),
+ accept_message_types: vec![],
+ }
+ }
+
+ #[test]
+ fn util_prune_lite_route_keeps_first_readable_master_queue() {
+ let route = Arc::new(Route {
+ index: Default::default(),
+ queue: vec![
+ message_queue(0, 1, crate::pb::Permission::ReadWrite), // slave
+ message_queue(1, 0, crate::pb::Permission::Write), //
write only master
+ message_queue(2, 0, crate::pb::Permission::ReadWrite), //
first readable master
+ message_queue(3, 0, crate::pb::Permission::ReadWrite),
+ ],
+ });
+ let pruned = prune_lite_route(route, "parent_topic", "test").unwrap();
+ assert_eq!(pruned.queue.len(), 1);
+ assert_eq!(pruned.queue[0].id, 2);
+ }
+
+ #[test]
+ fn util_prune_lite_route_fails_without_readable_queue() {
+ let route = Arc::new(Route {
+ index: Default::default(),
+ queue: vec![message_queue(0, 0, crate::pb::Permission::None)],
+ });
+ let result = prune_lite_route(route, "parent_topic", "test");
+ assert!(result.is_err());
+ let err = result.err().unwrap();
+ assert_eq!(err.kind, ErrorKind::NoBrokerAvailable);
+
+ let empty = Arc::new(Route {
+ index: Default::default(),
+ queue: vec![],
+ });
+ assert!(prune_lite_route(empty, "parent_topic", "test").is_err());
+ }
+
+ #[test]
+ fn util_build_ack_message_entry_carries_lite_topic() {
+ let message_view = crate::model::message::MessageView::from_pb_message(
+ crate::pb::Message {
+ topic: Some(Resource {
+ name: "parent_topic".to_string(),
+ ..Default::default()
+ }),
+ body: vec![1, 2, 3],
+ user_properties: Default::default(),
+ system_properties: Some(crate::pb::SystemProperties {
+ message_id: "lite_msg_id".to_string(),
+ receipt_handle: Some("receipt_handle".to_string()),
+ lite_topic: Some("lite_topic_001".to_string()),
+ ..Default::default()
+ }),
+ },
+ Endpoints::from_url("localhost:8081").unwrap(),
+ )
+ .unwrap();
+
+ let entry = build_ack_message_entry(&message_view);
+ assert_eq!(entry.message_id, "lite_msg_id");
+ assert_eq!(entry.receipt_handle, "receipt_handle");
+ assert_eq!(entry.lite_topic, Some("lite_topic_001".to_string()));
+
+ // A normal message must not carry a lite topic.
+ let message_view = crate::model::message::MessageView::from_pb_message(
+ crate::pb::Message {
+ topic: Some(Resource {
+ name: "normal_topic".to_string(),
+ ..Default::default()
+ }),
+ body: vec![1, 2, 3],
+ user_properties: Default::default(),
+ system_properties: Some(crate::pb::SystemProperties {
+ message_id: "normal_msg_id".to_string(),
+ receipt_handle: Some("receipt_handle".to_string()),
+ ..Default::default()
+ }),
+ },
+ Endpoints::from_url("localhost:8081").unwrap(),
+ )
+ .unwrap();
+ assert_eq!(build_ack_message_entry(&message_view).lite_topic, None);
+ }
+
#[test]
fn test_handle_response_status() {
let result = handle_response_status(None, "test");
diff --git a/rust/tests/lite_simple_consumer_integration_test.rs
b/rust/tests/lite_simple_consumer_integration_test.rs
new file mode 100644
index 00000000..cb4f6acf
--- /dev/null
+++ b/rust/tests/lite_simple_consumer_integration_test.rs
@@ -0,0 +1,289 @@
+/*
+ * 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.
+ */
+
+//! Real cluster integration tests for LiteSimpleConsumer.
+//!
+//! These tests need a running RocketMQ deployment:
+//! * broker started with `enableLmq=true` and `enableMultiDispatch=true`
+//! * a parent topic created with `message.type=LITE`
+//! * a consumer group created with `+lite.bind.topic=<parentTopic>`
+//! * a proxy running in CLUSTER mode
+//!
+//! They are skipped unless `ROCKETMQ_RUST_LITE_ENDPOINTS` is set, e.g.
+//! ```shell
+//! export ROCKETMQ_RUST_LITE_ENDPOINTS=127.0.0.1:8081
+//! export ROCKETMQ_RUST_LITE_PARENT_TOPIC=lite-parent-topic
+//! export ROCKETMQ_RUST_LITE_GROUP=rust-lite-unittest-group
+//! cargo test --test lite_simple_consumer_integration_test
+//! ```
+
+#[cfg(test)]
+mod lite_simple_consumer_integration_tests {
+ use std::time::Duration;
+
+ use rocketmq::conf::{ClientOption, ProducerOption, SimpleConsumerOption};
+ use rocketmq::error::ErrorKind;
+ use rocketmq::model::message::MessageBuilder;
+ use rocketmq::model::offset_option::{OffsetOption, OffsetPolicy};
+ use rocketmq::{LiteSimpleConsumer, LiteSimpleConsumerTrait, Producer};
+
+ struct Config {
+ endpoints: String,
+ parent_topic: String,
+ group: String,
+ }
+
+ impl Config {
+ fn from_env() -> Option<Self> {
+ Some(Self {
+ endpoints: std::env::var("ROCKETMQ_RUST_LITE_ENDPOINTS").ok()?,
+ parent_topic: std::env::var("ROCKETMQ_RUST_LITE_PARENT_TOPIC")
+ .unwrap_or("lite-parent-topic".to_string()),
+ group: std::env::var("ROCKETMQ_RUST_LITE_GROUP")
+ .unwrap_or("rust-lite-unittest-group".to_string()),
+ })
+ }
+
+ fn client_option(&self) -> ClientOption {
+ let mut client_option = ClientOption::default();
+ client_option.set_access_url(&self.endpoints);
+ client_option
+ }
+
+ fn consumer_option(&self) -> SimpleConsumerOption {
+ let mut option = SimpleConsumerOption::default();
+ option.set_consumer_group(&self.group);
+ option
+ }
+ }
+
+ async fn send_lite_messages(config: &Config, lite_topic: &str, num: usize)
-> Vec<String> {
+ let mut producer_option = ProducerOption::default();
+ producer_option.set_topics(vec![config.parent_topic.clone()]);
+ let mut producer =
+ Producer::new(producer_option,
config.client_option()).expect("create producer");
+ producer.start().await.expect("start producer");
+
+ let mut message_ids = Vec::with_capacity(num);
+ for i in 0..num {
+ let message = MessageBuilder::lite_message_builder(
+ config.parent_topic.clone(),
+ format!("rust-lite-simple-consumer-{}", i).into_bytes(),
+ lite_topic,
+ )
+ .build()
+ .expect("build lite message");
+ let receipt = producer.send(message).await.expect("send lite
message");
+ message_ids.push(receipt.message_id().to_string());
+ }
+ producer.shutdown().await.expect("shutdown producer");
+ message_ids
+ }
+
+ /// Receive every message of a lite topic and ack it.
+ #[tokio::test]
+ async fn test_receive_and_ack() {
+ let Some(config) = Config::from_env() else {
+ println!("ROCKETMQ_RUST_LITE_ENDPOINTS is not set, skip the
integration test");
+ return;
+ };
+
+ let lite_topic = format!("rust-lite-ack-{}", chrono_like_nonce());
+ let expected = send_lite_messages(&config, &lite_topic, 5).await;
+
+ let mut consumer = LiteSimpleConsumer::new(
+ config.client_option(),
+ config.consumer_option(),
+ config.parent_topic.clone(),
+ )
+ .expect("create lite simple consumer");
+ consumer.start().await.expect("start lite simple consumer");
+ consumer
+ .subscribe_lite_with_offset(
+ lite_topic.clone(),
+ OffsetOption::from_policy(OffsetPolicy::Min),
+ )
+ .await
+ .expect("subscribe lite topic");
+ assert!(consumer.get_lite_topic_set().contains(&lite_topic));
+
+ let mut received = Vec::new();
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+ while tokio::time::Instant::now() < deadline && received.len() <
expected.len() {
+ let messages = consumer
+ .receive(32, Duration::from_secs(15))
+ .await
+ .expect("receive messages");
+ for message in messages {
+ // The lite topic must be present, it is required to ack the
message.
+ assert_eq!(message.lite_topic(), Some(lite_topic.as_str()));
+ consumer.ack(&message).await.expect("ack message");
+ received.push(message.message_id().to_string());
+ }
+ }
+
+ for id in &expected {
+ assert!(received.contains(id), "message {} was not received", id);
+ }
+
+ consumer
+ .unsubscribe_lite(lite_topic.clone())
+ .await
+ .expect("unsubscribe lite topic");
+ assert!(!consumer.get_lite_topic_set().contains(&lite_topic));
+ consumer.shutdown().await.expect("shutdown consumer");
+ }
+
+ /// Messages sent after the subscription must be delivered too.
+ #[tokio::test]
+ async fn test_receive_after_subscribe() {
+ let Some(config) = Config::from_env() else {
+ println!("ROCKETMQ_RUST_LITE_ENDPOINTS is not set, skip the
integration test");
+ return;
+ };
+
+ let lite_topic = format!("rust-lite-live-{}", chrono_like_nonce());
+ let mut consumer = LiteSimpleConsumer::new(
+ config.client_option(),
+ config.consumer_option(),
+ config.parent_topic.clone(),
+ )
+ .expect("create lite simple consumer");
+ consumer.start().await.expect("start lite simple consumer");
+ consumer
+ .subscribe_lite(lite_topic.clone())
+ .await
+ .expect("subscribe lite topic");
+
+ let expected = send_lite_messages(&config, &lite_topic, 3).await;
+ let mut received = Vec::new();
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+ while tokio::time::Instant::now() < deadline && received.len() <
expected.len() {
+ let messages = consumer
+ .receive(32, Duration::from_secs(15))
+ .await
+ .expect("receive messages");
+ for message in messages {
+ consumer.ack(&message).await.expect("ack message");
+ received.push(message.message_id().to_string());
+ }
+ }
+
+ assert!(
+ received.iter().any(|id| expected.contains(id)),
+ "no message of the live lite topic was received"
+ );
+
+ consumer.shutdown().await.expect("shutdown consumer");
+ }
+
+ /// A stable nonce without extra dependencies.
+ fn chrono_like_nonce() -> u128 {
+ std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .unwrap()
+ .as_nanos()
+ }
+
+ /// `ack()` / `change_invisible_duration()` must report
`ClientIsNotRunning` after the
+ /// consumer was shut down instead of panicking: `shutdown(&mut self)`
leaves the consumer
+ /// accessible, and a caller may still hold a received `MessageView`.
+ #[tokio::test]
+ async fn test_ack_and_change_invisible_duration_after_shutdown() {
+ let Some(config) = Config::from_env() else {
+ println!("ROCKETMQ_RUST_LITE_ENDPOINTS is not set, skip the
integration test");
+ return;
+ };
+
+ let lite_topic = format!("rust-lite-shutdown-{}", chrono_like_nonce());
+ send_lite_messages(&config, &lite_topic, 1).await;
+
+ let mut consumer = LiteSimpleConsumer::new(
+ config.client_option(),
+ config.consumer_option(),
+ config.parent_topic.clone(),
+ )
+ .expect("create lite simple consumer");
+ consumer.start().await.expect("start lite simple consumer");
+ consumer
+ .subscribe_lite(lite_topic.clone())
+ .await
+ .expect("subscribe lite topic");
+
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+ let message = loop {
+ let messages = consumer
+ .receive(32, Duration::from_secs(15))
+ .await
+ .expect("receive messages");
+ if let Some(message) = messages.into_iter().next() {
+ break message;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "no message was received"
+ );
+ };
+
+ consumer.shutdown().await.expect("shutdown consumer");
+
+ // Both calls must fail with `ClientIsNotRunning`; before the
running-state guard was
+ // added they panicked on the missing telemetry channel instead.
+ let err = consumer
+ .ack(&message)
+ .await
+ .err()
+ .expect("ack must fail after shutdown");
+ assert_eq!(*err.kind(), ErrorKind::ClientIsNotRunning);
+
+ let err = consumer
+ .change_invisible_duration(&message, Duration::from_secs(1))
+ .await
+ .err()
+ .expect("changeInvisibleDuration must fail after shutdown");
+ assert_eq!(*err.kind(), ErrorKind::ClientIsNotRunning);
+ }
+
+ /// The first `subscribe_lite` right after `start()` must see the
server-provided
+ /// `max_lite_topic_size` (and quota), not the built-in defaults:
`start()` waits bounded
+ /// for the telemetry handler to apply the initial `Settings`.
+ #[tokio::test]
+ async fn test_immediate_first_subscribe_after_start() {
+ let Some(config) = Config::from_env() else {
+ println!("ROCKETMQ_RUST_LITE_ENDPOINTS is not set, skip the
integration test");
+ return;
+ };
+
+ let lite_topic = format!("rust-lite-first-{}", chrono_like_nonce());
+ let mut consumer = LiteSimpleConsumer::new(
+ config.client_option(),
+ config.consumer_option(),
+ config.parent_topic.clone(),
+ )
+ .expect("create lite simple consumer");
+
+ // No sleep, no warm-up: subscribe immediately after start.
+ consumer.start().await.expect("start lite simple consumer");
+ consumer
+ .subscribe_lite(lite_topic.clone())
+ .await
+ .expect("subscribe immediately after start");
+ assert!(consumer.get_lite_topic_set().contains(&lite_topic));
+
+ consumer.shutdown().await.expect("shutdown consumer");
+ }
+}
diff --git a/rust/tests/lite_simple_consumer_test.rs
b/rust/tests/lite_simple_consumer_test.rs
new file mode 100644
index 00000000..4f35283a
--- /dev/null
+++ b/rust/tests/lite_simple_consumer_test.rs
@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+
+//! Offline tests for LiteSimpleConsumer.
+//!
+//! These tests run without any RocketMQ deployment: they only assert the
public API contract
+//! (validation, accessors, life cycle checks) of `LiteSimpleConsumer`.
+
+#[cfg(test)]
+mod lite_simple_consumer_tests {
+ use rocketmq::conf::{ClientOption, SimpleConsumerOption};
+ use rocketmq::error::ErrorKind;
+ use rocketmq::model::offset_option::{OffsetOption, OffsetPolicy};
+ use rocketmq::{LiteSimpleConsumer, LiteSimpleConsumerTrait};
+
+ fn client_option() -> ClientOption {
+ let mut client_option = ClientOption::default();
+ client_option.set_access_url("http://localhost:8080";);
+ client_option
+ }
+
+ fn consumer_option(consumer_group: &str) -> SimpleConsumerOption {
+ let mut option = SimpleConsumerOption::default();
+ option.set_consumer_group(consumer_group);
+ option
+ }
+
+ #[test]
+ fn test_new_requires_consumer_group() {
+ let result = LiteSimpleConsumer::new(
+ client_option(),
+ consumer_option(""),
+ "parent_topic".to_string(),
+ );
+ let err = result.err().expect("consumer group is mandatory");
+ assert_eq!(*err.kind(), ErrorKind::Config);
+ assert_eq!(err.operation(), "lite_simple_consumer.new");
+ }
+
+ #[test]
+ fn test_new_requires_bind_topic() {
+ let result =
+ LiteSimpleConsumer::new(client_option(), consumer_option("group"),
"".to_string());
+ let err = result.err().expect("bind topic is mandatory");
+ assert_eq!(*err.kind(), ErrorKind::Config);
+ assert!(err.message().contains("bind topic"));
+ }
+
+ #[test]
+ fn test_new_succeeds_and_exposes_metadata() {
+ let consumer = LiteSimpleConsumer::new(
+ client_option(),
+ consumer_option("test_group"),
+ "parent_topic".to_string(),
+ )
+ .expect("consumer should be created");
+ assert_eq!(consumer.get_consumer_group(), "test_group");
+ assert_eq!(consumer.get_bind_topic(), "parent_topic");
+ assert!(consumer.get_lite_topic_set().is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_subscribe_lite_before_start_fails() {
+ let consumer = LiteSimpleConsumer::new(
+ client_option(),
+ consumer_option("test_group"),
+ "parent_topic".to_string(),
+ )
+ .expect("consumer should be created");
+
+ let err = consumer
+ .subscribe_lite("lite_topic".to_string())
+ .await
+ .err()
+ .expect("subscribeLite requires a started consumer");
+ assert_eq!(*err.kind(), ErrorKind::ClientIsNotRunning);
+
+ let err = consumer
+ .subscribe_lite_with_offset(
+ "lite_topic".to_string(),
+ OffsetOption::from_policy(OffsetPolicy::Min),
+ )
+ .await
+ .err()
+ .expect("subscribeLite requires a started consumer");
+ assert_eq!(*err.kind(), ErrorKind::ClientIsNotRunning);
+
+ let err = consumer
+ .unsubscribe_lite("lite_topic".to_string())
+ .await
+ .err()
+ .expect("unsubscribeLite requires a started consumer");
+ assert_eq!(*err.kind(), ErrorKind::ClientIsNotRunning);
+ }
+}
