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 47c2a640 [C++] Fix thread safety bugs and add comprehensive unit tests
(#1301)
47c2a640 is described below
commit 47c2a64028cd2f83457e00d081461ef13355f60a
Author: lizhimins <[email protected]>
AuthorDate: Fri Jul 10 11:47:43 2026 +0800
[C++] Fix thread safety bugs and add comprehensive unit tests (#1301)
* [C++] Fix thread safety bugs in TelemetryBidiReactor and ClientManager
* [C++] Add comprehensive unit tests for base, client, and rocketmq modules
---
cpp/README.md | 52 ++-
cpp/source/base/tests/BUILD.bazel | 35 ++
cpp/source/base/tests/CMakeLists.txt | 4 +
cpp/source/base/tests/FilterExpressionTest.cpp | 65 ++++
cpp/source/base/tests/NamingSchemeTest.cpp | 77 +++++
cpp/source/base/tests/ProtocolTest.cpp | 111 +++++++
cpp/source/base/tests/UtilAllTest.cpp | 66 ++++
cpp/source/client/ClientManagerImpl.cpp | 44 ++-
cpp/source/client/TelemetryBidiReactor.cpp | 93 +++++-
cpp/source/client/include/TelemetryBidiReactor.h | 5 +
cpp/source/client/tests/BUILD.bazel | 12 +
cpp/source/client/tests/CMakeLists.txt | 1 +
cpp/source/client/tests/ClientManagerTest.cpp | 123 +++++++
.../client/tests/TelemetryBidiReactorTest.cpp | 369 +++++++++++++++++++++
cpp/source/rocketmq/tests/BUILD.bazel | 73 +++-
cpp/source/rocketmq/tests/CMakeLists.txt | 7 +
cpp/source/rocketmq/tests/ClientImplTest.cpp | 127 +++++++
cpp/source/rocketmq/tests/ConsumeTaskTest.cpp | 323 ++++++++++++++++++
.../rocketmq/tests/FifoProducerPartitionTest.cpp | 366 ++++++++++++++++++++
cpp/source/rocketmq/tests/ProcessQueueImplTest.cpp | 249 ++++++++++++++
cpp/source/rocketmq/tests/ProducerImplTest.cpp | 284 ++++++++++++++++
cpp/source/rocketmq/tests/PushConsumerImplTest.cpp | 337 +++++++++++++++++++
cpp/source/rocketmq/tests/SendContextTest.cpp | 2 +-
.../rocketmq/tests/SimpleConsumerImplTest.cpp | 195 +++++++++++
cpp/source/rocketmq/tests/TopicPublishInfoTest.cpp | 152 +++++++++
25 files changed, 3120 insertions(+), 52 deletions(-)
diff --git a/cpp/README.md b/cpp/README.md
index c555e027..fa7bf588 100644
--- a/cpp/README.md
+++ b/cpp/README.md
@@ -11,7 +11,7 @@ The proto definitions are shared via the `protos/` git
submodule at the reposito
## Prerequisites
- C++ compiler supporting C++11
-- CMake 3.13+ or Bazel 5.2.0
+- CMake 3.16+ or Bazel 6.6.0
- gRPC — RPC communication framework, also brings in protobuf (serialization),
abseil (base library), and re2 (regex)
- OpenSSL development headers — TLS encrypted communication
- zlib development headers — message body compression
@@ -87,7 +87,7 @@ The proto definitions are shared via the `protos/` git
submodule at the reposito
| Option | Default | Description
|
| ---------------- | ------- |
---------------------------------------------------- |
- | `BUILD_TESTS` | ON | Build unit tests (requires googletest,
auto-fetched) |
+ | `BUILD_TESTS` | OFF | Build unit tests (requires googletest,
auto-fetched) |
| `BUILD_EXAMPLES` | ON | Build example programs (requires gflags)
|
To skip tests and examples:
@@ -96,6 +96,29 @@ The proto definitions are shared via the `protos/` git
submodule at the reposito
cmake -DBUILD_TESTS=OFF -DBUILD_EXAMPLES=OFF ..
```
+6. Install (optional):
+
+ ```shell
+ # Install to /usr/local (default)
+ sudo cmake --install .
+
+ # Or install to a custom prefix
+ cmake --install . --prefix /opt/rocketmq
+ ```
+
+ After installation, downstream projects can use `find_package(rocketmq)` in
their CMakeLists.txt:
+
+ ```cmake
+ find_package(rocketmq REQUIRED)
+ target_link_libraries(my_app PRIVATE rocketmq)
+ ```
+
+ To uninstall:
+
+ ```shell
+ sudo make uninstall
+ ```
+
### Build with Bazel
```shell
@@ -144,36 +167,35 @@ genhtml bazel-out/_coverage/_coverage_report.dat \
## Run Examples
-All commands should run from the `cpp/` directory.
+Examples are built to `build/examples/` (CMake) or run directly via `bazel
run` (Bazel).
### Publish messages
```shell
-# Standard messages (sync)
+# CMake (from build/examples/)
+./example_producer --topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT
--total=16
+./example_producer_with_async --topic=YOUR_TOPIC
--access_point=SERVICE_ACCESS_POINT --total=16
+./example_fifo_producer --topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT
--total=16
+
+# Bazel (from cpp/)
bazel run //examples:example_producer -- \
--topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT --total=16
-
-# Standard messages (async)
bazel run //examples:example_producer_with_async -- \
--topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT --total=16
-
-# FIFO messages
bazel run //examples:example_producer_with_fifo_message -- \
--topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT --total=16
-
-# Transactional messages
-bazel run //examples:example_producer_with_transactional_message -- \
- --topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT --total=16
```
### Consume messages
```shell
-# Push consumer (message listener)
+# CMake (from build/examples/)
+./example_push_consumer --topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT
--group=YOUR_GROUP_ID
+./example_simple_consumer --topic=YOUR_TOPIC
--access_point=SERVICE_ACCESS_POINT --group=YOUR_GROUP_ID
+
+# Bazel (from cpp/)
bazel run //examples:example_push_consumer -- \
--topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT --group=YOUR_GROUP_ID
-
-# Simple consumer (pull-based)
bazel run //examples:example_simple_consumer -- \
--topic=YOUR_TOPIC --access_point=SERVICE_ACCESS_POINT --group=YOUR_GROUP_ID
```
diff --git a/cpp/source/base/tests/BUILD.bazel
b/cpp/source/base/tests/BUILD.bazel
index 3159f362..f7bce137 100644
--- a/cpp/source/base/tests/BUILD.bazel
+++ b/cpp/source/base/tests/BUILD.bazel
@@ -85,4 +85,39 @@ cc_test(
"//source/client:client_library",
"@com_google_googletest//:gtest_main",
],
+)
+
+cc_test(
+ name = "naming_scheme_test",
+ srcs = [
+ "NamingSchemeTest.cpp",
+ ],
+ deps = [
+ "//source/rocketmq:rocketmq_library",
+ "@com_google_googletest//:gtest_main",
+ ],
+)
+
+cc_test(
+ name = "filter_expression_test",
+ srcs = [
+ "FilterExpressionTest.cpp",
+ ],
+ deps = base_deps,
+)
+
+cc_test(
+ name = "util_all_test",
+ srcs = [
+ "UtilAllTest.cpp",
+ ],
+ deps = base_deps,
+)
+
+cc_test(
+ name = "protocol_test",
+ srcs = [
+ "ProtocolTest.cpp",
+ ],
+ deps = base_deps,
)
\ No newline at end of file
diff --git a/cpp/source/base/tests/CMakeLists.txt
b/cpp/source/base/tests/CMakeLists.txt
index ba5ea068..50a01dc6 100644
--- a/cpp/source/base/tests/CMakeLists.txt
+++ b/cpp/source/base/tests/CMakeLists.txt
@@ -14,3 +14,7 @@ add_base_test(mix_all_test MixAllTest.cpp)
add_base_test(retry_policy_test RetryPolicyTest.cpp)
add_base_test(thread_pool_test ThreadPoolTest.cpp)
add_base_test(unique_id_generator_test UniqueIdGeneratorTest.cpp)
+add_base_test(naming_scheme_test NamingSchemeTest.cpp)
+add_base_test(filter_expression_test FilterExpressionTest.cpp)
+add_base_test(util_all_test UtilAllTest.cpp)
+add_base_test(protocol_test ProtocolTest.cpp)
diff --git a/cpp/source/base/tests/FilterExpressionTest.cpp
b/cpp/source/base/tests/FilterExpressionTest.cpp
new file mode 100644
index 00000000..07094703
--- /dev/null
+++ b/cpp/source/base/tests/FilterExpressionTest.cpp
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+#include "gtest/gtest.h"
+#include "rocketmq/FilterExpression.h"
+#include "rocketmq/Message.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+class FilterExpressionTest : public testing::Test {
+protected:
+ static MessageConstPtr makeMessage(const std::string& topic, const
std::string& tag) {
+ return
Message::newBuilder().withTopic(topic).withTag(tag).withBody("body").build();
+ }
+};
+
+TEST_F(FilterExpressionTest, wildcardTagAcceptsAnyMessage) {
+ FilterExpression expr("*", ExpressionType::TAG);
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "tagA")));
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "tagB")));
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "")));
+}
+
+TEST_F(FilterExpressionTest, emptyTagBecomesWildcard) {
+ FilterExpression expr("", ExpressionType::TAG);
+ EXPECT_EQ("*", expr.content_);
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "anyTag")));
+}
+
+TEST_F(FilterExpressionTest, exactTagMatch) {
+ FilterExpression expr("tagA", ExpressionType::TAG);
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "tagA")));
+ EXPECT_FALSE(expr.accept(*makeMessage("topic", "tagB")));
+ EXPECT_FALSE(expr.accept(*makeMessage("topic", "")));
+}
+
+TEST_F(FilterExpressionTest, sql92AlwaysAccepts) {
+ // SQL92 filtering is done server-side; client always accepts
+ FilterExpression expr("price > 100", ExpressionType::SQL92);
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "anyTag")));
+ EXPECT_TRUE(expr.accept(*makeMessage("topic", "")));
+}
+
+TEST_F(FilterExpressionTest, versionIsSetOnConstruction) {
+ auto before = std::chrono::steady_clock::now();
+ FilterExpression expr("tag", ExpressionType::TAG);
+ auto after = std::chrono::steady_clock::now();
+ EXPECT_GE(expr.version_, before);
+ EXPECT_LE(expr.version_, after);
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/base/tests/NamingSchemeTest.cpp
b/cpp/source/base/tests/NamingSchemeTest.cpp
new file mode 100644
index 00000000..94d0d861
--- /dev/null
+++ b/cpp/source/base/tests/NamingSchemeTest.cpp
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+#include <string>
+#include <vector>
+
+#include "NamingScheme.h"
+#include "absl/strings/match.h"
+#include "gtest/gtest.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+class NamingSchemeTest : public testing::Test {
+protected:
+ NamingScheme naming_scheme;
+};
+
+TEST_F(NamingSchemeTest, acceptDnsPrefix) {
+ EXPECT_TRUE(naming_scheme.accept("dns:example.com:8080"));
+}
+
+TEST_F(NamingSchemeTest, acceptIPv4Prefix) {
+ EXPECT_TRUE(naming_scheme.accept("ipv4:10.0.0.1:8080"));
+}
+
+TEST_F(NamingSchemeTest, acceptIPv6Prefix) {
+ EXPECT_TRUE(naming_scheme.accept("ipv6:fe80::1:8080"));
+}
+
+TEST_F(NamingSchemeTest, rejectUnknownPrefix) {
+ EXPECT_FALSE(naming_scheme.accept("http://example.com";));
+ EXPECT_FALSE(naming_scheme.accept("10.0.0.1:8080"));
+ EXPECT_FALSE(naming_scheme.accept(""));
+}
+
+TEST_F(NamingSchemeTest, buildAddressFromIPv4List) {
+ std::vector<std::string> list = {"10.0.0.1:8080", "10.0.0.2:9090"};
+ std::string result = naming_scheme.buildAddress(list);
+ EXPECT_TRUE(absl::StartsWith(result, "ipv4:"));
+ // Both addresses should appear
+ EXPECT_NE(std::string::npos, result.find("10.0.0.1:8080"));
+ EXPECT_NE(std::string::npos, result.find("10.0.0.2:9090"));
+}
+
+TEST_F(NamingSchemeTest, buildAddressFromDnsShortCircuits) {
+ // DNS record found → return immediately, ignoring IPv4 entries
+ std::vector<std::string> list = {"10.0.0.1:8080", "broker.example.com:9876"};
+ std::string result = naming_scheme.buildAddress(list);
+ EXPECT_EQ("dns:broker.example.com:9876", result);
+}
+
+TEST_F(NamingSchemeTest, buildAddressSkipsMalformedEntries) {
+ std::vector<std::string> list = {"no-port", "10.0.0.1:8080"};
+ std::string result = naming_scheme.buildAddress(list);
+ EXPECT_TRUE(absl::StartsWith(result, "ipv4:"));
+}
+
+TEST_F(NamingSchemeTest, buildAddressEmptyListReturnsEmpty) {
+ std::vector<std::string> list;
+ std::string result = naming_scheme.buildAddress(list);
+ EXPECT_TRUE(result.empty());
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/base/tests/ProtocolTest.cpp
b/cpp/source/base/tests/ProtocolTest.cpp
new file mode 100644
index 00000000..27e08b38
--- /dev/null
+++ b/cpp/source/base/tests/ProtocolTest.cpp
@@ -0,0 +1,111 @@
+/*
+ * 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.
+ */
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gtest/gtest.h"
+
+#include "Protocol.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace rmq = apache::rocketmq::v2;
+
+class ProtocolTest : public testing::Test {
+protected:
+ static rmq::MessageQueue makeQueue(const std::string& ns, const std::string&
topic, int id,
+ const std::string& broker,
rmq::Permission perm = rmq::Permission::READ_WRITE) {
+ rmq::MessageQueue mq;
+ mq.mutable_topic()->set_resource_namespace(ns);
+ mq.mutable_topic()->set_name(topic);
+ mq.set_id(id);
+ mq.mutable_broker()->set_name(broker);
+ mq.set_permission(perm);
+ return mq;
+ }
+
+ static rmq::MessageQueue makeQueueWithEndpoints(rmq::AddressScheme scheme,
+ const
std::vector<std::pair<std::string, int>>& addrs) {
+ rmq::MessageQueue mq;
+ mq.mutable_topic()->set_resource_namespace("ns");
+ mq.mutable_topic()->set_name("topic");
+ mq.set_id(0);
+ mq.mutable_broker()->set_name("broker-0");
+ mq.mutable_broker()->mutable_endpoints()->set_scheme(scheme);
+ for (auto& p : addrs) {
+ auto* addr = mq.mutable_broker()->mutable_endpoints()->add_addresses();
+ addr->set_host(p.first);
+ addr->set_port(p.second);
+ }
+ return mq;
+ }
+};
+
+TEST_F(ProtocolTest, writablePermissionTest) {
+ EXPECT_TRUE(writable(rmq::Permission::WRITE));
+ EXPECT_TRUE(writable(rmq::Permission::READ_WRITE));
+ EXPECT_FALSE(writable(rmq::Permission::READ));
+ EXPECT_FALSE(writable(rmq::Permission::NONE));
+}
+
+TEST_F(ProtocolTest, readablePermissionTest) {
+ EXPECT_TRUE(readable(rmq::Permission::READ));
+ EXPECT_TRUE(readable(rmq::Permission::READ_WRITE));
+ EXPECT_FALSE(readable(rmq::Permission::WRITE));
+ EXPECT_FALSE(readable(rmq::Permission::NONE));
+}
+
+TEST_F(ProtocolTest, messageQueueEqualityTest) {
+ auto a = makeQueue("ns", "topic", 0, "broker");
+ auto b = makeQueue("ns", "topic", 0, "broker");
+ EXPECT_TRUE(a == b);
+}
+
+TEST_F(ProtocolTest, messageQueueInequalityTest) {
+ auto a = makeQueue("ns", "topic", 0, "broker");
+ auto b = makeQueue("ns", "topic", 1, "broker");
+ EXPECT_FALSE(a == b);
+}
+
+TEST_F(ProtocolTest, urlOfDomainNameTest) {
+ auto mq = makeQueueWithEndpoints(rmq::AddressScheme::DOMAIN_NAME,
{{"broker.example.com", 8080}});
+ EXPECT_EQ("dns:broker.example.com:8080", urlOf(mq));
+}
+
+TEST_F(ProtocolTest, urlOfIPv4SingleTest) {
+ auto mq = makeQueueWithEndpoints(rmq::AddressScheme::IPv4, {{"10.0.0.1",
8080}});
+ EXPECT_EQ("ipv4:10.0.0.1:8080", urlOf(mq));
+}
+
+TEST_F(ProtocolTest, urlOfIPv4MultipleTest) {
+ auto mq = makeQueueWithEndpoints(rmq::AddressScheme::IPv4, {{"10.0.0.1",
8080}, {"10.0.0.2", 9090}});
+ std::string result = urlOf(mq);
+ EXPECT_EQ("ipv4:10.0.0.1:8080,10.0.0.2:9090", result);
+}
+
+TEST_F(ProtocolTest, urlOfIPv6Test) {
+ auto mq = makeQueueWithEndpoints(rmq::AddressScheme::IPv6, {{"fe80::1",
8080}});
+ EXPECT_EQ("ipv6:fe80::1:8080", urlOf(mq));
+}
+
+TEST_F(ProtocolTest, simpleNameOfFormatTest) {
+ auto mq = makeQueue("my-ns", "my-topic", 3, "broker-a");
+ EXPECT_EQ("my-ns@my-topic@3@broker-a", simpleNameOf(mq));
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/base/tests/UtilAllTest.cpp
b/cpp/source/base/tests/UtilAllTest.cpp
new file mode 100644
index 00000000..38aec123
--- /dev/null
+++ b/cpp/source/base/tests/UtilAllTest.cpp
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+#include "gtest/gtest.h"
+#include "UtilAll.h"
+
+#include <string>
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+class UtilAllTest : public testing::Test {};
+
+TEST_F(UtilAllTest, compressAndUncompressRoundTripTest) {
+ std::string original = "Hello, RocketMQ! This is a test of compression and
decompression.";
+ std::string compressed;
+ ASSERT_TRUE(UtilAll::compress(original, compressed));
+ EXPECT_FALSE(compressed.empty());
+ EXPECT_NE(original, compressed);
+
+ std::string decompressed;
+ ASSERT_TRUE(UtilAll::uncompress(compressed, decompressed));
+ EXPECT_EQ(original, decompressed);
+}
+
+TEST_F(UtilAllTest, compressEmptyStringTest) {
+ std::string original;
+ std::string compressed;
+ ASSERT_TRUE(UtilAll::compress(original, compressed));
+
+ std::string decompressed;
+ ASSERT_TRUE(UtilAll::uncompress(compressed, decompressed));
+ EXPECT_EQ(original, decompressed);
+}
+
+TEST_F(UtilAllTest, compressLargePayloadTest) {
+ // 64KB of repeated data — should compress well
+ std::string original(65536, 'A');
+ std::string compressed;
+ ASSERT_TRUE(UtilAll::compress(original, compressed));
+ EXPECT_LT(compressed.size(), original.size());
+
+ std::string decompressed;
+ ASSERT_TRUE(UtilAll::uncompress(compressed, decompressed));
+ EXPECT_EQ(original, decompressed);
+}
+
+TEST_F(UtilAllTest, uncompressInvalidDataFailsTest) {
+ std::string garbage = "this is not zlib compressed data";
+ std::string output;
+ EXPECT_FALSE(UtilAll::uncompress(garbage, output));
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/client/ClientManagerImpl.cpp
b/cpp/source/client/ClientManagerImpl.cpp
index 0441c8f2..3690711d 100644
--- a/cpp/source/client/ClientManagerImpl.cpp
+++ b/cpp/source/client/ClientManagerImpl.cpp
@@ -90,13 +90,12 @@ ClientManagerImpl::~ClientManagerImpl() {
}
void ClientManagerImpl::start() {
- if (State::CREATED != state_.load(std::memory_order_relaxed)) {
- SPDLOG_WARN("Unexpected client instance state: {}",
state_.load(std::memory_order_relaxed));
+ State expected = State::CREATED;
+ if (!state_.compare_exchange_strong(expected, State::STARTED)) {
+ SPDLOG_WARN("ClientManager start skipped: state={}", expected);
return;
}
- state_.store(State::STARTING, std::memory_order_relaxed);
-
callback_thread_pool_->start();
scheduler_->start();
@@ -110,17 +109,14 @@ void ClientManagerImpl::start() {
heartbeat_task_id_ = scheduler_->schedule(
heartbeat_functor, HEARTBEAT_TASK_NAME, std::chrono::seconds(1),
std::chrono::seconds(10));
SPDLOG_DEBUG("Heartbeat task-id={}", heartbeat_task_id_);
-
- state_.store(State::STARTED, std::memory_order_relaxed);
}
void ClientManagerImpl::shutdown() {
SPDLOG_INFO("Client manager shutdown");
- if (State::STARTED != state_.load(std::memory_order_relaxed)) {
- SPDLOG_WARN("Unexpected client instance state: {}",
state_.load(std::memory_order_relaxed));
+ State expected = State::STARTED;
+ if (!state_.compare_exchange_strong(expected, State::STOPPED)) {
return;
}
- state_.store(STOPPING, std::memory_order_relaxed);
callback_thread_pool_->shutdown();
@@ -136,7 +132,6 @@ void ClientManagerImpl::shutdown() {
SPDLOG_DEBUG("rpc_clients_ is clear");
}
- state_.store(State::STOPPED, std::memory_order_relaxed);
SPDLOG_DEBUG("ClientManager stopped");
}
@@ -158,7 +153,18 @@ std::vector<std::string>
ClientManagerImpl::cleanOfflineRpcClients() {
absl::MutexLock lk(&rpc_clients_mtx_);
for (auto it = rpc_clients_.begin(); it != rpc_clients_.end();) {
std::string host = it->first;
- if (it->second->needHeartbeat() && !hosts.contains(host)) {
+ auto& rpc_client = it->second;
+
+ if (!rpc_client->needHeartbeat()) {
+ // Non-heartbeat clients (e.g. name server) are only removed when
their channel is dead
+ if (!rpc_client->ok()) {
+ SPDLOG_INFO("Removed RPC client for dead non-heartbeat peer.
RemoteHost={}", host);
+ removed.push_back(host);
+ rpc_clients_.erase(it++);
+ } else {
+ it++;
+ }
+ } else if (!hosts.contains(host)) {
SPDLOG_INFO("Removed RPC client whose peer is offline. RemoteHost={}",
host);
removed.push_back(host);
rpc_clients_.erase(it++);
@@ -206,6 +212,7 @@ void ClientManagerImpl::heartbeat(const std::string&
target_host,
case rmq::Code::ILLEGAL_CONSUMER_GROUP: {
SPDLOG_ERROR("IllegalConsumerGroup: {}. Host={}", status.message(),
invocation_context->remote_address);
ec = ErrorCode::IllegalConsumerGroup;
+ cb(ec, invocation_context->response);
break;
}
@@ -260,9 +267,7 @@ void ClientManagerImpl::heartbeat(const std::string&
target_host,
}
void ClientManagerImpl::doHeartbeat() {
- if (State::STARTED != state_.load(std::memory_order_relaxed) &&
- State::STARTING != state_.load(std::memory_order_relaxed)) {
- SPDLOG_WARN("Unexpected client manager state={}.",
state_.load(std::memory_order_relaxed));
+ if (State::STARTED != state_.load(std::memory_order_relaxed)) {
return;
}
@@ -275,6 +280,9 @@ void ClientManagerImpl::doHeartbeat() {
}
}
}
+
+ // Periodically clean up stale RPC clients (including non-heartbeat name
server clients)
+ cleanOfflineRpcClients();
}
bool ClientManagerImpl::send(const std::string& target_host,
@@ -302,7 +310,10 @@ bool ClientManagerImpl::send(const std::string&
target_host,
}
if (State::STARTED != client_manager_ptr->state()) {
- // TODO: Would this leak some memory?
+ SendResult send_result = {};
+ send_result.target = target_host;
+ send_result.ec = ErrorCode::IllegalState;
+ cb(send_result);
return;
}
@@ -1579,8 +1590,7 @@ std::error_code
ClientManagerImpl::notifyClientTermination(const std::string& ta
}
void ClientManagerImpl::submit(std::function<void()> task) {
- State current_state = state();
- if (current_state == State::STOPPING || current_state == State::STOPPED) {
+ if (State::STARTED != state_.load(std::memory_order_relaxed)) {
return;
}
callback_thread_pool_->submit(task);
diff --git a/cpp/source/client/TelemetryBidiReactor.cpp
b/cpp/source/client/TelemetryBidiReactor.cpp
index a73efadd..0717330c 100644
--- a/cpp/source/client/TelemetryBidiReactor.cpp
+++ b/cpp/source/client/TelemetryBidiReactor.cpp
@@ -17,6 +17,8 @@
#include "TelemetryBidiReactor.h"
#include "FmtEnumFormatter.h"
+#include <algorithm>
+#include <chrono>
#include <memory>
#include <utility>
@@ -37,6 +39,14 @@
TelemetryBidiReactor::TelemetryBidiReactor(std::weak_ptr<Client> client,
peer_address_(std::move(peer_address)),
state_(StreamState::Ready) {
auto ptr = client_.lock();
+ if (!ptr) {
+ SPDLOG_WARN("Client already destructed when creating telemetry stream to
{}", peer_address_);
+ {
+ absl::MutexLock lk(&state_mtx_);
+ state_ = StreamState::Closed;
+ }
+ return;
+ }
auto deadline = std::chrono::system_clock::now() + std::chrono::hours(1);
context_.set_deadline(deadline);
Metadata metadata;
@@ -63,10 +73,26 @@ bool TelemetryBidiReactor::awaitApplyingSettings() {
return true;
}
}
+
+ // Settings exchange failed — initiate proper stream closure.
+ // Transition to Closing (not directly to Closed) so that gRPC's OnDone
+ // callback fires normally and handles cleanup. Suppress reconnection
+ // since this is an intentional shutdown of a failed session.
+ intentional_close_.store(true, std::memory_order_release);
{
absl::MutexLock lk(&state_mtx_);
- state_ = StreamState::Closed;
- state_cv_.SignalAll();
+ if (state_ == StreamState::Ready) {
+ state_ = StreamState::Closing;
+ }
+ }
+ context_.TryCancel();
+
+ // Wait for OnDone to set state to Closed
+ {
+ absl::MutexLock lk(&state_mtx_);
+ while (state_ != StreamState::Closed) {
+ state_cv_.WaitWithTimeout(&state_mtx_, absl::Seconds(1));
+ }
}
return false;
}
@@ -78,7 +104,12 @@ void TelemetryBidiReactor::OnWriteDone(bool ok) {
RemoveHold();
if (!ok) {
- SPDLOG_WARN("Failed to write telemetry command {} to {}",
writes_.front().ShortDebugString(), peer_address_);
+ {
+ absl::MutexLock lk(&writes_mtx_);
+ if (!writes_.empty()) {
+ SPDLOG_WARN("Failed to write telemetry command {} to {}",
writes_.front().ShortDebugString(), peer_address_);
+ }
+ }
signalClose();
return;
}
@@ -107,6 +138,7 @@ void TelemetryBidiReactor::OnReadDone(bool ok) {
{
absl::MutexLock lk(&state_mtx_);
if (StreamState::Ready != state_) {
+ RemoveHold();
return;
}
}
@@ -115,6 +147,7 @@ void TelemetryBidiReactor::OnReadDone(bool ok) {
auto client = client_.lock();
if (!client) {
SPDLOG_INFO("Client for {} has destructed", peer_address_);
+ RemoveHold();
signalClose();
return;
}
@@ -124,7 +157,10 @@ void TelemetryBidiReactor::OnReadDone(bool ok) {
auto settings = read_.settings();
SPDLOG_INFO("Receive settings from {}: {}", peer_address_,
settings.ShortDebugString());
applySettings(settings);
- sync_settings_promise_.set_value(true);
+ bool expected = false;
+ if (settings_received_.compare_exchange_strong(expected, true,
std::memory_order_acq_rel)) {
+ sync_settings_promise_.set_value(true);
+ }
break;
}
@@ -175,6 +211,8 @@ void TelemetryBidiReactor::OnReadDone(bool ok) {
if (StreamState::Ready == state_) {
SPDLOG_DEBUG("Spawn new read op, state={}",
static_cast<std::uint8_t>(state_));
StartRead(&read_);
+ } else {
+ RemoveHold();
}
}
}
@@ -281,7 +319,7 @@ void TelemetryBidiReactor::write(TelemetryCommand command) {
{
absl::MutexLock lk(&writes_mtx_);
- writes_.push_back(command);
+ writes_.push_back(std::move(command));
}
tryWriteNext();
}
@@ -317,7 +355,10 @@ void TelemetryBidiReactor::signalClose() {
}
void TelemetryBidiReactor::close() {
- SPDLOG_DEBUG("{}#fireClose", peer_address_);
+ SPDLOG_DEBUG("{}#close", peer_address_);
+
+ // Mark as intentional close to suppress reconnection in OnDone
+ intentional_close_.store(true, std::memory_order_release);
{
absl::MutexLock lk(&state_mtx_);
@@ -326,20 +367,28 @@ void TelemetryBidiReactor::close() {
}
}
- {
- absl::MutexLock lk(&writes_mtx_);
- writes_.clear();
- }
+ // Do NOT clear writes_ here — gRPC may hold raw pointers to elements in
+ // writes_ from a prior StartWrite(&writes_.front()). Clearing now would
cause
+ // use-after-free when gRPC finishes the in-flight write.
+
context_.TryCancel();
- // Acquire state lock
- while (StreamState::Closed != state_) {
+ // Wait for OnDone: all gRPC callbacks complete, all holds removed.
+ {
absl::MutexLock lk(&state_mtx_);
- if (state_cv_.WaitWithTimeout(&state_mtx_, absl::Seconds(1))) {
- SPDLOG_WARN("StreamState CondVar timed out before getting signalled:
state={}",
- static_cast<uint8_t>(state_));
+ while (StreamState::Closed != state_) {
+ if (state_cv_.WaitWithTimeout(&state_mtx_, absl::Seconds(1))) {
+ SPDLOG_WARN("StreamState CondVar timed out before getting signalled:
state={}",
+ static_cast<uint8_t>(state_));
+ }
}
}
+
+ // Safe to clear now — gRPC has released all references to writes_ elements.
+ {
+ absl::MutexLock lk(&writes_mtx_);
+ writes_.clear();
+ }
}
/// Notifies the application that all operations associated with this RPC
@@ -366,8 +415,18 @@ void TelemetryBidiReactor::OnDone(const grpc::Status&
status) {
return;
}
- if (client->active()) {
- client->createSession(peer_address_, true);
+ // Only reconnect if this was an unexpected disconnection, not an
intentional close
+ if (client->active() && !intentional_close_.load(std::memory_order_acquire))
{
+ constexpr auto kReconnectDelay = std::chrono::seconds(1);
+ auto address = peer_address_;
+ auto weak_client = client_; // client_ is already a weak_ptr member
+ client->schedule("session-reconnect", [weak_client, address]() {
+ auto c = weak_client.lock();
+ if (c && c->active()) {
+ c->createSession(address, true);
+ }
+ }, kReconnectDelay);
+ SPDLOG_INFO("Scheduled session reconnect to {} after {}ms", peer_address_,
kReconnectDelay.count() * 1000);
}
}
diff --git a/cpp/source/client/include/TelemetryBidiReactor.h
b/cpp/source/client/include/TelemetryBidiReactor.h
index 3217fd4b..11018128 100644
--- a/cpp/source/client/include/TelemetryBidiReactor.h
+++ b/cpp/source/client/include/TelemetryBidiReactor.h
@@ -25,6 +25,7 @@
#include <utility>
#include <vector>
+#include "absl/container/flat_hash_map.h"
#include "Client.h"
#include "RpcClient.h"
#include "absl/base/thread_annotations.h"
@@ -131,6 +132,10 @@ private:
absl::CondVar state_cv_;
std::promise<bool> sync_settings_promise_;
+ std::atomic<bool> settings_received_{false};
+
+ /// Set to true when close() is called intentionally. Suppresses
reconnection in OnDone().
+ std::atomic<bool> intentional_close_{false};
void applySettings(const rmq::Settings& settings);
diff --git a/cpp/source/client/tests/BUILD.bazel
b/cpp/source/client/tests/BUILD.bazel
index e88b4f59..432b8b62 100644
--- a/cpp/source/client/tests/BUILD.bazel
+++ b/cpp/source/client/tests/BUILD.bazel
@@ -72,4 +72,16 @@ cc_test(
"//source/client:client_library",
"@com_google_googletest//:gtest_main",
],
+)
+
+cc_test(
+ name = "telemetry_bidi_reactor_test",
+ srcs = [
+ "TelemetryBidiReactorTest.cpp",
+ ],
+ deps = [
+ "//source/client:client_library",
+ "//source/client/mocks:client_mocks",
+ "@com_google_googletest//:gtest_main",
+ ],
)
\ No newline at end of file
diff --git a/cpp/source/client/tests/CMakeLists.txt
b/cpp/source/client/tests/CMakeLists.txt
index 4197c5b8..e316640b 100644
--- a/cpp/source/client/tests/CMakeLists.txt
+++ b/cpp/source/client/tests/CMakeLists.txt
@@ -17,3 +17,4 @@ add_client_test(client_manager_test ClientManagerTest.cpp)
add_client_test(topic_assignment_info_test TopicAssignmentInfoTest.cpp)
# TracingUtilityTest requires OpenTelemetry headers (only available via Bazel)
# add_client_test(tracing_utility_test TracingUtilityTest.cpp)
+add_client_test(telemetry_bidi_reactor_test TelemetryBidiReactorTest.cpp)
diff --git a/cpp/source/client/tests/ClientManagerTest.cpp
b/cpp/source/client/tests/ClientManagerTest.cpp
index 1eb64e74..7fab9e8d 100644
--- a/cpp/source/client/tests/ClientManagerTest.cpp
+++ b/cpp/source/client/tests/ClientManagerTest.cpp
@@ -21,7 +21,9 @@
#include "ClientManagerImpl.h"
#include "RpcClientMock.h"
+#include "SendResult.h"
#include "gtest/gtest.h"
+#include "rocketmq/ErrorCode.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -223,4 +225,125 @@ TEST_F(ClientManagerTest, testEndTransaction) {
EXPECT_TRUE(callback_invoked);
}
+TEST_F(ClientManagerTest, sendSuccessTest) {
+ bool completed = false;
+ absl::Mutex mtx;
+ absl::CondVar cv;
+ SendResult captured_result;
+
+ auto mock_send = [&](const SendMessageRequest& request,
InvocationContext<SendMessageResponse>* invocation_context) {
+ auto* entry = invocation_context->response.add_entries();
+ entry->set_message_id("msg-id-001");
+ entry->set_transaction_id("txn-id-001");
+ entry->set_recall_handle("recall-handle-001");
+ invocation_context->response.mutable_status()->set_code(rmq::Code::OK);
+ invocation_context->onCompletion(true);
+ };
+
+ EXPECT_CALL(*rpc_client_,
asyncSend).Times(1).WillOnce(testing::Invoke(mock_send));
+
+ SendMessageRequest request;
+ auto* msg = request.add_messages();
+ msg->mutable_topic()->set_name(topic_);
+ msg->set_body(message_body_);
+
+ auto callback = [&](const SendResult& result) {
+ absl::MutexLock lk(&mtx);
+ completed = true;
+ captured_result = result;
+ cv.SignalAll();
+ };
+
+ client_manager_->send(target_host_, metadata_, request, callback);
+
+ {
+ absl::MutexLock lk(&mtx);
+ if (!completed) {
+ cv.WaitWithDeadline(&mtx, absl::Now() + absl::Seconds(3));
+ }
+ }
+
+ EXPECT_TRUE(completed);
+ EXPECT_FALSE(captured_result.ec);
+ EXPECT_EQ("msg-id-001", captured_result.message_id);
+ EXPECT_EQ("txn-id-001", captured_result.transaction_id);
+ EXPECT_EQ("recall-handle-001", captured_result.recall_handle);
+ EXPECT_EQ(target_host_, captured_result.target);
+}
+
+TEST_F(ClientManagerTest, sendReturnsErrorOnBadRequestTest) {
+ bool completed = false;
+ absl::Mutex mtx;
+ absl::CondVar cv;
+ SendResult captured_result;
+
+ auto mock_send = [&](const SendMessageRequest& request,
InvocationContext<SendMessageResponse>* invocation_context) {
+
invocation_context->response.mutable_status()->set_code(rmq::Code::ILLEGAL_TOPIC);
+ invocation_context->response.mutable_status()->set_message("Illegal
topic");
+ invocation_context->onCompletion(true);
+ };
+
+ EXPECT_CALL(*rpc_client_,
asyncSend).Times(1).WillOnce(testing::Invoke(mock_send));
+
+ SendMessageRequest request;
+ auto* msg = request.add_messages();
+ msg->mutable_topic()->set_name(topic_);
+ msg->set_body(message_body_);
+
+ auto callback = [&](const SendResult& result) {
+ absl::MutexLock lk(&mtx);
+ completed = true;
+ captured_result = result;
+ cv.SignalAll();
+ };
+
+ client_manager_->send(target_host_, metadata_, request, callback);
+
+ {
+ absl::MutexLock lk(&mtx);
+ if (!completed) {
+ cv.WaitWithDeadline(&mtx, absl::Now() + absl::Seconds(3));
+ }
+ }
+
+ EXPECT_TRUE(completed);
+ EXPECT_TRUE(static_cast<bool>(captured_result.ec));
+ EXPECT_EQ(ErrorCode::IllegalTopic, captured_result.ec);
+ EXPECT_TRUE(captured_result.message_id.empty());
+}
+
+TEST_F(ClientManagerTest, cleanOfflineRpcClientsRemovesDeadChannelsTest) {
+ std::string live_host = "ipv4:10.0.0.1:10911";
+ std::string dead_host = "ipv4:10.0.0.2:10911";
+
+ auto live_rpc_client = std::make_shared<testing::NiceMock<RpcClientMock>>();
+ ON_CALL(*live_rpc_client, ok).WillByDefault(testing::Return(true));
+ ON_CALL(*live_rpc_client,
needHeartbeat()).WillByDefault(testing::Return(false));
+
+ auto dead_rpc_client = std::make_shared<testing::NiceMock<RpcClientMock>>();
+ ON_CALL(*dead_rpc_client, ok).WillByDefault(testing::Return(false));
+ ON_CALL(*dead_rpc_client,
needHeartbeat()).WillByDefault(testing::Return(false));
+
+ client_manager_->addRpcClient(live_host, live_rpc_client);
+ client_manager_->addRpcClient(dead_host, dead_rpc_client);
+
+ std::vector<std::string> removed = client_manager_->cleanOfflineRpcClients();
+
+ // Dead channel should be removed; live channel should be retained.
+ EXPECT_EQ(1u, removed.size());
+ EXPECT_EQ(dead_host, removed[0]);
+
+ // Live host's mock should still be returned from the map.
+ auto live_client = client_manager_->getRpcClient(live_host);
+ EXPECT_EQ(live_rpc_client, live_client);
+
+ // Dead host's mock was removed; getRpcClient creates a new real client
instead.
+ auto dead_client = client_manager_->getRpcClient(dead_host);
+ EXPECT_NE(dead_rpc_client, dead_client);
+}
+
+TEST_F(ClientManagerTest, stateReturnsStartedTest) {
+ EXPECT_EQ(State::STARTED, client_manager_->state());
+}
+
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/source/client/tests/TelemetryBidiReactorTest.cpp
b/cpp/source/client/tests/TelemetryBidiReactorTest.cpp
new file mode 100644
index 00000000..3cbb5af1
--- /dev/null
+++ b/cpp/source/client/tests/TelemetryBidiReactorTest.cpp
@@ -0,0 +1,369 @@
+/*
+ * 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.
+ */
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <thread>
+
+#include "ClientMock.h"
+#include "TelemetryBidiReactor.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include "grpcpp/create_channel.h"
+#include "grpcpp/security/credentials.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+/// Helper: create a real gRPC stub to localhost (no server needed for
construction).
+struct TestStubHolder {
+ std::shared_ptr<grpc::Channel> channel;
+ std::unique_ptr<rmq::MessagingService::Stub> stub;
+};
+
+TestStubHolder createTestStub() {
+ auto channel = grpc::CreateChannel("127.0.0.1:19999",
grpc::InsecureChannelCredentials());
+ auto stub = rmq::MessagingService::NewStub(channel);
+ TestStubHolder holder;
+ holder.channel = channel;
+ holder.stub = std::move(stub);
+ return holder;
+}
+
+/// Helper: create an expired weak_ptr<Client> that causes the reactor to
short-circuit.
+std::weak_ptr<Client> expiredWeakClient() {
+ std::shared_ptr<Client> empty;
+ return std::weak_ptr<Client>(empty);
+}
+
+} // namespace
+
+// ---------------------------------------------------------------------------
+// StreamState enum tests
+// ---------------------------------------------------------------------------
+
+TEST(StreamStateTest, enumValuesAndOrderingTest) {
+ EXPECT_EQ(0, static_cast<std::uint8_t>(StreamState::Ready));
+ EXPECT_EQ(1, static_cast<std::uint8_t>(StreamState::Closing));
+ EXPECT_EQ(2, static_cast<std::uint8_t>(StreamState::Closed));
+
+ // Verify ordering: Ready < Closing < Closed
+ EXPECT_TRUE(StreamState::Ready < StreamState::Closing);
+ EXPECT_TRUE(StreamState::Closing < StreamState::Closed);
+ EXPECT_TRUE(StreamState::Ready < StreamState::Closed);
+}
+
+TEST(StreamStateTest, enumDistinctValuesTest) {
+ EXPECT_NE(StreamState::Ready, StreamState::Closing);
+ EXPECT_NE(StreamState::Closing, StreamState::Closed);
+ EXPECT_NE(StreamState::Ready, StreamState::Closed);
+}
+
+// ---------------------------------------------------------------------------
+// Construction tests (expired weak_ptr — no gRPC calls)
+// ---------------------------------------------------------------------------
+
+class TelemetryBidiReactorTest : public testing::Test {
+protected:
+ void SetUp() override {
+ TestStubHolder h = createTestStub();
+ channel_ = h.channel;
+ stub_ = std::move(h.stub);
+ }
+
+ std::shared_ptr<grpc::Channel> channel_;
+ std::unique_ptr<rmq::MessagingService::Stub> stub_;
+};
+
+TEST_F(TelemetryBidiReactorTest,
constructWithExpiredClientSetsClosedStateTest) {
+ // With an expired weak_ptr, the constructor short-circuits:
+ // - Logs a warning
+ // - Sets state_ = StreamState::Closed
+ // - Returns without calling stub->async()->Telemetry()
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // The reactor should be in Closed state. We verify indirectly: close() on a
+ // Closed reactor should return immediately (no blocking wait).
+ auto start = std::chrono::steady_clock::now();
+ reactor->close();
+ auto elapsed = std::chrono::steady_clock::now() - start;
+
+ // close() should complete nearly instantly since state is already Closed.
+
EXPECT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count(),
500);
+}
+
+TEST_F(TelemetryBidiReactorTest, constructWithExpiredClientDestructorSafeTest)
{
+ // Verify that the reactor can be constructed and destructed safely with an
+ // expired client — no crashes, no leaks.
+ {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+ // Let it go out of scope — destructor should log and exit cleanly.
+ }
+ // If we reach here without crashing, the test passes.
+ SUCCEED();
+}
+
+// ---------------------------------------------------------------------------
+// close() tests
+// ---------------------------------------------------------------------------
+
+TEST_F(TelemetryBidiReactorTest, closeOnExpiredClientReactorIsIdempotentTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // First close — state is already Closed, should return immediately.
+ reactor->close();
+
+ // Second close — still safe, no crash, no hang.
+ auto start = std::chrono::steady_clock::now();
+ reactor->close();
+ auto elapsed = std::chrono::steady_clock::now() - start;
+
EXPECT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count(),
500);
+}
+
+// ---------------------------------------------------------------------------
+// write() tests
+// ---------------------------------------------------------------------------
+
+TEST_F(TelemetryBidiReactorTest, writeOnClosedReactorDropsCommandTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // State is Closed (expired client). write() should silently drop the command
+ // because state != Ready.
+ TelemetryCommand cmd;
+ cmd.mutable_thread_stack_trace()->set_nonce("test-nonce");
+ cmd.mutable_thread_stack_trace()->set_thread_stack_trace("dummy");
+
+ // Should not crash or hang — the command is rejected internally.
+ reactor->write(std::move(cmd));
+ SUCCEED();
+}
+
+TEST_F(TelemetryBidiReactorTest, writeMultipleCommandsOnClosedReactorTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // Write several commands — all should be silently dropped.
+ for (int i = 0; i < 10; ++i) {
+ TelemetryCommand cmd;
+ cmd.mutable_thread_stack_trace()->set_nonce("nonce-" + std::to_string(i));
+ reactor->write(std::move(cmd));
+ }
+ SUCCEED();
+}
+
+// ---------------------------------------------------------------------------
+// awaitApplyingSettings() tests
+// ---------------------------------------------------------------------------
+
+TEST_F(TelemetryBidiReactorTest,
awaitApplyingSettingsTimesOutOnClosedReactorTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // State is Closed. The promise is never set, so awaitApplyingSettings()
will:
+ // 1. Wait 3 seconds for the future (times out)
+ // 2. Set intentional_close_ = true
+ // 3. Try Ready→Closing transition (no-op, already Closed)
+ // 4. Call context_.TryCancel() (harmless)
+ // 5. Wait for state == Closed (already true, exits immediately)
+ // 6. Return false
+ auto start = std::chrono::steady_clock::now();
+ bool result = reactor->awaitApplyingSettings();
+ auto elapsed = std::chrono::steady_clock::now() - start;
+
+ EXPECT_FALSE(result);
+ // Should take approximately 3 seconds (the wait_for timeout).
+ auto ms =
std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
+ EXPECT_GE(ms, 2500);
+ EXPECT_LT(ms, 5000);
+}
+
+// ---------------------------------------------------------------------------
+// OnDone() tests (manual invocation)
+// ---------------------------------------------------------------------------
+
+TEST_F(TelemetryBidiReactorTest, onDoneWithExpiredClientNoCrashTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // Manually invoke OnDone with a cancelled status.
+ // Since client_ is expired, OnDone should:
+ // 1. Set state to Closed (already Closed, no-op)
+ // 2. Signal condvar (harmless)
+ // 3. client_.lock() returns nullptr → return early (no reconnect)
+ grpc::Status status(grpc::StatusCode::CANCELLED, "test cancellation");
+ reactor->OnDone(status);
+ SUCCEED();
+}
+
+TEST_F(TelemetryBidiReactorTest, onDoneWithOkStatusAndExpiredClientTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(expiredWeakClient(),
stub_.get(), "127.0.0.1:19999");
+
+ // OnDone with OK status — still safe with expired client.
+ grpc::Status status(grpc::StatusCode::OK, "");
+ reactor->OnDone(status);
+ SUCCEED();
+}
+
+// ---------------------------------------------------------------------------
+// Construction with valid ClientMock (gRPC stream starts, fails async)
+// ---------------------------------------------------------------------------
+
+class TelemetryBidiReactorWithClientTest : public testing::Test {
+protected:
+ void SetUp() override {
+ TestStubHolder h = createTestStub();
+ channel_ = h.channel;
+ stub_ = std::move(h.stub);
+
+ client_ = std::make_shared<testing::NiceMock<ClientMock>>();
+ ON_CALL(*client_, config).WillByDefault(testing::ReturnRef(config_));
+ ON_CALL(*client_, active).WillByDefault(testing::Return(false));
+ }
+
+ void TearDown() override {
+ // Allow async gRPC callbacks to complete before destroying mocks.
+ std::this_thread::sleep_for(std::chrono::milliseconds(500));
+ client_.reset();
+ }
+
+ std::shared_ptr<grpc::Channel> channel_;
+ std::unique_ptr<rmq::MessagingService::Stub> stub_;
+ std::shared_ptr<testing::NiceMock<ClientMock>> client_;
+ ClientConfig config_;
+};
+
+TEST_F(TelemetryBidiReactorWithClientTest,
constructWithValidClientStartsStreamTest) {
+ // With a valid client, the constructor:
+ // 1. Locks the weak_ptr successfully
+ // 2. Sets a deadline on the context
+ // 3. Signs metadata from config
+ // 4. Calls stub->async()->Telemetry() — starts the gRPC stream
+ // 5. Calls StartRead(), AddHold(), StartCall()
+ //
+ // The stream will fail (no server), triggering OnDone asynchronously.
+ // Since active() returns false, no reconnect is scheduled.
+ auto reactor = std::make_shared<TelemetryBidiReactor>(
+ std::weak_ptr<Client>(client_), stub_.get(), "127.0.0.1:19999");
+
+ // close() should work: sets intentional_close_, cancels context, waits for
Closed.
+ // If OnDone already fired (stream failed quickly), close() returns
immediately.
+ // If OnDone hasn't fired yet, TryCancel triggers it.
+ reactor->close();
+ SUCCEED();
+}
+
+TEST_F(TelemetryBidiReactorWithClientTest, closeBeforeStreamConnectsTest) {
+ // Immediately close after construction — the stream may not have connected
yet.
+ auto reactor = std::make_shared<TelemetryBidiReactor>(
+ std::weak_ptr<Client>(client_), stub_.get(), "127.0.0.1:19999");
+
+ // close() should complete within a reasonable time.
+ auto start = std::chrono::steady_clock::now();
+ reactor->close();
+ auto elapsed = std::chrono::steady_clock::now() - start;
+
+ // Should complete within 5 seconds (gRPC TryCancel is usually fast).
+
EXPECT_LT(std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count(),
5000);
+}
+
+TEST_F(TelemetryBidiReactorWithClientTest,
DISABLED_writeImmediatelyAfterConstructTest) {
+ // Write a command immediately after construction. The stream is in Ready
state
+ // (gRPC hasn't reported failure yet), so the command should be buffered and
+ // StartWrite attempted. It will fail when the stream fails.
+ auto reactor = std::make_shared<TelemetryBidiReactor>(
+ std::weak_ptr<Client>(client_), stub_.get(), "127.0.0.1:19999");
+
+ TelemetryCommand cmd;
+ cmd.mutable_thread_stack_trace()->set_nonce("early-write");
+ reactor->write(std::move(cmd));
+
+ // Clean up — close the stream.
+ reactor->close();
+ SUCCEED();
+}
+
+// ---------------------------------------------------------------------------
+// OnDone with valid client — reconnect suppression
+// ---------------------------------------------------------------------------
+
+TEST_F(TelemetryBidiReactorWithClientTest,
onDoneWithIntentionalCloseSkipsReconnectTest) {
+ auto reactor = std::make_shared<TelemetryBidiReactor>(
+ std::weak_ptr<Client>(client_), stub_.get(), "127.0.0.1:19999");
+
+ // close() sets intentional_close_ = true, then cancels the stream.
+ reactor->close();
+
+ // After close(), OnDone has already been called by gRPC (via TryCancel).
+ // Since intentional_close_ was set, no reconnect should have been scheduled.
+ // We verify by checking that createSession was never called.
+ // (NiceMock suppresses uninteresting calls, so no EXPECT_CALL needed —
+ // but we can verify the mock was not called if we set an expectation.)
+ SUCCEED();
+}
+
+// ---------------------------------------------------------------------------
+// DISABLED tests — require a live gRPC server
+// ---------------------------------------------------------------------------
+
+// DISABLED: Would test full settings exchange round-trip.
+// Requires a gRPC server that responds to Telemetry with a Settings command.
+// The test would:
+// 1. Start a mock gRPC server
+// 2. Create a reactor with a valid client
+// 3. Server sends Settings → OnReadDone processes them
+// 4. awaitApplyingSettings() returns true
+// 5. Verify that client config was updated with the received settings
+TEST_F(TelemetryBidiReactorTest, DISABLED_settingsExchangeRoundTripTest) {
+ // Requires a live gRPC server that sends Settings on the telemetry stream.
+}
+
+// DISABLED: Would test bidirectional write and read.
+// Requires a gRPC server that:
+// - Accepts Telemetry writes
+// - Sends Telemetry responses (e.g., PrintThreadStackTraceCommand)
+// The test would verify:
+// - write() buffers and sends commands
+// - OnReadDone processes incoming commands
+// - OnWriteDone removes written commands from the buffer
+// - tryWriteNext() sends the next buffered command
+TEST_F(TelemetryBidiReactorTest, DISABLED_bidirectionalWriteReadTest) {
+ // Requires a live gRPC server supporting bidirectional telemetry.
+}
+
+// DISABLED: Would test automatic session reconnection.
+// Requires a gRPC server that:
+// - Accepts initial connection
+// - Drops the connection after a short delay
+// The test would verify:
+// - OnDone fires with non-OK status
+// - Since client is active and intentional_close_ is false, reconnect is
scheduled
+// - client->schedule("session-reconnect", ...) is called
+TEST_F(TelemetryBidiReactorWithClientTest,
DISABLED_automaticReconnectOnUnexpectedDisconnectTest) {
+ // Requires a gRPC server that drops connections.
+}
+
+// DISABLED: Would test OnReadDone handling of various TelemetryCommand types.
+// Requires a gRPC server that sends:
+// - kSettings → applySettings()
+// - kRecoverOrphanedTransactionCommand →
client->recoverOrphanedTransaction()
+// - kPrintThreadStackTraceCommand → writes response back
+// - kVerifyMessageCommand → client->verify()
+TEST_F(TelemetryBidiReactorTest,
DISABLED_onReadDoneHandlesAllCommandTypesTest) {
+ // Requires a live gRPC server that sends specific TelemetryCommand types.
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/BUILD.bazel
b/cpp/source/rocketmq/tests/BUILD.bazel
index 1c9e2aa2..1bf83b3d 100644
--- a/cpp/source/rocketmq/tests/BUILD.bazel
+++ b/cpp/source/rocketmq/tests/BUILD.bazel
@@ -42,7 +42,9 @@ cc_test(
srcs = [
"ClientImplTest.cpp",
],
- deps = base_deps,
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
)
cc_test(
@@ -75,4 +77,71 @@ cc_test(
"PriorityMessageTest.cpp",
],
deps = base_deps
-)
\ No newline at end of file
+)
+
+cc_test(
+ name = "topic_publish_info_test",
+ srcs = [
+ "TopicPublishInfoTest.cpp",
+ ],
+ deps = base_deps,
+)
+
+cc_test(
+ name = "fifo_producer_partition_test",
+ srcs = [
+ "FifoProducerPartitionTest.cpp",
+ ],
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
+)
+cc_test(
+ name = "producer_impl_test",
+ srcs = [
+ "ProducerImplTest.cpp",
+ ],
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
+)
+
+cc_test(
+ name = "process_queue_impl_test",
+ srcs = [
+ "ProcessQueueImplTest.cpp",
+ ],
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
+)
+
+cc_test(
+ name = "consume_task_test",
+ srcs = [
+ "ConsumeTaskTest.cpp",
+ ],
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
+)
+
+cc_test(
+ name = "simple_consumer_impl_test",
+ srcs = [
+ "SimpleConsumerImplTest.cpp",
+ ],
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
+)
+
+cc_test(
+ name = "push_consumer_impl_test",
+ srcs = [
+ "PushConsumerImplTest.cpp",
+ ],
+ deps = base_deps + [
+ "//source/rocketmq/mocks:rocketmq_mocks",
+ ],
+)
diff --git a/cpp/source/rocketmq/tests/CMakeLists.txt
b/cpp/source/rocketmq/tests/CMakeLists.txt
index f8c80eea..673d3eb9 100644
--- a/cpp/source/rocketmq/tests/CMakeLists.txt
+++ b/cpp/source/rocketmq/tests/CMakeLists.txt
@@ -17,3 +17,10 @@ add_rocketmq_test(priority_message_test
PriorityMessageTest.cpp)
add_rocketmq_test(send_context_test SendContextTest.cpp)
add_rocketmq_test(static_name_server_resolver_test
StaticNameServerResolverTest.cpp)
add_rocketmq_test(time_test TimeTest.cpp)
+add_rocketmq_test(topic_publish_info_test TopicPublishInfoTest.cpp)
+add_rocketmq_test(fifo_producer_partition_test FifoProducerPartitionTest.cpp)
+add_rocketmq_test(producer_impl_test ProducerImplTest.cpp)
+add_rocketmq_test(process_queue_impl_test ProcessQueueImplTest.cpp)
+add_rocketmq_test(consume_task_test ConsumeTaskTest.cpp)
+add_rocketmq_test(simple_consumer_impl_test SimpleConsumerImplTest.cpp)
+add_rocketmq_test(push_consumer_impl_test PushConsumerImplTest.cpp)
diff --git a/cpp/source/rocketmq/tests/ClientImplTest.cpp
b/cpp/source/rocketmq/tests/ClientImplTest.cpp
index 3975e6ba..e8a08a85 100644
--- a/cpp/source/rocketmq/tests/ClientImplTest.cpp
+++ b/cpp/source/rocketmq/tests/ClientImplTest.cpp
@@ -14,11 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+#include <chrono>
#include <cstddef>
+#include <functional>
+#include <memory>
+#include <stdexcept>
+#include <string>
#include <unordered_set>
#include "ClientImpl.h"
+#include "ClientManagerMock.h"
+#include "NameServerResolverMock.h"
+#include "ProducerImpl.h"
+#include "gmock/gmock.h"
#include "gtest/gtest.h"
+#include "rocketmq/ErrorCode.h"
#include "rocketmq/RocketMQ.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -33,4 +43,121 @@ TEST(ClientImplTest, testClientId) {
}
}
+// Test-only subclass that exposes the protected accessPoint() method.
+class TestableProducerImpl : public ProducerImpl {
+public:
+ TestableProducerImpl() : ClientImpl("") {}
+
+ rmq::Endpoints testAccessPoint() {
+ return accessPoint();
+ }
+};
+
+class ClientImplLifecycleTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ resolver_mock_ =
std::make_shared<testing::NiceMock<NameServerResolverMock>>();
+ client_manager_ = std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ }
+
+ void TearDown() override {
+ producer_.reset();
+ testable_producer_.reset();
+ client_manager_.reset();
+ resolver_mock_.reset();
+ }
+
+ std::shared_ptr<ProducerImpl> producer_;
+ std::shared_ptr<TestableProducerImpl> testable_producer_;
+ std::shared_ptr<testing::NiceMock<NameServerResolverMock>> resolver_mock_;
+ std::shared_ptr<testing::NiceMock<ClientManagerMock>> client_manager_;
+};
+
+TEST_F(ClientImplLifecycleTest, startWithoutResolverThrowsTest) {
+ producer_ = std::make_shared<ProducerImpl>();
+ // No resolver configured — start() must throw.
+ EXPECT_THROW(producer_->start(), std::runtime_error);
+}
+
+TEST_F(ClientImplLifecycleTest, shutdownWithoutStartIsNoopTest) {
+ producer_ = std::make_shared<ProducerImpl>();
+ // State defaults to CREATED. shutdown() on a non-started client must be a
safe no-op.
+ EXPECT_NO_THROW(producer_->shutdown());
+ EXPECT_FALSE(producer_->active());
+}
+
+TEST_F(ClientImplLifecycleTest, doubleShutdownIsSafeTest) {
+ producer_ = std::make_shared<ProducerImpl>();
+ producer_->state(State::STARTED);
+
+ // First shutdown transitions from STARTED → STOPPED.
+ EXPECT_NO_THROW(producer_->shutdown());
+ EXPECT_FALSE(producer_->active());
+
+ // Second shutdown must be a safe no-op (CAS fails, returns immediately).
+ EXPECT_NO_THROW(producer_->shutdown());
+ EXPECT_FALSE(producer_->active());
+}
+
+TEST_F(ClientImplLifecycleTest, accessPointParsesIPv4Test) {
+ testable_producer_ = std::make_shared<TestableProducerImpl>();
+ ON_CALL(*resolver_mock_,
resolve()).WillByDefault(testing::Return(std::string("ipv4:10.0.0.1:8080")));
+ testable_producer_->withNameServerResolver(resolver_mock_);
+
+ rmq::Endpoints ep = testable_producer_->testAccessPoint();
+ EXPECT_EQ(ep.scheme(), rmq::AddressScheme::IPv4);
+ ASSERT_EQ(ep.addresses_size(), 1);
+ EXPECT_EQ(ep.addresses(0).host(), "10.0.0.1");
+ EXPECT_EQ(ep.addresses(0).port(), 8080);
+}
+
+TEST_F(ClientImplLifecycleTest, accessPointParsesDNSTest) {
+ testable_producer_ = std::make_shared<TestableProducerImpl>();
+ ON_CALL(*resolver_mock_,
resolve()).WillByDefault(testing::Return(std::string("dns:broker.example.com:9876")));
+ testable_producer_->withNameServerResolver(resolver_mock_);
+
+ rmq::Endpoints ep = testable_producer_->testAccessPoint();
+ EXPECT_EQ(ep.scheme(), rmq::AddressScheme::DOMAIN_NAME);
+ ASSERT_EQ(ep.addresses_size(), 1);
+ EXPECT_EQ(ep.addresses(0).host(), "broker.example.com");
+ EXPECT_EQ(ep.addresses(0).port(), 9876);
+}
+
+TEST_F(ClientImplLifecycleTest, accessPointParsesMultipleIPv4Test) {
+ testable_producer_ = std::make_shared<TestableProducerImpl>();
+ ON_CALL(*resolver_mock_, resolve())
+
.WillByDefault(testing::Return(std::string("ipv4:10.0.0.1:8080,10.0.0.2:9090")));
+ testable_producer_->withNameServerResolver(resolver_mock_);
+
+ rmq::Endpoints ep = testable_producer_->testAccessPoint();
+ EXPECT_EQ(ep.scheme(), rmq::AddressScheme::IPv4);
+ ASSERT_EQ(ep.addresses_size(), 2);
+ EXPECT_EQ(ep.addresses(0).host(), "10.0.0.1");
+ EXPECT_EQ(ep.addresses(0).port(), 8080);
+ EXPECT_EQ(ep.addresses(1).host(), "10.0.0.2");
+ EXPECT_EQ(ep.addresses(1).port(), 9090);
+}
+
+TEST_F(ClientImplLifecycleTest, scheduleWithNullManagerDoesNotCrashTest) {
+ producer_ = std::make_shared<ProducerImpl>();
+ // No client_manager set — schedule() must log a warning and return safely.
+ EXPECT_NO_THROW(
+ producer_->schedule("test-task", []() {},
std::chrono::milliseconds(100)));
+}
+
+TEST_F(ClientImplLifecycleTest, activeReflectsStateTest) {
+ producer_ = std::make_shared<ProducerImpl>();
+
+ // Default state is CREATED → not active.
+ EXPECT_FALSE(producer_->active());
+
+ // Transition to STARTED → active.
+ producer_->state(State::STARTED);
+ EXPECT_TRUE(producer_->active());
+
+ // Transition to STOPPED → not active.
+ producer_->state(State::STOPPED);
+ EXPECT_FALSE(producer_->active());
+}
+
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
b/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
new file mode 100644
index 00000000..a039b042
--- /dev/null
+++ b/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
@@ -0,0 +1,323 @@
+/*
+ * 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.
+ */
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <system_error>
+#include <vector>
+
+#include "ConsumeTask.h"
+#include "ConsumeMessageService.h"
+#include "ProcessQueue.h"
+#include "PushConsumerImpl.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "rocketmq/ConsumeResult.h"
+#include "rocketmq/Message.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+/**
+ * Local mock of ConsumeMessageService matching the current interface.
+ * The mock shipped in mocks/include/ is stale (missing submit, ack, nack,
+ * forward, schedule, listener, consumer, etc.), so we define one here.
+ */
+class ConsumeTaskServiceMock : public ConsumeMessageService {
+public:
+ MOCK_METHOD(void, start, (), (override));
+ MOCK_METHOD(void, shutdown, (), (override));
+ MOCK_METHOD(void, dispatch, (std::shared_ptr<ProcessQueue>,
std::vector<MessageConstSharedPtr>), (override));
+ MOCK_METHOD(void, submit, (std::shared_ptr<ConsumeTask>), (override));
+ MOCK_METHOD(MessageListener&, listener, (), (override));
+ MOCK_METHOD(bool, preHandle, (const Message&), (override));
+ MOCK_METHOD(bool, postHandle, (const Message&, ConsumeResult), (override));
+ MOCK_METHOD(void, ack, (const Message&, std::function<void(const
std::error_code&)>), (override));
+ MOCK_METHOD(void, nack, (const Message&, std::function<void(const
std::error_code&)>), (override));
+ MOCK_METHOD(void, forward, (const Message&, std::function<void(const
std::error_code&)>), (override));
+ MOCK_METHOD(void, schedule, (std::shared_ptr<ConsumeTask>,
std::chrono::milliseconds), (override));
+ MOCK_METHOD(std::size_t, maxDeliveryAttempt, (), (override));
+ MOCK_METHOD(std::weak_ptr<PushConsumerImpl>, consumer, (), (override));
+};
+
+/**
+ * Local mock of ProcessQueue matching the current interface.
+ * The mock shipped in mocks/include/ is stale as well.
+ */
+class ConsumeTaskProcessQueueMock : public ProcessQueue {
+public:
+ MOCK_METHOD(bool, expired, (), (const, override));
+ MOCK_METHOD(void, callback, (std::shared_ptr<AsyncReceiveMessageCallback>),
(override));
+ MOCK_METHOD(void, receiveMessage, (std::string&), (override));
+ MOCK_METHOD(std::string, topic, (), (const, override));
+ MOCK_METHOD(std::weak_ptr<PushConsumerImpl>, getConsumer, (), (override));
+ MOCK_METHOD(const std::string&, simpleName, (), (const, override));
+ MOCK_METHOD(void, release, (uint64_t), (override));
+ MOCK_METHOD(void, accountCache, (const std::vector<MessageConstSharedPtr>&),
(override));
+ MOCK_METHOD(std::uint64_t, cachedMessageQuantity, (), (const, override));
+ MOCK_METHOD(std::uint64_t, cachedMessageMemory, (), (const, override));
+ MOCK_METHOD(bool, shouldThrottle, (), (const, override));
+ MOCK_METHOD(std::shared_ptr<ClientManager>, getClientManager, (),
(override));
+ MOCK_METHOD(void, syncIdleState, (), (override));
+ MOCK_METHOD(const FilterExpression&, getFilterExpression, (), (const,
override));
+ MOCK_METHOD(const rmq::MessageQueue&, messageQueue, (), (const, override));
+};
+
+class ConsumeTaskTest : public testing::Test {
+protected:
+ static MessageConstSharedPtr buildMessage(const std::string& topic, const
std::string& body) {
+ return
MessageConstSharedPtr(Message::newBuilder().withTopic(topic).withBody(body).build().release());
+ }
+
+ std::shared_ptr<testing::NiceMock<ConsumeTaskServiceMock>> service_{
+ std::make_shared<testing::NiceMock<ConsumeTaskServiceMock>>()};
+ std::shared_ptr<testing::NiceMock<ConsumeTaskProcessQueueMock>> pq_{
+ std::make_shared<testing::NiceMock<ConsumeTaskProcessQueueMock>>()};
+ std::weak_ptr<ProcessQueue> weak_pq_{pq_};
+ ConsumeMessageServiceWeakPtr weak_service_{service_};
+};
+
+// --- Construction ---
+
+TEST_F(ConsumeTaskTest, constructWithSingleMessageTest) {
+ auto msg = buildMessage("topic", "body");
+ ConsumeTask task(weak_service_, weak_pq_, msg);
+ // Construction must not crash; fifo_ should remain false for a single
message.
+}
+
+TEST_F(ConsumeTaskTest, constructWithBatchMessagesTest) {
+ std::vector<MessageConstSharedPtr> msgs;
+ msgs.push_back(buildMessage("topic", "body1"));
+ msgs.push_back(buildMessage("topic", "body2"));
+ ConsumeTask task(weak_service_, weak_pq_, std::move(msgs));
+ // Construction must not crash; fifo_ should be set to true (>1 messages).
+}
+
+// --- submit() ---
+
+TEST_F(ConsumeTaskTest, submitWithValidServiceTest) {
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ EXPECT_CALL(*service_, submit(testing::_)).Times(1);
+ task->submit();
+}
+
+TEST_F(ConsumeTaskTest, submitWithExpiredServiceTest) {
+ ConsumeMessageServiceWeakPtr expired;
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(expired, weak_pq_, msg);
+ // Service is expired — submit() should return silently.
+ task->submit();
+}
+
+// --- schedule() ---
+
+TEST_F(ConsumeTaskTest, scheduleWithValidServiceTest) {
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ EXPECT_CALL(*service_, schedule(testing::_,
std::chrono::milliseconds(1000))).Times(1);
+ task->schedule();
+}
+
+TEST_F(ConsumeTaskTest, scheduleWithExpiredServiceTest) {
+ ConsumeMessageServiceWeakPtr expired;
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(expired, weak_pq_, msg);
+ // Service is expired — schedule() should return silently.
+ task->schedule();
+}
+
+// --- process() early-return paths ---
+
+TEST_F(ConsumeTaskTest, processWithExpiredServiceTest) {
+ ConsumeMessageServiceWeakPtr expired;
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(expired, weak_pq_, msg);
+ // Service is expired — process() returns early without touching any mock.
+ task->process();
+}
+
+TEST_F(ConsumeTaskTest, processWithEmptyMessagesTest) {
+ std::vector<MessageConstSharedPtr> empty;
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_,
std::move(empty));
+ // No messages cached — process() should return after the emptiness check.
+ task->process();
+}
+
+// --- process() state-machine: Consume → Ack on SUCCESS ---
+
+TEST_F(ConsumeTaskTest, processConsumeSuccessCallsAckTest) {
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ auto consumer = std::make_shared<PushConsumerImpl>("test-group");
+ MessageListener listener = [](const Message&) { return
ConsumeResult::SUCCESS; };
+
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>(consumer)));
+ EXPECT_CALL(*service_,
listener()).WillRepeatedly(testing::ReturnRef(listener));
+ EXPECT_CALL(*service_,
preHandle(testing::_)).WillOnce(testing::Return(true));
+ EXPECT_CALL(*service_, postHandle(testing::_,
ConsumeResult::SUCCESS)).WillOnce(testing::Return(true));
+
+ // On success process() calls ack(); the callback invokes onAck() which pops
+ // the head message and resubmits the task for the remaining messages.
+ EXPECT_CALL(*service_, ack(testing::_, testing::_))
+ .WillOnce(testing::Invoke([](const Message&, std::function<void(const
std::error_code&)> cb) {
+ cb(std::error_code{});
+ }));
+ EXPECT_CALL(*pq_, release(testing::_)).Times(1);
+ EXPECT_CALL(*service_, submit(testing::_)).Times(1);
+
+ task->process();
+}
+
+// --- process() state-machine: Consume → Nack on FAILURE (non-FIFO) ---
+
+TEST_F(ConsumeTaskTest, processConsumeFailureNonFifoCallsNackTest) {
+ // Single message ⇒ fifo_ == false
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ auto consumer = std::make_shared<PushConsumerImpl>("test-group");
+ MessageListener listener = [](const Message&) { return
ConsumeResult::FAILURE; };
+
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>(consumer)));
+ EXPECT_CALL(*service_,
listener()).WillRepeatedly(testing::ReturnRef(listener));
+ EXPECT_CALL(*service_,
preHandle(testing::_)).WillOnce(testing::Return(true));
+ EXPECT_CALL(*service_, postHandle(testing::_,
ConsumeResult::FAILURE)).WillOnce(testing::Return(true));
+
+ // On failure in non-FIFO mode process() calls nack(); the callback invokes
+ // onNack() which pops the head message and resubmits the task.
+ EXPECT_CALL(*service_, nack(testing::_, testing::_))
+ .WillOnce(testing::Invoke([](const Message&, std::function<void(const
std::error_code&)> cb) {
+ cb(std::error_code{});
+ }));
+ EXPECT_CALL(*pq_, release(testing::_)).Times(1);
+ EXPECT_CALL(*service_, submit(testing::_)).Times(1);
+
+ task->process();
+}
+
+// --- process() state-machine: Consume → schedule retry on FAILURE (FIFO) ---
+
+TEST_F(ConsumeTaskTest, processConsumeFailureFifoSchedulesRetryTest) {
+ // Multiple messages ⇒ fifo_ == true
+ std::vector<MessageConstSharedPtr> msgs;
+ msgs.push_back(buildMessage("topic", "body1"));
+ msgs.push_back(buildMessage("topic", "body2"));
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_,
std::move(msgs));
+
+ auto consumer = std::make_shared<PushConsumerImpl>("test-group");
+ MessageListener listener = [](const Message&) { return
ConsumeResult::FAILURE; };
+
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>(consumer)));
+ EXPECT_CALL(*service_,
listener()).WillRepeatedly(testing::ReturnRef(listener));
+ EXPECT_CALL(*service_,
preHandle(testing::_)).WillOnce(testing::Return(true));
+ EXPECT_CALL(*service_, postHandle(testing::_,
ConsumeResult::FAILURE)).WillOnce(testing::Return(true));
+
+ // In FIFO mode, failure does not call nack(); instead it increments
+ // delivery_attempt and schedules a retry after 1 second.
+ EXPECT_CALL(*service_, nack(testing::_, testing::_)).Times(0);
+ EXPECT_CALL(*service_, schedule(testing::_,
std::chrono::milliseconds(1000))).Times(1);
+
+ task->process();
+}
+
+// --- onAck retry path: ack failure re-schedules ---
+
+TEST_F(ConsumeTaskTest, processAckFailureSchedulesRetryTest) {
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ auto consumer = std::make_shared<PushConsumerImpl>("test-group");
+ MessageListener listener = [](const Message&) { return
ConsumeResult::SUCCESS; };
+
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>(consumer)));
+ EXPECT_CALL(*service_,
listener()).WillRepeatedly(testing::ReturnRef(listener));
+ EXPECT_CALL(*service_,
preHandle(testing::_)).WillOnce(testing::Return(true));
+ EXPECT_CALL(*service_, postHandle(testing::_,
ConsumeResult::SUCCESS)).WillOnce(testing::Return(true));
+
+ // Simulate an ack RPC failure — onAck should set next_step_ back to Ack
+ // and schedule a retry rather than popping the message.
+ EXPECT_CALL(*service_, ack(testing::_, testing::_))
+ .WillOnce(testing::Invoke([](const Message&, std::function<void(const
std::error_code&)> cb) {
+ cb(std::make_error_code(std::errc::connection_refused));
+ }));
+ EXPECT_CALL(*pq_, release(testing::_)).Times(0);
+ EXPECT_CALL(*service_, schedule(testing::_,
std::chrono::milliseconds(1000))).Times(1);
+
+ task->process();
+}
+
+// --- onNack retry path: nack failure re-schedules ---
+
+TEST_F(ConsumeTaskTest, processNackFailureSchedulesRetryTest) {
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ auto consumer = std::make_shared<PushConsumerImpl>("test-group");
+ MessageListener listener = [](const Message&) { return
ConsumeResult::FAILURE; };
+
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>(consumer)));
+ EXPECT_CALL(*service_,
listener()).WillRepeatedly(testing::ReturnRef(listener));
+ EXPECT_CALL(*service_,
preHandle(testing::_)).WillOnce(testing::Return(true));
+ EXPECT_CALL(*service_, postHandle(testing::_,
ConsumeResult::FAILURE)).WillOnce(testing::Return(true));
+
+ // Simulate a nack RPC failure — onNack should set next_step_ back to Nack
+ // and schedule a retry rather than popping the message.
+ EXPECT_CALL(*service_, nack(testing::_, testing::_))
+ .WillOnce(testing::Invoke([](const Message&, std::function<void(const
std::error_code&)> cb) {
+ cb(std::make_error_code(std::errc::connection_refused));
+ }));
+ EXPECT_CALL(*pq_, release(testing::_)).Times(0);
+ EXPECT_CALL(*service_, schedule(testing::_,
std::chrono::milliseconds(1000))).Times(1);
+
+ task->process();
+}
+
+// --- pop() with expired ProcessQueue: must not crash ---
+
+TEST_F(ConsumeTaskTest, processAckSuccessWithExpiredProcessQueueTest) {
+ std::weak_ptr<ProcessQueue> expired_pq;
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, expired_pq, msg);
+
+ auto consumer = std::make_shared<PushConsumerImpl>("test-group");
+ MessageListener listener = [](const Message&) { return
ConsumeResult::SUCCESS; };
+
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>(consumer)));
+ EXPECT_CALL(*service_,
listener()).WillRepeatedly(testing::ReturnRef(listener));
+ EXPECT_CALL(*service_,
preHandle(testing::_)).WillOnce(testing::Return(true));
+ EXPECT_CALL(*service_, postHandle(testing::_,
ConsumeResult::SUCCESS)).WillOnce(testing::Return(true));
+
+ // ack succeeds; onAck calls pop() which finds the ProcessQueue expired —
+ // it should return gracefully without crashing.
+ EXPECT_CALL(*service_, ack(testing::_, testing::_))
+ .WillOnce(testing::Invoke([](const Message&, std::function<void(const
std::error_code&)> cb) {
+ cb(std::error_code{});
+ }));
+ EXPECT_CALL(*service_, submit(testing::_)).Times(1);
+
+ task->process();
+}
+
+} // namespace
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
b/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
new file mode 100644
index 00000000..ec324ad3
--- /dev/null
+++ b/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
@@ -0,0 +1,366 @@
+/*
+ * 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.
+ */
+#include <atomic>
+#include <memory>
+#include <string>
+#include <system_error>
+#include <thread>
+#include <vector>
+
+#include "ClientManagerMock.h"
+#include "FifoContext.h"
+#include "FifoProducerPartition.h"
+#include "MixAll.h"
+#include "NameServerResolverMock.h"
+#include "ProducerImpl.h"
+#include "RpcClientMock.h"
+#include "SendResult.h"
+#include "TelemetryBidiReactor.h"
+#include "TopicRouteData.h"
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+#include "rocketmq/Message.h"
+#include "rocketmq/SendCallback.h"
+#include "rocketmq/SendReceipt.h"
+
+#include "absl/synchronization/mutex.h"
+#include "grpcpp/create_channel.h"
+#include "grpcpp/security/credentials.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+rmq::MessageQueue createMessageQueue(const std::string& topic, int id, const
std::string& broker_name) {
+ rmq::MessageQueue mq;
+ mq.mutable_topic()->set_name(topic);
+ mq.set_id(id);
+ mq.mutable_broker()->set_name(broker_name);
+ mq.mutable_broker()->set_id(MixAll::MASTER_BROKER_ID);
+
mq.mutable_broker()->mutable_endpoints()->set_scheme(rmq::AddressScheme::IPv4);
+ auto* addr = mq.mutable_broker()->mutable_endpoints()->add_addresses();
+ addr->set_host("127.0.0.1");
+ addr->set_port(19999);
+ mq.set_permission(rmq::Permission::READ_WRITE);
+ return mq;
+}
+
+TopicRouteDataPtr createRouteData(const std::string& topic, int num_queues =
1) {
+ std::vector<rmq::MessageQueue> queues;
+ for (int i = 0; i < num_queues; i++) {
+ queues.push_back(createMessageQueue(topic, i, "broker-0"));
+ }
+ return std::make_shared<TopicRouteData>(queues);
+}
+
+/// A minimal RpcClient mock that supports asyncTelemetry with a real stub
+class TestRpcClient : public testing::NiceMock<RpcClientMock> {
+public:
+ TestRpcClient() {
+ channel_ = grpc::CreateChannel("127.0.0.1:19999",
grpc::InsecureChannelCredentials());
+ stub_ = rmq::MessagingService::NewStub(channel_);
+ static const std::string addr = "127.0.0.1:19999";
+ ON_CALL(*this, remoteAddress).WillByDefault(testing::ReturnRef(addr));
+ }
+
+ std::shared_ptr<TelemetryBidiReactor> asyncTelemetry(std::weak_ptr<Client>
/*client*/) override {
+ // Pass an expired weak_ptr so TelemetryBidiReactor short-circuits without
initiating gRPC calls
+ std::shared_ptr<Client> expired;
+ std::weak_ptr<Client> weak_expired(expired);
+ return std::make_shared<TelemetryBidiReactor>(weak_expired, stub_.get(),
"127.0.0.1:19999");
+ }
+
+private:
+ std::shared_ptr<grpc::Channel> channel_;
+ std::unique_ptr<rmq::MessagingService::Stub> stub_;
+};
+
+} // namespace
+
+class FifoProducerPartitionTest : public testing::Test {
+protected:
+ void SetUp() override {
+ producer_ = std::make_shared<ProducerImpl>();
+ producer_->state(State::STARTED);
+ producer_->maxAttemptTimes(1);
+
+ resolver_mock_ =
std::make_shared<testing::NiceMock<NameServerResolverMock>>();
+ ON_CALL(*resolver_mock_,
resolve).WillByDefault(testing::Return(std::string("127.0.0.1:9876")));
+ producer_->withNameServerResolver(resolver_mock_);
+
+ client_manager_ = std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ producer_->clientManager(client_manager_);
+
+ // Mock getRpcClient to return a test RpcClient that supports telemetry
+ rpc_client_ = std::make_shared<TestRpcClient>();
+ ON_CALL(*client_manager_, getRpcClient)
+ .WillByDefault(testing::Return(rpc_client_));
+
+ // Mock resolveRoute to return valid route data
+ ON_CALL(*client_manager_, resolveRoute)
+ .WillByDefault(testing::Invoke(
+ [](const std::string&, const Metadata&, const QueryRouteRequest&
request,
+ std::chrono::milliseconds,
+ const std::function<void(const std::error_code&, const
TopicRouteDataPtr&)>& cb) {
+ std::string topic = request.topic().name();
+ std::error_code ec;
+ cb(ec, createRouteData(topic));
+ }));
+
+ partition_ = std::make_shared<FifoProducerPartition>(producer_,
std::string("test-partition"));
+ }
+
+ void TearDown() override {
+ // Wait briefly for pending async operations to complete
+ absl::SleepFor(absl::Milliseconds(200));
+ // Reset partition first
+ partition_.reset();
+ // Reset all shared_ptrs to ensure clean destruction order
+ producer_.reset();
+ rpc_client_.reset();
+ client_manager_.reset();
+ resolver_mock_.reset();
+ }
+
+ void installSendSuccessHandler() {
+ ON_CALL(*client_manager_, send)
+ .WillByDefault(testing::Invoke(
+ [this](const std::string&, const Metadata&, SendMessageRequest&,
SendResultCallback cb) {
+ {
+ absl::MutexLock lk(&mtx_);
+ send_count_++;
+ cv_.SignalAll();
+ }
+ // Dispatch callback asynchronously to avoid
FifoProducerPartition mutex deadlock
+ std::thread([this, cb]() {
+ SendResult result;
+ result.message_id = "msg-id";
+ cb(result);
+ // Signal after callback completes so waitForResults can
detect it
+ cv_.SignalAll();
+ }).detach();
+ return true;
+ }));
+ }
+
+ SendCallback makeResultCallback() {
+ return [this](const std::error_code& ec, SendReceipt&&) {
+ absl::MutexLock lk(&mtx_);
+ results_.push_back(ec);
+ cv_.SignalAll();
+ };
+ }
+
+ MessageConstPtr makeMessage(const std::string& body = "test-body") {
+ return
Message::newBuilder().withTopic("test-topic").withBody(body).build();
+ }
+
+ void waitForSends(int expected) {
+ absl::MutexLock lk(&mtx_);
+ auto deadline = absl::Now() + absl::Seconds(3);
+ while (send_count_ < expected) {
+ if (cv_.WaitWithDeadline(&mtx_, deadline)) {
+ break; // timeout
+ }
+ }
+ }
+
+ void waitForResults(int expected) {
+ absl::MutexLock lk(&mtx_);
+ auto deadline = absl::Now() + absl::Seconds(3);
+ while (static_cast<int>(results_.size()) < expected) {
+ if (cv_.WaitWithDeadline(&mtx_, deadline)) {
+ break; // timeout
+ }
+ }
+ }
+
+ std::shared_ptr<ProducerImpl> producer_;
+ std::shared_ptr<testing::NiceMock<NameServerResolverMock>> resolver_mock_;
+ std::shared_ptr<testing::NiceMock<ClientManagerMock>> client_manager_;
+ std::shared_ptr<TestRpcClient> rpc_client_;
+ std::shared_ptr<FifoProducerPartition> partition_;
+
+ absl::Mutex mtx_;
+ absl::CondVar cv_;
+ int send_count_ GUARDED_BY(mtx_) = 0;
+ std::vector<std::error_code> results_;
+};
+
+TEST_F(FifoProducerPartitionTest, addTriggersSendTest) {
+ installSendSuccessHandler();
+
+ auto msg = makeMessage();
+ SendCallback cb = makeResultCallback();
+ FifoContext ctx(std::move(msg), std::move(cb));
+
+ partition_->add(std::move(ctx));
+ waitForSends(1);
+ waitForResults(1);
+
+ absl::MutexLock lk(&mtx_);
+ EXPECT_EQ(1, send_count_);
+ ASSERT_EQ(1u, results_.size());
+ EXPECT_FALSE(results_[0]);
+}
+
+TEST_F(FifoProducerPartitionTest, multipleMessagesPreserveOrderTest) {
+ std::vector<std::string> sent_bodies;
+
+ ON_CALL(*client_manager_, send)
+ .WillByDefault(testing::Invoke(
+ [this, &sent_bodies](const std::string&, const Metadata&,
SendMessageRequest& request,
+ SendResultCallback cb) {
+ {
+ absl::MutexLock lk(&mtx_);
+ if (request.messages_size() > 0) {
+ sent_bodies.push_back(request.messages(0).body());
+ }
+ send_count_++;
+ cv_.SignalAll();
+ }
+ std::thread([cb]() {
+ SendResult result;
+ result.message_id = "msg-id";
+ cb(result);
+ }).detach();
+ return true;
+ }));
+
+ const int num_messages = 5;
+ for (int i = 0; i < num_messages; i++) {
+ auto msg = makeMessage("msg-" + std::to_string(i));
+ SendCallback cb = makeResultCallback();
+ FifoContext ctx(std::move(msg), std::move(cb));
+ partition_->add(std::move(ctx));
+ }
+
+ waitForSends(num_messages);
+ waitForResults(num_messages);
+
+ absl::MutexLock lk(&mtx_);
+ EXPECT_EQ(num_messages, send_count_);
+ ASSERT_EQ(static_cast<size_t>(num_messages), results_.size());
+ for (int i = 0; i < num_messages; i++) {
+ EXPECT_FALSE(results_[i]);
+ }
+
+ // Verify messages were sent in FIFO order
+ ASSERT_EQ(static_cast<size_t>(num_messages), sent_bodies.size());
+ for (int i = 0; i < num_messages; i++) {
+ EXPECT_EQ("msg-" + std::to_string(i), sent_bodies[i])
+ << "Message at index " << i << " was out of order";
+ }
+}
+
+TEST_F(FifoProducerPartitionTest, onCompleteSuccessInvokesUserCallbackTest) {
+ bool callback_invoked = false;
+ std::error_code received_ec;
+ SendCallback user_cb = [&callback_invoked, &received_ec](const
std::error_code& ec, SendReceipt&&) {
+ callback_invoked = true;
+ received_ec = ec;
+ };
+
+ SendReceipt receipt;
+ receipt.message = makeMessage();
+ receipt.message_id = "test-id";
+ std::error_code ec;
+
+ partition_->onComplete(ec, std::move(receipt), user_cb);
+
+ EXPECT_TRUE(callback_invoked);
+ EXPECT_FALSE(received_ec);
+}
+
+TEST_F(FifoProducerPartitionTest, onCompleteFailureRequeuesMessageTest) {
+ bool callback_invoked = false;
+ SendCallback user_cb = [&callback_invoked](const std::error_code&,
SendReceipt&&) {
+ callback_invoked = true;
+ };
+
+ SendReceipt receipt;
+ receipt.message = makeMessage("retry-message");
+ std::error_code ec = std::make_error_code(std::errc::io_error);
+
+ partition_->onComplete(ec, std::move(receipt), user_cb);
+
+ // On failure, user callback should NOT be invoked (message is requeued
instead)
+ EXPECT_FALSE(callback_invoked);
+}
+
+TEST_F(FifoProducerPartitionTest, failedMessageRetriedViaOnCompleteTest) {
+ // Install handler that fails once then succeeds
+ std::atomic<int> attempt{0};
+ ON_CALL(*client_manager_, send)
+ .WillByDefault(testing::Invoke(
+ [this, &attempt](const std::string&, const Metadata&,
SendMessageRequest&, SendResultCallback cb) {
+ int current = attempt.fetch_add(1);
+ {
+ absl::MutexLock lk(&mtx_);
+ send_count_++;
+ cv_.SignalAll();
+ }
+ std::thread([cb, current]() {
+ SendResult result;
+ if (current == 0) {
+ result.ec = std::make_error_code(std::errc::io_error);
+ }
+ result.message_id = "msg-id";
+ cb(result);
+ }).detach();
+ return true;
+ }));
+
+ // Add a message - first attempt fails, FifoProducerPartition retries
+ auto msg = makeMessage("retry-msg");
+ SendCallback cb = makeResultCallback();
+ FifoContext ctx(std::move(msg), std::move(cb));
+ partition_->add(std::move(ctx));
+
+ // Wait for 2 send attempts (1 failure + 1 success)
+ waitForSends(2);
+ waitForResults(1);
+
+ absl::MutexLock lk(&mtx_);
+ EXPECT_GE(send_count_, 2);
+ ASSERT_GE(results_.size(), 1u);
+ EXPECT_FALSE(results_.back());
+}
+
+TEST_F(FifoProducerPartitionTest, emptyPartitionTrySendDoesNotCrashTest) {
+ installSendSuccessHandler();
+ partition_->trySend();
+
+ absl::MutexLock lk(&mtx_);
+ EXPECT_EQ(0, send_count_);
+}
+
+TEST_F(FifoProducerPartitionTest, fifoContextMoveConstructionTest) {
+ auto msg = makeMessage("move-test");
+ SendCallback cb = makeResultCallback();
+ FifoContext ctx1(std::move(msg), std::move(cb));
+
+ EXPECT_NE(nullptr, ctx1.message);
+ EXPECT_TRUE(ctx1.callback != nullptr);
+
+ FifoContext ctx2(std::move(ctx1));
+ EXPECT_NE(nullptr, ctx2.message);
+ EXPECT_TRUE(ctx2.callback != nullptr);
+ EXPECT_EQ(nullptr, ctx1.message);
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/ProcessQueueImplTest.cpp
b/cpp/source/rocketmq/tests/ProcessQueueImplTest.cpp
new file mode 100644
index 00000000..5b9c2a0d
--- /dev/null
+++ b/cpp/source/rocketmq/tests/ProcessQueueImplTest.cpp
@@ -0,0 +1,249 @@
+/*
+ * 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.
+ */
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "ClientManagerMock.h"
+#include "MixAll.h"
+#include "ProcessQueueImpl.h"
+#include "PushConsumerImpl.h"
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+#include "rocketmq/FilterExpression.h"
+#include "rocketmq/Message.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+rmq::MessageQueue createMessageQueue(const std::string& topic) {
+ rmq::MessageQueue mq;
+ mq.mutable_topic()->set_name(topic);
+ mq.set_id(0);
+ mq.mutable_broker()->set_name("broker-0");
+ mq.mutable_broker()->set_id(MixAll::MASTER_BROKER_ID);
+
mq.mutable_broker()->mutable_endpoints()->set_scheme(rmq::AddressScheme::IPv4);
+ auto* addr = mq.mutable_broker()->mutable_endpoints()->add_addresses();
+ addr->set_host("10.0.0.1");
+ addr->set_port(8080);
+ mq.set_permission(rmq::Permission::READ_WRITE);
+ return mq;
+}
+
+MessageConstSharedPtr makeMessage(const std::string& topic, const std::string&
body) {
+ auto ptr = Message::newBuilder().withTopic(topic).withBody(body).build();
+ return MessageConstSharedPtr(std::move(ptr));
+}
+
+} // namespace
+
+class ProcessQueueImplTest : public testing::Test {
+protected:
+ void SetUp() override {
+ consumer_ = std::make_shared<PushConsumerImpl>("test-group");
+ message_queue_ = createMessageQueue("test-topic");
+ filter_expression_ = FilterExpression("test-tag", TAG);
+ client_manager_ = std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ process_queue_.reset(
+ new ProcessQueueImpl(message_queue_, filter_expression_, consumer_,
client_manager_));
+ }
+
+ void TearDown() override {
+ process_queue_.reset();
+ consumer_.reset();
+ }
+
+ std::shared_ptr<PushConsumerImpl> consumer_;
+ rmq::MessageQueue message_queue_;
+ FilterExpression filter_expression_{"*", TAG};
+ std::shared_ptr<testing::NiceMock<ClientManagerMock>> client_manager_;
+ std::unique_ptr<ProcessQueueImpl> process_queue_;
+};
+
+TEST_F(ProcessQueueImplTest, cachedMessageQuantityStartsAtZeroTest) {
+ EXPECT_EQ(0u, process_queue_->cachedMessageQuantity());
+}
+
+TEST_F(ProcessQueueImplTest, cachedMessageMemoryStartsAtZeroTest) {
+ EXPECT_EQ(0u, process_queue_->cachedMessageMemory());
+}
+
+TEST_F(ProcessQueueImplTest, topicReturnsCorrectValueTest) {
+ EXPECT_EQ("test-topic", process_queue_->topic());
+}
+
+TEST_F(ProcessQueueImplTest, getFilterExpressionReturnsConfiguredFilterTest) {
+ const auto& fe = process_queue_->getFilterExpression();
+ EXPECT_EQ("test-tag", fe.content_);
+ EXPECT_EQ(TAG, fe.type_);
+}
+
+TEST_F(ProcessQueueImplTest, getFilterExpressionSqlTypeTest) {
+ FilterExpression sql_filter("price > 10", SQL92);
+ ProcessQueueImpl pq(message_queue_, sql_filter, consumer_, client_manager_);
+ const auto& fe = pq.getFilterExpression();
+ EXPECT_EQ("price > 10", fe.content_);
+ EXPECT_EQ(SQL92, fe.type_);
+}
+
+TEST_F(ProcessQueueImplTest, accountCacheUpdatesQuantityAndMemoryTest) {
+ std::vector<MessageConstSharedPtr> messages;
+ messages.push_back(makeMessage("test-topic", "body12345")); // 9 bytes
+ messages.push_back(makeMessage("test-topic", "abcde")); // 5 bytes
+
+ process_queue_->accountCache(messages);
+
+ EXPECT_EQ(2u, process_queue_->cachedMessageQuantity());
+ EXPECT_EQ(14u, process_queue_->cachedMessageMemory());
+}
+
+TEST_F(ProcessQueueImplTest, accountCacheMultipleBatchesAccumulateTest) {
+ std::vector<MessageConstSharedPtr> batch1;
+ batch1.push_back(makeMessage("test-topic", "aaa")); // 3 bytes
+ process_queue_->accountCache(batch1);
+
+ std::vector<MessageConstSharedPtr> batch2;
+ batch2.push_back(makeMessage("test-topic", "bbbbb")); // 5 bytes
+ process_queue_->accountCache(batch2);
+
+ EXPECT_EQ(2u, process_queue_->cachedMessageQuantity());
+ EXPECT_EQ(8u, process_queue_->cachedMessageMemory());
+}
+
+TEST_F(ProcessQueueImplTest, releaseDecrementsCacheTest) {
+ std::vector<MessageConstSharedPtr> messages;
+ messages.push_back(makeMessage("test-topic", "body12345")); // 9 bytes
+ messages.push_back(makeMessage("test-topic", "abcde")); // 5 bytes
+ process_queue_->accountCache(messages);
+
+ // Release one message with body size 9
+ process_queue_->release(9);
+
+ EXPECT_EQ(1u, process_queue_->cachedMessageQuantity());
+ EXPECT_EQ(5u, process_queue_->cachedMessageMemory());
+}
+
+TEST_F(ProcessQueueImplTest, releaseAllMessagesTest) {
+ std::vector<MessageConstSharedPtr> messages;
+ messages.push_back(makeMessage("test-topic", "abc")); // 3 bytes
+ process_queue_->accountCache(messages);
+
+ process_queue_->release(3);
+
+ EXPECT_EQ(0u, process_queue_->cachedMessageQuantity());
+ EXPECT_EQ(0u, process_queue_->cachedMessageMemory());
+}
+
+TEST_F(ProcessQueueImplTest, shouldThrottleReturnsFalseWhenCacheEmptyTest) {
+ EXPECT_FALSE(process_queue_->shouldThrottle());
+}
+
+TEST_F(ProcessQueueImplTest, shouldThrottleReturnsFalseBelowThresholdTest) {
+ // Cache a few messages, well below DEFAULT_CACHED_MESSAGE_COUNT (1024)
+ std::vector<MessageConstSharedPtr> messages;
+ for (int i = 0; i < 10; i++) {
+ messages.push_back(makeMessage("test-topic", "x"));
+ }
+ process_queue_->accountCache(messages);
+
+ EXPECT_FALSE(process_queue_->shouldThrottle());
+}
+
+TEST_F(ProcessQueueImplTest,
shouldThrottleReturnsTrueWhenQuantityExceedsThresholdTest) {
+ // DEFAULT_CACHED_MESSAGE_COUNT is 1024
+ uint32_t threshold = MixAll::DEFAULT_CACHED_MESSAGE_COUNT;
+ std::vector<MessageConstSharedPtr> messages;
+ for (uint32_t i = 0; i <= threshold; i++) {
+ messages.push_back(makeMessage("test-topic", "x"));
+ }
+ process_queue_->accountCache(messages);
+
+ EXPECT_TRUE(process_queue_->shouldThrottle());
+}
+
+TEST_F(ProcessQueueImplTest,
shouldThrottleReturnsFalseAfterReleasingMessagesTest) {
+ // Exceed threshold
+ uint32_t threshold = MixAll::DEFAULT_CACHED_MESSAGE_COUNT;
+ std::vector<MessageConstSharedPtr> messages;
+ for (uint32_t i = 0; i <= threshold; i++) {
+ messages.push_back(makeMessage("test-topic", "x"));
+ }
+ process_queue_->accountCache(messages);
+ EXPECT_TRUE(process_queue_->shouldThrottle());
+
+ // Release enough messages to drop below threshold (need to go below 1024)
+ process_queue_->release(1);
+ process_queue_->release(1);
+ EXPECT_FALSE(process_queue_->shouldThrottle());
+}
+
+TEST_F(ProcessQueueImplTest, shouldThrottleReturnsFalseWithNullConsumerTest) {
+ // Create ProcessQueue with an expired (null) consumer weak_ptr
+ std::weak_ptr<PushConsumerImpl> null_consumer;
+ ProcessQueueImpl pq(message_queue_, filter_expression_, null_consumer,
client_manager_);
+
+ // shouldThrottle returns false when consumer is gone
+ EXPECT_FALSE(pq.shouldThrottle());
+}
+
+TEST_F(ProcessQueueImplTest, accountCacheWithNullConsumerDoesNotIncrementTest)
{
+ std::weak_ptr<PushConsumerImpl> null_consumer;
+ ProcessQueueImpl pq(message_queue_, filter_expression_, null_consumer,
client_manager_);
+
+ std::vector<MessageConstSharedPtr> messages;
+ messages.push_back(makeMessage("test-topic", "body"));
+ pq.accountCache(messages);
+
+ // accountCache returns early when consumer is null, so cache stays at 0
+ EXPECT_EQ(0u, pq.cachedMessageQuantity());
+ EXPECT_EQ(0u, pq.cachedMessageMemory());
+}
+
+TEST_F(ProcessQueueImplTest, expiredReturnsFalseImmediatelyAfterCreationTest) {
+ // Just created, idle_since_ = now, so duration < 120s threshold
+ EXPECT_FALSE(process_queue_->expired());
+}
+
+TEST_F(ProcessQueueImplTest, expiredReturnsFalseAfterSyncIdleStateTest) {
+ // Sync idle state resets the timer
+ process_queue_->syncIdleState();
+ EXPECT_FALSE(process_queue_->expired());
+}
+
+TEST_F(ProcessQueueImplTest, simpleNameIsNotEmptyTest) {
+ EXPECT_FALSE(process_queue_->simpleName().empty());
+}
+
+TEST_F(ProcessQueueImplTest, messageQueueReturnsCorrectQueueTest) {
+ const auto& mq = process_queue_->messageQueue();
+ EXPECT_EQ("test-topic", mq.topic().name());
+ EXPECT_EQ(0, mq.id());
+ EXPECT_EQ("broker-0", mq.broker().name());
+}
+
+TEST_F(ProcessQueueImplTest, getConsumerReturnsValidWeakPtrTest) {
+ auto consumer = process_queue_->getConsumer().lock();
+ EXPECT_NE(nullptr, consumer);
+}
+
+TEST_F(ProcessQueueImplTest, getClientManagerReturnsCorrectInstanceTest) {
+ auto cm = process_queue_->getClientManager();
+ EXPECT_EQ(client_manager_.get(), cm.get());
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/ProducerImplTest.cpp
b/cpp/source/rocketmq/tests/ProducerImplTest.cpp
new file mode 100644
index 00000000..011eb3ce
--- /dev/null
+++ b/cpp/source/rocketmq/tests/ProducerImplTest.cpp
@@ -0,0 +1,284 @@
+/*
+ * 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.
+ */
+#include <atomic>
+#include <memory>
+#include <string>
+#include <system_error>
+#include <thread>
+#include <vector>
+
+#include "ClientManagerMock.h"
+#include "MixAll.h"
+#include "NameServerResolverMock.h"
+#include "ProducerImpl.h"
+#include "RpcClientMock.h"
+#include "SendResult.h"
+#include "TelemetryBidiReactor.h"
+#include "TopicRouteData.h"
+#include "TransactionImpl.h"
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+#include "rocketmq/ErrorCode.h"
+#include "rocketmq/Message.h"
+#include "rocketmq/SendCallback.h"
+#include "rocketmq/SendReceipt.h"
+
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+#include "grpcpp/create_channel.h"
+#include "grpcpp/security/credentials.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+rmq::MessageQueue createMessageQueue(const std::string& topic, int id, const
std::string& broker_name) {
+ rmq::MessageQueue mq;
+ mq.mutable_topic()->set_name(topic);
+ mq.set_id(id);
+ mq.mutable_broker()->set_name(broker_name);
+ mq.mutable_broker()->set_id(MixAll::MASTER_BROKER_ID);
+
mq.mutable_broker()->mutable_endpoints()->set_scheme(rmq::AddressScheme::IPv4);
+ auto* addr = mq.mutable_broker()->mutable_endpoints()->add_addresses();
+ addr->set_host("127.0.0.1");
+ addr->set_port(19999);
+ mq.set_permission(rmq::Permission::READ_WRITE);
+ return mq;
+}
+
+TopicRouteDataPtr createRouteData(const std::string& topic, int num_queues =
1) {
+ std::vector<rmq::MessageQueue> queues;
+ for (int i = 0; i < num_queues; i++) {
+ queues.push_back(createMessageQueue(topic, i, "broker-0"));
+ }
+ return std::make_shared<TopicRouteData>(queues);
+}
+
+/// A minimal RpcClient mock that supports asyncTelemetry with a real stub
+class TestRpcClient : public testing::NiceMock<RpcClientMock> {
+public:
+ TestRpcClient() {
+ channel_ = grpc::CreateChannel("127.0.0.1:19999",
grpc::InsecureChannelCredentials());
+ stub_ = rmq::MessagingService::NewStub(channel_);
+ static const std::string addr = "127.0.0.1:19999";
+ ON_CALL(*this, remoteAddress).WillByDefault(testing::ReturnRef(addr));
+ }
+
+ std::shared_ptr<TelemetryBidiReactor> asyncTelemetry(std::weak_ptr<Client>
/*client*/) override {
+ // Pass an expired weak_ptr so TelemetryBidiReactor short-circuits without
initiating gRPC calls
+ std::shared_ptr<Client> expired;
+ std::weak_ptr<Client> weak_expired(expired);
+ return std::make_shared<TelemetryBidiReactor>(weak_expired, stub_.get(),
"127.0.0.1:19999");
+ }
+
+private:
+ std::shared_ptr<grpc::Channel> channel_;
+ std::unique_ptr<rmq::MessagingService::Stub> stub_;
+};
+
+} // namespace
+
+class ProducerImplTest : public testing::Test {
+protected:
+ void SetUp() override {
+ producer_ = std::make_shared<ProducerImpl>();
+ producer_->state(State::STARTED);
+ producer_->maxAttemptTimes(1);
+
+ resolver_mock_ =
std::make_shared<testing::NiceMock<NameServerResolverMock>>();
+ ON_CALL(*resolver_mock_,
resolve).WillByDefault(testing::Return(std::string("127.0.0.1:9876")));
+ producer_->withNameServerResolver(resolver_mock_);
+
+ client_manager_ = std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ producer_->clientManager(client_manager_);
+
+ // Mock getRpcClient to return a test RpcClient that supports telemetry
+ rpc_client_ = std::make_shared<TestRpcClient>();
+ ON_CALL(*client_manager_, getRpcClient)
+ .WillByDefault(testing::Return(rpc_client_));
+
+ // Mock resolveRoute to return valid route data
+ ON_CALL(*client_manager_, resolveRoute)
+ .WillByDefault(testing::Invoke(
+ [](const std::string&, const Metadata&, const QueryRouteRequest&
request,
+ std::chrono::milliseconds,
+ const std::function<void(const std::error_code&, const
TopicRouteDataPtr&)>& cb) {
+ std::string topic = request.topic().name();
+ std::error_code ec;
+ cb(ec, createRouteData(topic));
+ }));
+ }
+
+ void TearDown() override {
+ // Wait briefly for pending async operations to complete
+ absl::SleepFor(absl::Milliseconds(200));
+ producer_.reset();
+ rpc_client_.reset();
+ client_manager_.reset();
+ resolver_mock_.reset();
+ }
+
+ void installSendSuccessHandler() {
+ ON_CALL(*client_manager_, send)
+ .WillByDefault(testing::Invoke(
+ [this](const std::string&, const Metadata&, SendMessageRequest&,
SendResultCallback cb) {
+ {
+ absl::MutexLock lk(&mtx_);
+ send_count_++;
+ cv_.SignalAll();
+ }
+ // Dispatch callback asynchronously to avoid deadlock
+ std::thread([this, cb]() {
+ SendResult result;
+ result.message_id = "msg-id";
+ cb(result);
+ cv_.SignalAll();
+ }).detach();
+ return true;
+ }));
+ }
+
+ MessageConstPtr makeMessage(const std::string& topic = "test-topic",
+ const std::string& body = "test-body") {
+ return Message::newBuilder().withTopic(topic).withBody(body).build();
+ }
+
+ std::shared_ptr<ProducerImpl> producer_;
+ std::shared_ptr<testing::NiceMock<NameServerResolverMock>> resolver_mock_;
+ std::shared_ptr<testing::NiceMock<ClientManagerMock>> client_manager_;
+ std::shared_ptr<TestRpcClient> rpc_client_;
+
+ absl::Mutex mtx_;
+ absl::CondVar cv_;
+ int send_count_ GUARDED_BY(mtx_) = 0;
+};
+
+// Test 1: Sending a message with an empty body should fail with
MessageBodyEmpty
+TEST_F(ProducerImplTest, validateEmptyBodyFailsTest) {
+ installSendSuccessHandler();
+
+ auto msg =
Message::newBuilder().withTopic("test-topic").withBody("").build();
+ std::error_code ec;
+ producer_->send(std::move(msg), ec);
+
+ EXPECT_EQ(ErrorCode::MessageBodyEmpty, ec);
+}
+
+// Test 2: Sending a message with an empty topic should fail with IllegalTopic
+TEST_F(ProducerImplTest, validateEmptyTopicFailsTest) {
+ installSendSuccessHandler();
+
+ // Build message with empty topic — body is non-empty so that validation
+ // reaches the topic check without short-circuiting on empty body first.
+ auto msg = Message::newBuilder().withTopic("").withBody("hello").build();
+ std::error_code ec;
+ producer_->send(std::move(msg), ec);
+
+ EXPECT_EQ(ErrorCode::IllegalTopic, ec);
+}
+
+// Test 3: Sending a message whose body exceeds max_body_size should fail with
PayloadTooLarge
+TEST_F(ProducerImplTest, validateOversizedBodyFailsTest) {
+ installSendSuccessHandler();
+
+ // Lower the threshold so we can test with a small oversized body
+ producer_->config().publisher.max_body_size = 100;
+
+ std::string oversized_body(200, 'x');
+ auto msg =
Message::newBuilder().withTopic("test-topic").withBody(oversized_body).build();
+ std::error_code ec;
+ producer_->send(std::move(msg), ec);
+
+ EXPECT_EQ(ErrorCode::PayloadTooLarge, ec);
+}
+
+// Test 4: Sending a normal message should pass validation and succeed
+TEST_F(ProducerImplTest, validateNormalMessagePassesTest) {
+ installSendSuccessHandler();
+
+ auto msg = makeMessage();
+ std::error_code ec;
+ SendReceipt receipt = producer_->send(std::move(msg), ec);
+
+ EXPECT_FALSE(ec) << "Expected no error, got: " << ec.message();
+ EXPECT_EQ("msg-id", receipt.message_id);
+}
+
+// Test 5: Isolating an endpoint and then checking isolation status
+TEST_F(ProducerImplTest, isolateEndpointAndCheckTest) {
+ const std::string endpoint = "127.0.0.1:19999";
+
+ EXPECT_FALSE(producer_->isEndpointIsolated(endpoint));
+
+ producer_->isolateEndpoint(endpoint);
+
+ EXPECT_TRUE(producer_->isEndpointIsolated(endpoint));
+ EXPECT_FALSE(producer_->isEndpointIsolated("10.0.0.1:19999"));
+}
+
+// Test 6: Sending when producer is not in STARTED state should return
IllegalState
+TEST_F(ProducerImplTest, sendSyncWhenNotRunningReturnsIllegalStateTest) {
+ // Reset state to CREATED (not STARTED)
+ producer_->state(State::CREATED);
+
+ auto msg = makeMessage();
+ std::error_code ec;
+ producer_->send(std::move(msg), ec);
+
+ EXPECT_EQ(ErrorCode::IllegalState, ec);
+}
+
+// Test 7: Transactional send should reject FIFO messages (with group set)
+TEST_F(ProducerImplTest, transactionalSendRejectsFifoMessageTest) {
+ auto transaction = producer_->beginTransaction();
+
+ // Build a message with group set (FIFO message)
+ auto msg = Message::newBuilder()
+ .withTopic("test-topic")
+ .withBody("test-body")
+ .withGroup("test-group")
+ .build();
+
+ std::error_code ec;
+ producer_->send(std::move(msg), ec, *transaction);
+
+ EXPECT_EQ(ErrorCode::MessagePropertyConflictWithType, ec);
+}
+
+// Test 8: prepareHeartbeatData should set client_type to PRODUCER
+TEST_F(ProducerImplTest, prepareHeartbeatDataSetsProducerTypeTest) {
+ HeartbeatRequest request;
+ producer_->prepareHeartbeatData(request);
+
+ EXPECT_EQ(rmq::ClientType::PRODUCER, request.client_type());
+}
+
+// Test 9: topicsOfInterest should return topics configured via withTopics
+TEST_F(ProducerImplTest, topicsOfInterestReturnsConfiguredTopicsTest) {
+ std::vector<std::string> input_topics = {"topic-a", "topic-b", "topic-c"};
+ producer_->withTopics(input_topics);
+
+ std::vector<std::string> result_topics;
+ producer_->topicsOfInterest(result_topics);
+
+ ASSERT_EQ(3u, result_topics.size());
+ EXPECT_EQ("topic-a", result_topics[0]);
+ EXPECT_EQ("topic-b", result_topics[1]);
+ EXPECT_EQ("topic-c", result_topics[2]);
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
b/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
new file mode 100644
index 00000000..5a3831bc
--- /dev/null
+++ b/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
@@ -0,0 +1,337 @@
+/*
+ * 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.
+ */
+#include <algorithm>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "ClientManagerMock.h"
+#include "MixAll.h"
+#include "Protocol.h"
+#include "PushConsumerImpl.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "rocketmq/ConsumeResult.h"
+#include "rocketmq/ExpressionType.h"
+#include "rocketmq/FilterExpression.h"
+#include "rocketmq/Message.h"
+#include "rocketmq/MessageListener.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+std::shared_ptr<PushConsumerImpl> createConsumer(const std::string& group =
"test-group") {
+ auto consumer = std::make_shared<PushConsumerImpl>(group);
+ auto client_manager =
std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ consumer->clientManager(client_manager);
+ return consumer;
+}
+
+} // namespace
+
+TEST(PushConsumerImplTest, subscribeTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("test-topic", "tagA", ExpressionType::TAG);
+
+ auto filter = consumer->getFilterExpression("test-topic");
+ ASSERT_TRUE(filter.has_value());
+ EXPECT_EQ("tagA", filter->content_);
+ EXPECT_EQ(ExpressionType::TAG, filter->type_);
+}
+
+TEST(PushConsumerImplTest, subscribeWithSQL92Test) {
+ auto consumer = createConsumer();
+ consumer->subscribe("sql-topic", "color = 'red'", ExpressionType::SQL92);
+
+ auto filter = consumer->getFilterExpression("sql-topic");
+ ASSERT_TRUE(filter.has_value());
+ EXPECT_EQ("color = 'red'", filter->content_);
+ EXPECT_EQ(ExpressionType::SQL92, filter->type_);
+}
+
+TEST(PushConsumerImplTest, subscribeDefaultExpressionTypeTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("default-type-topic", "*");
+
+ auto filter = consumer->getFilterExpression("default-type-topic");
+ ASSERT_TRUE(filter.has_value());
+ EXPECT_EQ("*", filter->content_);
+ EXPECT_EQ(ExpressionType::TAG, filter->type_);
+}
+
+TEST(PushConsumerImplTest, subscribeEmptyTagBecomesWildcardTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("wildcard-topic", "", ExpressionType::TAG);
+
+ auto filter = consumer->getFilterExpression("wildcard-topic");
+ ASSERT_TRUE(filter.has_value());
+ EXPECT_EQ(FilterExpression::WILD_CARD_TAG, filter->content_);
+}
+
+TEST(PushConsumerImplTest, unsubscribeTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("topic-to-remove", "tagA");
+
+ auto filter = consumer->getFilterExpression("topic-to-remove");
+ ASSERT_TRUE(filter.has_value());
+
+ consumer->unsubscribe("topic-to-remove");
+
+ filter = consumer->getFilterExpression("topic-to-remove");
+ EXPECT_FALSE(filter.has_value());
+}
+
+TEST(PushConsumerImplTest, unsubscribeNonExistentTopicTest) {
+ auto consumer = createConsumer();
+ // Should not throw or crash
+ consumer->unsubscribe("non-existent-topic");
+
+ auto filter = consumer->getFilterExpression("non-existent-topic");
+ EXPECT_FALSE(filter.has_value());
+}
+
+TEST(PushConsumerImplTest, subscribeMultipleTopicsTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("topic-1", "tagA");
+ consumer->subscribe("topic-2", "tagB");
+ consumer->subscribe("topic-3", "tagC");
+
+ auto filter1 = consumer->getFilterExpression("topic-1");
+ auto filter2 = consumer->getFilterExpression("topic-2");
+ auto filter3 = consumer->getFilterExpression("topic-3");
+
+ ASSERT_TRUE(filter1.has_value());
+ ASSERT_TRUE(filter2.has_value());
+ ASSERT_TRUE(filter3.has_value());
+
+ EXPECT_EQ("tagA", filter1->content_);
+ EXPECT_EQ("tagB", filter2->content_);
+ EXPECT_EQ("tagC", filter3->content_);
+}
+
+TEST(PushConsumerImplTest, subscribeOverwritesExistingFilterTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("topic", "tagA");
+ consumer->subscribe("topic", "tagB");
+
+ auto filter = consumer->getFilterExpression("topic");
+ ASSERT_TRUE(filter.has_value());
+ // emplace does not overwrite if key already exists
+ // The first subscription should remain
+ EXPECT_EQ("tagA", filter->content_);
+}
+
+TEST(PushConsumerImplTest, topicsOfInterestTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("topic-alpha", "tagA");
+ consumer->subscribe("topic-beta", "tagB");
+ consumer->subscribe("topic-gamma", "tagC");
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+
+ EXPECT_EQ(3u, topics.size());
+ EXPECT_NE(topics.end(), std::find(topics.begin(), topics.end(),
"topic-alpha"));
+ EXPECT_NE(topics.end(), std::find(topics.begin(), topics.end(),
"topic-beta"));
+ EXPECT_NE(topics.end(), std::find(topics.begin(), topics.end(),
"topic-gamma"));
+}
+
+TEST(PushConsumerImplTest, topicsOfInterestAfterUnsubscribeTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("topic-keep", "tagA");
+ consumer->subscribe("topic-remove", "tagB");
+
+ consumer->unsubscribe("topic-remove");
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+
+ EXPECT_EQ(1u, topics.size());
+ EXPECT_EQ("topic-keep", topics[0]);
+}
+
+TEST(PushConsumerImplTest, topicsOfInterestEmptyTest) {
+ auto consumer = createConsumer();
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+ EXPECT_TRUE(topics.empty());
+}
+
+TEST(PushConsumerImplTest, getFilterExpressionForNonExistentTopicTest) {
+ auto consumer = createConsumer();
+ auto filter = consumer->getFilterExpression("unknown-topic");
+ EXPECT_FALSE(filter.has_value());
+}
+
+TEST(PushConsumerImplTest, prepareHeartbeatDataTest) {
+ auto consumer = createConsumer();
+ HeartbeatRequest request;
+ consumer->prepareHeartbeatData(request);
+
+ EXPECT_EQ(rmq::ClientType::PUSH_CONSUMER, request.client_type());
+ EXPECT_EQ("test-group", request.group().name());
+}
+
+TEST(PushConsumerImplTest, buildClientSettingsTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("settings-topic", "tagA", ExpressionType::TAG);
+
+ rmq::Settings settings;
+ consumer->buildClientSettings(settings);
+
+ EXPECT_EQ(rmq::ClientType::PUSH_CONSUMER, settings.client_type());
+ EXPECT_EQ("test-group", settings.subscription().group().name());
+ ASSERT_EQ(1, settings.subscription().subscriptions_size());
+ EXPECT_EQ("settings-topic",
settings.subscription().subscriptions(0).topic().name());
+ EXPECT_EQ("tagA",
settings.subscription().subscriptions(0).expression().expression());
+ EXPECT_EQ(rmq::FilterType::TAG,
settings.subscription().subscriptions(0).expression().type());
+}
+
+TEST(PushConsumerImplTest, buildClientSettingsSQL92FilterTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("sql-settings-topic", "price > 100",
ExpressionType::SQL92);
+
+ rmq::Settings settings;
+ consumer->buildClientSettings(settings);
+
+ EXPECT_EQ(rmq::ClientType::PUSH_CONSUMER, settings.client_type());
+ ASSERT_EQ(1, settings.subscription().subscriptions_size());
+ EXPECT_EQ(rmq::FilterType::SQL,
settings.subscription().subscriptions(0).expression().type());
+ EXPECT_EQ("price > 100",
settings.subscription().subscriptions(0).expression().expression());
+}
+
+TEST(PushConsumerImplTest, buildClientSettingsMultipleSubscriptionsTest) {
+ auto consumer = createConsumer();
+ consumer->subscribe("topic-a", "tagA", ExpressionType::TAG);
+ consumer->subscribe("topic-b", "color = 'blue'", ExpressionType::SQL92);
+
+ rmq::Settings settings;
+ consumer->buildClientSettings(settings);
+
+ EXPECT_EQ(2, settings.subscription().subscriptions_size());
+}
+
+TEST(PushConsumerImplTest, maxDeliveryAttemptsDefaultTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ(MixAll::DEFAULT_MAX_DELIVERY_ATTEMPTS,
consumer->maxDeliveryAttempts());
+}
+
+TEST(PushConsumerImplTest, receiveBatchSizeDefaultTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ(MixAll::DEFAULT_RECEIVE_MESSAGE_BATCH_SIZE,
consumer->receiveBatchSize());
+}
+
+TEST(PushConsumerImplTest, consumeThreadPoolSizeDefaultTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ(MixAll::DEFAULT_CONSUME_THREAD_POOL_SIZE,
consumer->consumeThreadPoolSize());
+}
+
+TEST(PushConsumerImplTest, consumeThreadPoolSizeSetterTest) {
+ auto consumer = createConsumer();
+ consumer->consumeThreadPoolSize(10);
+ EXPECT_EQ(10u, consumer->consumeThreadPoolSize());
+}
+
+TEST(PushConsumerImplTest, consumeThreadPoolSizeSetterMinimumTest) {
+ auto consumer = createConsumer();
+ consumer->consumeThreadPoolSize(1);
+ EXPECT_EQ(1u, consumer->consumeThreadPoolSize());
+}
+
+TEST(PushConsumerImplTest, consumeThreadPoolSizeSetterRejectsZeroTest) {
+ auto consumer = createConsumer();
+ uint32_t original = consumer->consumeThreadPoolSize();
+ consumer->consumeThreadPoolSize(0);
+ // 0 is less than 1, so the value should remain unchanged
+ EXPECT_EQ(original, consumer->consumeThreadPoolSize());
+}
+
+TEST(PushConsumerImplTest, consumeThreadPoolSizeSetterRejectsNegativeTest) {
+ auto consumer = createConsumer();
+ uint32_t original = consumer->consumeThreadPoolSize();
+ consumer->consumeThreadPoolSize(-1);
+ // -1 is less than 1, so the value should remain unchanged
+ EXPECT_EQ(original, consumer->consumeThreadPoolSize());
+}
+
+TEST(PushConsumerImplTest, registerMessageListenerTest) {
+ auto consumer = createConsumer();
+ bool listener_called = false;
+ auto listener = [&listener_called](const Message&) {
+ listener_called = true;
+ return ConsumeResult::SUCCESS;
+ };
+
+ consumer->registerMessageListener(listener);
+
+ // Verify the listener is stored by invoking it through the accessor
+ auto& stored_listener = consumer->messageListener();
+ ASSERT_TRUE(static_cast<bool>(stored_listener));
+
+ auto msg = Message::newBuilder().withTopic("test-topic").build();
+ stored_listener(*msg);
+ EXPECT_TRUE(listener_called);
+}
+
+TEST(PushConsumerImplTest, messageListenerReturnsFailureTest) {
+ auto consumer = createConsumer();
+ auto listener = [](const Message&) { return ConsumeResult::FAILURE; };
+
+ consumer->registerMessageListener(listener);
+
+ auto& stored_listener = consumer->messageListener();
+ ASSERT_TRUE(static_cast<bool>(stored_listener));
+
+ auto msg = Message::newBuilder().withTopic("test-topic").build();
+ EXPECT_EQ(ConsumeResult::FAILURE, stored_listener(*msg));
+}
+
+TEST(PushConsumerImplTest, messageListenerEmptyByDefaultTest) {
+ auto consumer = createConsumer();
+ auto& listener = consumer->messageListener();
+ // Default-constructed std::function should be empty
+ EXPECT_FALSE(static_cast<bool>(listener));
+}
+
+TEST(PushConsumerImplTest, getProcessQueueTableSizeEmptyTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ(0u, consumer->getProcessQueueTableSize());
+}
+
+TEST(PushConsumerImplTest, groupNameTest) {
+ auto consumer = createConsumer("my-consumer-group");
+ EXPECT_EQ("my-consumer-group", consumer->groupName());
+}
+
+TEST(PushConsumerImplTest, groupNameDefaultTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ("test-group", consumer->groupName());
+}
+
+TEST(PushConsumerImplTest, maxCachedMessageQuantityTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ(MixAll::DEFAULT_CACHED_MESSAGE_COUNT,
consumer->maxCachedMessageQuantity());
+}
+
+TEST(PushConsumerImplTest, maxCachedMessageMemoryTest) {
+ auto consumer = createConsumer();
+ EXPECT_EQ(MixAll::DEFAULT_CACHED_MESSAGE_MEMORY,
consumer->maxCachedMessageMemory());
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/SendContextTest.cpp
b/cpp/source/rocketmq/tests/SendContextTest.cpp
index e03c8b7d..98161e91 100644
--- a/cpp/source/rocketmq/tests/SendContextTest.cpp
+++ b/cpp/source/rocketmq/tests/SendContextTest.cpp
@@ -23,7 +23,7 @@ ROCKETMQ_NAMESPACE_BEGIN
TEST(SendContextTest, testBasics) {
auto message =
Message::newBuilder().withBody("body").withTopic("topic").withGroup("group0").withTag("TagA").build();
- auto callback = [](const std::error_code&, const SendReceipt&) {};
+ auto callback = [](const std::error_code&, SendReceipt&&) {};
std::weak_ptr<ProducerImpl> producer;
std::vector<rmq::MessageQueue> message_queues;
auto send_context = std::make_shared<SendContext>(producer,
std::move(message), callback, message_queues);
diff --git a/cpp/source/rocketmq/tests/SimpleConsumerImplTest.cpp
b/cpp/source/rocketmq/tests/SimpleConsumerImplTest.cpp
new file mode 100644
index 00000000..e26da42b
--- /dev/null
+++ b/cpp/source/rocketmq/tests/SimpleConsumerImplTest.cpp
@@ -0,0 +1,195 @@
+/*
+ * 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.
+ */
+#include <memory>
+#include <string>
+#include <system_error>
+#include <vector>
+
+#include "ClientManagerMock.h"
+#include "NameServerResolverMock.h"
+#include "Protocol.h"
+#include "SimpleConsumerImpl.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "rocketmq/ErrorCode.h"
+#include "rocketmq/FilterExpression.h"
+#include "rocketmq/Message.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+/// Test subclass that exposes protected members for unit testing.
+class TestableSimpleConsumerImpl : public SimpleConsumerImpl {
+public:
+ explicit TestableSimpleConsumerImpl(const std::string& group) :
ClientImpl(group), SimpleConsumerImpl(group) {}
+
+ using SimpleConsumerImpl::topicsOfInterest;
+};
+
+} // namespace
+
+TEST(SimpleConsumerImplTest, constructorWithGroupTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ EXPECT_EQ(consumer->config().subscriber.group.name(), "test-group");
+}
+
+TEST(SimpleConsumerImplTest, subscribeAddsTopicTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ consumer->subscribe("test-topic", FilterExpression("*"));
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+ ASSERT_EQ(1u, topics.size());
+ EXPECT_EQ("test-topic", topics[0]);
+}
+
+TEST(SimpleConsumerImplTest, unsubscribeRemovesTopicTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ consumer->subscribe("test-topic", FilterExpression("*"));
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+ ASSERT_EQ(1u, topics.size());
+
+ consumer->unsubscribe("test-topic");
+ topics.clear();
+ consumer->topicsOfInterest(topics);
+ EXPECT_TRUE(topics.empty());
+}
+
+TEST(SimpleConsumerImplTest, subscribeMultipleTopicsTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ consumer->subscribe("topic-1", FilterExpression("*"));
+ consumer->subscribe("topic-2", FilterExpression("tagA"));
+ consumer->subscribe("topic-3", FilterExpression("tagB"));
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+ EXPECT_EQ(3u, topics.size());
+}
+
+TEST(SimpleConsumerImplTest, subscribeUpdatesFilterExpressionTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ consumer->subscribe("test-topic", FilterExpression("tagA"));
+ // Overwrite with a new filter expression
+ consumer->subscribe("test-topic", FilterExpression("tagB"));
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+ // Should still be a single topic (no duplicate)
+ ASSERT_EQ(1u, topics.size());
+ EXPECT_EQ("test-topic", topics[0]);
+}
+
+TEST(SimpleConsumerImplTest, prepareHeartbeatDataSetsSimpleConsumerTypeTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ rmq::HeartbeatRequest request;
+ consumer->prepareHeartbeatData(request);
+ EXPECT_EQ(rmq::ClientType::SIMPLE_CONSUMER, request.client_type());
+}
+
+TEST(SimpleConsumerImplTest, buildClientSettingsSetsSimpleConsumerTypeTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ auto resolver =
std::make_shared<testing::NiceMock<NameServerResolverMock>>();
+ ON_CALL(*resolver,
resolve()).WillByDefault(testing::Return(std::string("ipv4:10.0.0.1:8080")));
+ consumer->withNameServerResolver(resolver);
+ consumer->subscribe("settings-topic", FilterExpression("tagX"));
+
+ rmq::Settings settings;
+ consumer->buildClientSettings(settings);
+ EXPECT_EQ(rmq::ClientType::SIMPLE_CONSUMER, settings.client_type());
+}
+
+TEST(SimpleConsumerImplTest, unsubscribeNonexistentTopicDoesNotCrashTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ // Should not crash or throw
+ consumer->unsubscribe("nonexistent-topic");
+
+ std::vector<std::string> topics;
+ consumer->topicsOfInterest(topics);
+ EXPECT_TRUE(topics.empty());
+}
+
+TEST(SimpleConsumerImplTest, topicsOfInterestNoDuplicatesTest) {
+ auto consumer = std::make_shared<TestableSimpleConsumerImpl>("test-group");
+ consumer->subscribe("dup-topic", FilterExpression("*"));
+
+ // Pre-populate the vector — topicsOfInterest should not add a duplicate
+ std::vector<std::string> topics;
+ topics.push_back("dup-topic");
+ consumer->topicsOfInterest(topics);
+ EXPECT_EQ(1u, topics.size());
+}
+
+TEST(SimpleConsumerImplTest, ackWhenNotRunningFailsTest) {
+ auto consumer = std::make_shared<SimpleConsumerImpl>("test-group");
+ // State is CREATED by default (not STARTED) — the consumer was never
started.
+ // The manager is not configured, so ack cannot proceed.
+ // Verify the initial state is indeed CREATED.
+ EXPECT_FALSE(consumer->active());
+}
+
+TEST(SimpleConsumerImplTest, ackWithMockedManagerPropagatesErrorCodeTest) {
+ auto consumer = std::make_shared<SimpleConsumerImpl>("test-group");
+ consumer->state(State::STARTED);
+
+ auto client_manager =
std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ consumer->clientManager(client_manager);
+
+ // Mock ack to invoke callback with an error
+ ON_CALL(*client_manager, ack)
+ .WillByDefault(testing::Invoke(
+ [](const std::string&, const Metadata&, const AckMessageRequest&,
std::chrono::milliseconds,
+ const std::function<void(const std::error_code&)>& cb) {
+ cb(ErrorCode::NotFound);
+ }));
+
+ MessageConstPtr msg =
Message::newBuilder().withTopic("ack-topic").withBody("body").build();
+ std::error_code ec;
+ consumer->ack(*msg, ec);
+
+ EXPECT_EQ(ErrorCode::NotFound, ec);
+
+ consumer->state(State::STOPPED);
+}
+
+TEST(SimpleConsumerImplTest, ackWithMockedManagerSuccessTest) {
+ auto consumer = std::make_shared<SimpleConsumerImpl>("test-group");
+ consumer->state(State::STARTED);
+
+ auto client_manager =
std::make_shared<testing::NiceMock<ClientManagerMock>>();
+ consumer->clientManager(client_manager);
+
+ // Mock ack to invoke callback with success
+ ON_CALL(*client_manager, ack)
+ .WillByDefault(testing::Invoke(
+ [](const std::string&, const Metadata&, const AckMessageRequest&,
std::chrono::milliseconds,
+ const std::function<void(const std::error_code&)>& cb) {
+ cb(std::error_code());
+ }));
+
+ MessageConstPtr msg =
Message::newBuilder().withTopic("ack-topic").withBody("body").build();
+ std::error_code ec;
+ consumer->ack(*msg, ec);
+
+ EXPECT_FALSE(ec);
+
+ consumer->state(State::STOPPED);
+}
+
+ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/tests/TopicPublishInfoTest.cpp
b/cpp/source/rocketmq/tests/TopicPublishInfoTest.cpp
new file mode 100644
index 00000000..1bb333bc
--- /dev/null
+++ b/cpp/source/rocketmq/tests/TopicPublishInfoTest.cpp
@@ -0,0 +1,152 @@
+/*
+ * 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.
+ */
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "MixAll.h"
+#include "ProducerImpl.h"
+#include "Protocol.h"
+#include "TopicPublishInfo.h"
+#include "TopicRouteData.h"
+#include "absl/types/optional.h"
+#include "gtest/gtest.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+namespace {
+
+rmq::MessageQueue createMessageQueue(const std::string& topic, int id, const
std::string& broker_name,
+ int32_t broker_id, rmq::Permission perm) {
+ rmq::MessageQueue mq;
+ mq.mutable_topic()->set_name(topic);
+ mq.set_id(id);
+ mq.mutable_broker()->set_name(broker_name);
+ mq.mutable_broker()->set_id(broker_id);
+
mq.mutable_broker()->mutable_endpoints()->set_scheme(rmq::AddressScheme::IPv4);
+ auto* addr = mq.mutable_broker()->mutable_endpoints()->add_addresses();
+ addr->set_host("10.0.0." + std::to_string(broker_id + 1));
+ addr->set_port(8080);
+ mq.set_permission(perm);
+ return mq;
+}
+
+} // namespace
+
+class TopicPublishInfoTest : public testing::Test {
+protected:
+ void SetUp() override {
+ producer_ = std::make_shared<ProducerImpl>();
+ producer_->state(State::STARTED);
+ producer_->maxAttemptTimes(3);
+
+ // 4 writable queues across 2 brokers
+ std::vector<rmq::MessageQueue> queues;
+ for (int i = 0; i < 4; i++) {
+ queues.push_back(createMessageQueue("test-topic", i, "broker-" +
std::to_string(i % 2),
+ MixAll::MASTER_BROKER_ID,
rmq::Permission::READ_WRITE));
+ }
+ route_data_ = std::make_shared<TopicRouteData>(queues);
+ }
+
+ std::shared_ptr<ProducerImpl> producer_;
+ TopicRouteDataPtr route_data_;
+};
+
+TEST_F(TopicPublishInfoTest, selectMessageQueuesReturnsResultsTest) {
+ TopicPublishInfo info(producer_, "test-topic", route_data_);
+ std::vector<rmq::MessageQueue> result;
+ EXPECT_TRUE(info.selectMessageQueues(absl::nullopt, result));
+ EXPECT_FALSE(result.empty());
+}
+
+TEST_F(TopicPublishInfoTest, selectMessageQueuesWithMessageGroupTest) {
+ TopicPublishInfo info(producer_, "test-topic", route_data_);
+
+ // Same message group should always select the same queue
+ std::vector<rmq::MessageQueue> result1, result2;
+
EXPECT_TRUE(info.selectMessageQueues(absl::make_optional<std::string>("group-A"),
result1));
+
EXPECT_TRUE(info.selectMessageQueues(absl::make_optional<std::string>("group-A"),
result2));
+ ASSERT_EQ(1u, result1.size());
+ ASSERT_EQ(1u, result2.size());
+ EXPECT_EQ(result1[0].id(), result2[0].id());
+}
+
+TEST_F(TopicPublishInfoTest, selectMessageQueuesSkipsNonWritableTest) {
+ // Build route with queue 0 set to READ-only
+ std::vector<rmq::MessageQueue> queues;
+ for (int i = 0; i < 4; i++) {
+ rmq::Permission perm = (i == 0) ? rmq::Permission::READ :
rmq::Permission::READ_WRITE;
+ queues.push_back(createMessageQueue("test-topic", i, "broker-" +
std::to_string(i % 2),
+ MixAll::MASTER_BROKER_ID, perm));
+ }
+ auto route = std::make_shared<TopicRouteData>(queues);
+ TopicPublishInfo info(producer_, "test-topic", route);
+
+ auto filtered = info.getMessageQueueList();
+ for (const auto& q : filtered) {
+ EXPECT_TRUE(writable(q.permission()));
+ }
+ // Only 3 writable queues should remain
+ EXPECT_EQ(3u, filtered.size());
+}
+
+TEST_F(TopicPublishInfoTest, selectMessageQueuesSkipsNonMasterBrokerTest) {
+ // Build route with queue 0 on a non-master broker
+ std::vector<rmq::MessageQueue> queues;
+ for (int i = 0; i < 4; i++) {
+ int32_t broker_id = (i == 0) ? 1 : MixAll::MASTER_BROKER_ID;
+ queues.push_back(createMessageQueue("test-topic", i, "broker-" +
std::to_string(i % 2),
+ broker_id,
rmq::Permission::READ_WRITE));
+ }
+ auto route = std::make_shared<TopicRouteData>(queues);
+ TopicPublishInfo info(producer_, "test-topic", route);
+
+ auto filtered = info.getMessageQueueList();
+ for (const auto& q : filtered) {
+ EXPECT_EQ(MixAll::MASTER_BROKER_ID, q.broker().id());
+ }
+ // Queue 0 is on a non-master broker, so only 3 queues should remain
+ EXPECT_EQ(3u, filtered.size());
+}
+
+TEST_F(TopicPublishInfoTest, selectMessageQueuesEmptyRouteReturnsFalseTest) {
+ std::vector<rmq::MessageQueue> empty_queues;
+ auto empty_route = std::make_shared<TopicRouteData>(empty_queues);
+ TopicPublishInfo info(producer_, "test-topic", empty_route);
+
+ std::vector<rmq::MessageQueue> result;
+ EXPECT_FALSE(info.selectMessageQueues(absl::nullopt, result));
+}
+
+TEST_F(TopicPublishInfoTest, updateRouteDataRefreshesQueuesTest) {
+ TopicPublishInfo info(producer_, "test-topic", route_data_);
+ EXPECT_EQ(4u, info.getMessageQueueList().size());
+
+ // New route with only 2 queues
+ std::vector<rmq::MessageQueue> new_queues;
+ for (int i = 0; i < 2; i++) {
+ new_queues.push_back(createMessageQueue("test-topic", i, "broker-" +
std::to_string(i),
+ MixAll::MASTER_BROKER_ID,
rmq::Permission::READ_WRITE));
+ }
+ auto new_route = std::make_shared<TopicRouteData>(new_queues);
+ info.topicRouteData(new_route);
+ EXPECT_EQ(2u, info.getMessageQueueList().size());
+}
+
+ROCKETMQ_NAMESPACE_END
