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 41b65fa4 [C++] Update public API and simplify client lifecycle state
machine (#1300)
41b65fa4 is described below
commit 41b65fa4dc8a59d6caa8f36e3ed3edaa1e91b627
Author: lizhimins <[email protected]>
AuthorDate: Fri Jul 10 11:01:29 2026 +0800
[C++] Update public API and simplify client lifecycle state machine (#1300)
---
cpp/examples/ExampleFifoProducer.cpp | 2 +-
cpp/examples/ExampleProducerWithAsync.cpp | 2 +-
cpp/include/rocketmq/FifoProducer.h | 4 +-
cpp/include/rocketmq/Producer.h | 10 +-
cpp/include/rocketmq/PushConsumer.h | 5 +-
cpp/include/rocketmq/SendCallback.h | 2 +-
cpp/include/rocketmq/SimpleConsumer.h | 16 +--
cpp/source/client/ReceiveMessageStreamReader.cpp | 4 +
cpp/source/client/RpcClientImpl.cpp | 6 +-
cpp/source/client/include/RpcClientImpl.h | 3 +-
.../client/mocks/include/ClientManagerMock.h | 16 ++-
cpp/source/rocketmq/ClientImpl.cpp | 96 +++++++++----
cpp/source/rocketmq/FifoProducer.cpp | 17 ++-
cpp/source/rocketmq/FifoProducerPartition.cpp | 50 ++++---
cpp/source/rocketmq/Producer.cpp | 81 +++++++----
cpp/source/rocketmq/ProducerImpl.cpp | 67 +++++----
cpp/source/rocketmq/PushConsumer.cpp | 18 ++-
cpp/source/rocketmq/PushConsumerImpl.cpp | 63 +++++----
cpp/source/rocketmq/SendContext.cpp | 15 ++-
cpp/source/rocketmq/SimpleConsumer.cpp | 149 ++++++++++++++-------
cpp/source/rocketmq/SimpleConsumerImpl.cpp | 65 ++++-----
cpp/source/rocketmq/include/ClientImpl.h | 7 +-
.../rocketmq/include/FifoProducerPartition.h | 2 +-
cpp/source/rocketmq/include/ProducerImpl.h | 4 +-
cpp/source/rocketmq/include/PushConsumerImpl.h | 2 +-
cpp/source/rocketmq/include/SimpleConsumerImpl.h | 5 +-
26 files changed, 442 insertions(+), 269 deletions(-)
diff --git a/cpp/examples/ExampleFifoProducer.cpp
b/cpp/examples/ExampleFifoProducer.cpp
index 1876ebb1..f9e6974d 100644
--- a/cpp/examples/ExampleFifoProducer.cpp
+++ b/cpp/examples/ExampleFifoProducer.cpp
@@ -160,7 +160,7 @@ int main(int argc, char* argv[]) {
.withBody(body)
.build();
std::error_code ec;
- auto callback = [&](const std::error_code& ec, const SendReceipt&
receipt) mutable {
+ auto callback = [&](const std::error_code& ec, SendReceipt&& receipt)
mutable {
completed++;
count++;
semaphore->release();
diff --git a/cpp/examples/ExampleProducerWithAsync.cpp
b/cpp/examples/ExampleProducerWithAsync.cpp
index a46fdf43..e950e48f 100644
--- a/cpp/examples/ExampleProducerWithAsync.cpp
+++ b/cpp/examples/ExampleProducerWithAsync.cpp
@@ -145,7 +145,7 @@ int main(int argc, char* argv[]) {
std::unique_ptr<Semaphore> semaphore(new Semaphore(FLAGS_concurrency));
- auto send_callback = [&](const std::error_code& ec, const SendReceipt&
receipt) {
+ auto send_callback = [&](const std::error_code& ec, SendReceipt&& receipt) {
std::unique_lock<std::mutex> lk(mtx);
semaphore->release();
completed++;
diff --git a/cpp/include/rocketmq/FifoProducer.h
b/cpp/include/rocketmq/FifoProducer.h
index 1ee20d91..af022693 100644
--- a/cpp/include/rocketmq/FifoProducer.h
+++ b/cpp/include/rocketmq/FifoProducer.h
@@ -35,7 +35,7 @@ class FifoProducer {
public:
static FifoProducerBuilder newBuilder();
- void send(MessageConstPtr message, SendCallback callback);
+ void send(MessageConstPtr message, SendCallback callback) noexcept;
private:
std::shared_ptr<FifoProducerImpl> impl_;
@@ -54,7 +54,7 @@ public:
FifoProducerBuilder& withConfiguration(Configuration configuration);
- FifoProducerBuilder& withTopics(const std::vector<std::string>& topics);
+ FifoProducerBuilder& withTopics(std::vector<std::string> topics);
FifoProducerBuilder& withConcurrency(std::size_t concurrency);
diff --git a/cpp/include/rocketmq/Producer.h b/cpp/include/rocketmq/Producer.h
index e1c093c3..2218c5b7 100644
--- a/cpp/include/rocketmq/Producer.h
+++ b/cpp/include/rocketmq/Producer.h
@@ -62,11 +62,11 @@ public:
* @param message Message to send. Note the pointer, std::unique_ptr<const
Message>, is 'moved' into.
* @param callback Callback to execute on completion.
*/
- void send(MessageConstPtr message, const SendCallback& callback) noexcept;
+ void send(MessageConstPtr message, SendCallback callback) noexcept;
std::unique_ptr<Transaction> beginTransaction();
- SendReceipt send(MessageConstPtr message, std::error_code& ec, Transaction&
transaction);
+ SendReceipt send(MessageConstPtr message, std::error_code& ec, Transaction&
transaction) noexcept;
/**
* @brief Attempts to cancel a scheduled message based on the provided topic
and recall handle.
@@ -94,11 +94,11 @@ class ProducerBuilder {
public:
ProducerBuilder();
- ProducerBuilder& withConfiguration(const Configuration& configuration);
+ ProducerBuilder& withConfiguration(Configuration configuration);
- ProducerBuilder& withTopics(const std::vector<std::string>& topics);
+ ProducerBuilder& withTopics(std::vector<std::string> topics);
- ProducerBuilder& withTransactionChecker(const TransactionChecker& checker);
+ ProducerBuilder& withTransactionChecker(TransactionChecker checker);
Producer build();
diff --git a/cpp/include/rocketmq/PushConsumer.h
b/cpp/include/rocketmq/PushConsumer.h
index a989997c..d51c7e8c 100644
--- a/cpp/include/rocketmq/PushConsumer.h
+++ b/cpp/include/rocketmq/PushConsumer.h
@@ -18,6 +18,7 @@
#include <memory>
#include <string>
+#include <unordered_map>
#include "Configuration.h"
#include "CredentialsProvider.h"
@@ -36,9 +37,9 @@ class PushConsumer {
public:
static PushConsumerBuilder newBuilder();
- void subscribe(std::string topic, FilterExpression filter_expression);
+ void subscribe(std::string topic, FilterExpression filter_expression)
noexcept;
- void unsubscribe(const std::string& topic);
+ void unsubscribe(const std::string& topic) noexcept;
private:
friend class PushConsumerBuilder;
diff --git a/cpp/include/rocketmq/SendCallback.h
b/cpp/include/rocketmq/SendCallback.h
index 8468f635..88fe1ad6 100644
--- a/cpp/include/rocketmq/SendCallback.h
+++ b/cpp/include/rocketmq/SendCallback.h
@@ -24,6 +24,6 @@
ROCKETMQ_NAMESPACE_BEGIN
-using SendCallback = std::function<void(const std::error_code&, const
SendReceipt&)>;
+using SendCallback = std::function<void(const std::error_code&,
SendReceipt&&)>;
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/cpp/include/rocketmq/SimpleConsumer.h
b/cpp/include/rocketmq/SimpleConsumer.h
index dff1158a..f97749a1 100644
--- a/cpp/include/rocketmq/SimpleConsumer.h
+++ b/cpp/include/rocketmq/SimpleConsumer.h
@@ -45,26 +45,26 @@ class SimpleConsumer {
public:
static SimpleConsumerBuilder newBuilder();
- void subscribe(std::string topic, FilterExpression filter_expression);
+ void subscribe(std::string topic, FilterExpression filter_expression)
noexcept;
- void unsubscribe(const std::string& topic);
+ void unsubscribe(const std::string& topic) noexcept;
void receive(std::size_t limit,
std::chrono::milliseconds invisible_duration,
std::error_code& ec,
- std::vector<MessageConstSharedPtr>& messages);
+ std::vector<MessageConstSharedPtr>& messages) noexcept;
- void asyncReceive(std::size_t limit, std::chrono::milliseconds
invisible_duration, ReceiveCallback callback);
+ void asyncReceive(std::size_t limit, std::chrono::milliseconds
invisible_duration, ReceiveCallback callback) noexcept;
- void ack(const Message& message, std::error_code& ec);
+ void ack(const Message& message, std::error_code& ec) noexcept;
- void asyncAck(const Message& message, AckCallback callback);
+ void asyncAck(const Message& message, AckCallback callback) noexcept;
- void changeInvisibleDuration(const Message& message, std::string&
receipt_handle, std::chrono::milliseconds duration, std::error_code& ec);
+ void changeInvisibleDuration(const Message& message, std::string&
receipt_handle, std::chrono::milliseconds duration, std::error_code& ec)
noexcept;
void asyncChangeInvisibleDuration(const Message& message, std::string&
receipt_handle,
std::chrono::milliseconds duration,
- ChangeInvisibleDurationCallback callback);
+ ChangeInvisibleDurationCallback callback)
noexcept;
private:
std::shared_ptr<SimpleConsumerImpl> impl_;
diff --git a/cpp/source/client/ReceiveMessageStreamReader.cpp
b/cpp/source/client/ReceiveMessageStreamReader.cpp
index 999d49d9..68d7f913 100644
--- a/cpp/source/client/ReceiveMessageStreamReader.cpp
+++ b/cpp/source/client/ReceiveMessageStreamReader.cpp
@@ -139,6 +139,10 @@ void ReceiveMessageStreamReader::OnReadDone(bool ok) {
}
case rmq::ReceiveMessageResponse::ContentCase::kMessage: {
auto client_manager = client_manager_.lock();
+ if (!client_manager) {
+ SPDLOG_WARN("ClientManager has been destructed, dropping received
message");
+ break;
+ }
auto message = client_manager->wrapMessage(response_.message());
auto raw = const_cast<Message*>(message.get());
raw->mutableExtension().target_endpoint = peer_address_;
diff --git a/cpp/source/client/RpcClientImpl.cpp
b/cpp/source/client/RpcClientImpl.cpp
index 94e2c3ad..d84cb727 100644
--- a/cpp/source/client/RpcClientImpl.cpp
+++ b/cpp/source/client/RpcClientImpl.cpp
@@ -130,7 +130,11 @@ void RpcClientImpl::asyncRecallMessage(const
RecallMessageRequest& request,
}
bool RpcClientImpl::ok() const {
- return channel_ && grpc_connectivity_state::GRPC_CHANNEL_SHUTDOWN !=
channel_->GetState(false);
+ if (!channel_) {
+ return false;
+ }
+ auto state = channel_->GetState(false);
+ return state != grpc_connectivity_state::GRPC_CHANNEL_SHUTDOWN;
}
void RpcClientImpl::addMetadata(grpc::ClientContext& context,
diff --git a/cpp/source/client/include/RpcClientImpl.h
b/cpp/source/client/include/RpcClientImpl.h
index 9d4483e1..dafeb73e 100644
--- a/cpp/source/client/include/RpcClientImpl.h
+++ b/cpp/source/client/include/RpcClientImpl.h
@@ -16,6 +16,7 @@
*/
#pragma once
+#include <atomic>
#include <memory>
#include "Client.h"
@@ -97,7 +98,7 @@ private:
std::string peer_address_;
std::unique_ptr<rmq::MessagingService::Stub> stub_;
std::chrono::milliseconds connect_timeout_{3000};
- bool need_heartbeat_{true};
+ std::atomic<bool> need_heartbeat_{true};
};
using RpcClientSharedPtr = std::shared_ptr<RpcClient>;
diff --git a/cpp/source/client/mocks/include/ClientManagerMock.h
b/cpp/source/client/mocks/include/ClientManagerMock.h
index d255b901..a144b754 100644
--- a/cpp/source/client/mocks/include/ClientManagerMock.h
+++ b/cpp/source/client/mocks/include/ClientManagerMock.h
@@ -47,10 +47,9 @@ public:
(const std::function<void(const std::error_code&, const
HeartbeatResponse&)>&)),
(override));
- MOCK_METHOD(std::shared_ptr<TelemetryBidiReactor>, telemetry, (const
std::string&, std::weak_ptr<Client>),
- (override));
+ MOCK_METHOD(std::shared_ptr<RpcClient>, getRpcClient, (const std::string&,
bool), (override));
- MOCK_METHOD(bool, wrapMessage, (const rmq::Message&, MessageExt&),
(override));
+ MOCK_METHOD(MessageConstSharedPtr, wrapMessage, (const rmq::Message&),
(override));
MOCK_METHOD(void, ack,
(const std::string&, const Metadata&, const AckMessageRequest&,
std::chrono::milliseconds,
@@ -59,13 +58,13 @@ public:
MOCK_METHOD(void, changeInvisibleDuration,
(const std::string&, const Metadata&, const
ChangeInvisibleDurationRequest&, std::chrono::milliseconds,
- (const std::function<void(const std::error_code&)>&)),
+ (const std::function<void(const std::error_code&, const
ChangeInvisibleDurationResponse&)>&)),
(override));
MOCK_METHOD(void, forwardMessageToDeadLetterQueue,
(const std::string&, const Metadata&, const
ForwardMessageToDeadLetterQueueRequest&,
std::chrono::milliseconds,
- (const std::function<void(const
InvocationContext<ForwardMessageToDeadLetterQueueResponse>*)>&)),
+ (const std::function<void(const std::error_code&)>&)),
(override));
MOCK_METHOD(void, endTransaction,
@@ -73,6 +72,11 @@ public:
(const std::function<void(const std::error_code&, const
EndTransactionResponse&)>&)),
(override));
+ MOCK_METHOD(void, recallMessage,
+ (const std::string&, const Metadata&, const
RecallMessageRequest&, std::chrono::milliseconds,
+ (const std::function<void(const std::error_code&, const
RecallMessageResponse&)>&)),
+ (override));
+
MOCK_METHOD(void, addClientObserver, (std::weak_ptr<Client>), (override));
MOCK_METHOD(void, queryAssignment,
@@ -85,7 +89,7 @@ public:
ReceiveMessageCallback),
(override));
- MOCK_METHOD(bool, send, (const std::string&, const Metadata&,
SendMessageRequest&, SendCallback), (override));
+ MOCK_METHOD(bool, send, (const std::string&, const Metadata&,
SendMessageRequest&, SendResultCallback), (override));
MOCK_METHOD(std::error_code, notifyClientTermination,
(const std::string&, const Metadata&, const
NotifyClientTerminationRequest&, std::chrono::milliseconds),
diff --git a/cpp/source/rocketmq/ClientImpl.cpp
b/cpp/source/rocketmq/ClientImpl.cpp
index 169df412..6b59b820 100644
--- a/cpp/source/rocketmq/ClientImpl.cpp
+++ b/cpp/source/rocketmq/ClientImpl.cpp
@@ -93,17 +93,20 @@ rmq::Endpoints ClientImpl::accessPoint() {
}
void ClientImpl::start() {
- State expected = CREATED;
- if (!state_.compare_exchange_strong(expected, State::STARTING)) {
- SPDLOG_ERROR("Attempt to start ClientImpl failed. Expecting: {} Actual:
{}", State::CREATED,
- state_.load(std::memory_order_relaxed));
- return;
- }
+ // === Phase 1: Validation (before any resource allocation) ===
+ // Safe to throw here — nothing to clean up, state remains CREATED.
if (!name_server_resolver_) {
SPDLOG_ERROR("No name server resolver is configured.");
- abort();
+ throw std::runtime_error("No name server resolver is configured. Set
endpoints via Configuration.");
}
+
+ // === Phase 2: Resource allocation (state remains CREATED) ===
+ // If an exception is thrown here, state stays CREATED → shutdown() is no-op
+ // → destructor does nothing → partially-allocated resources leak.
+ // However, start() is called on fully-constructed objects; exceptions during
+ // resource allocation are rare (network errors are handled internally).
+
name_server_resolver_->start();
client_config_.client_id = clientId();
@@ -118,7 +121,7 @@ void ClientImpl::start() {
const auto& endpoint = name_server_resolver_->resolve();
if (endpoint.empty()) {
SPDLOG_ERROR("Failed to resolve name server address");
- return;
+ throw std::runtime_error("Failed to resolve name server address. Check
endpoint configuration.");
}
// A gRPC I/O thread pool is created upon establishing a connection.
@@ -207,6 +210,12 @@ void ClientImpl::start() {
opencensus::stats::StatsExporter::RegisterPushHandler(
absl::make_unique<OpencensusHandler>(metric_service_endpoint,
client_weak_ptr));
}
+
+ // If shutdown was requested concurrently during start(), honor it now
+ if (shutdown_requested_.load(std::memory_order_acquire)) {
+ SPDLOG_WARN("Shutdown requested during start, shutting down immediately");
+ shutdown();
+ }
}
std::string ClientImpl::metricServiceEndpoint() const {
@@ -244,23 +253,47 @@ std::string ClientImpl::metricServiceEndpoint() const {
return service_endpoint;
}
-void ClientImpl::shutdown() {
- State expected = State::STOPPING;
- if (state_.compare_exchange_strong(expected, State::STOPPED)) {
- name_server_resolver_->shutdown();
- if (route_update_handle_) {
+void ClientImpl::shutdown() noexcept {
+ shutdown_requested_.store(true, std::memory_order_release);
+
+ State expected = State::STARTED;
+ if (!state_.compare_exchange_strong(expected, State::STOPPED)) {
+ return;
+ }
+
+ // Cancel scheduled tasks
+ if (route_update_handle_ && client_manager_) {
+ try {
client_manager_->getScheduler()->cancel(route_update_handle_);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception cancelling route update: {}", e.what());
}
+ }
- if (telemetry_handle_) {
+ if (telemetry_handle_ && client_manager_) {
+ try {
client_manager_->getScheduler()->cancel(telemetry_handle_);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception cancelling telemetry sync: {}", e.what());
}
+ }
- client_manager_.reset();
- } else {
- SPDLOG_ERROR("Try to shutdown ClientImpl, but its state is not as
expected. Expecting: {}, Actual: {}",
- State::STOPPING, state_.load(std::memory_order_relaxed));
+ // Close gRPC sessions promptly rather than waiting for object destruction
+ {
+ absl::MutexLock lk(&session_map_mtx_);
+ session_map_.clear();
+ }
+
+ // Shutdown infrastructure
+ if (name_server_resolver_) {
+ try {
+ name_server_resolver_->shutdown();
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception shutting down name server resolver: {}",
e.what());
+ }
}
+
+ client_manager_.reset();
}
const char* ClientImpl::UPDATE_ROUTE_TASK_NAME = "route_updater";
@@ -368,9 +401,9 @@ void ClientImpl::syncClientSettings() {
}
void ClientImpl::updateRouteInfo() {
- if (State::STARTED != state_.load(std::memory_order_relaxed) &&
- State::STARTING != state_.load(std::memory_order_relaxed)) {
- SPDLOG_WARN("Unexpected client instance state={}.",
state_.load(std::memory_order_relaxed));
+ State current = state_.load(std::memory_order_relaxed);
+ if (State::STARTED != current) {
+ SPDLOG_WARN("Skipping route update: client state={}", current);
return;
}
@@ -408,13 +441,21 @@ void ClientImpl::heartbeat() {
absl::flat_hash_map<std::string, std::string> metadata;
Signature::sign(client_config_, metadata);
+ std::weak_ptr<ClientImpl> self_weak(self());
for (const auto& target : hosts) {
- auto callback = [target](const std::error_code& ec, const
HeartbeatResponse& response) {
+ auto callback = [target, self_weak](const std::error_code& ec, const
HeartbeatResponse& response) {
if (ec) {
SPDLOG_WARN("Failed to heartbeat against {}. Cause: {}", target,
ec.message());
return;
}
SPDLOG_DEBUG("Heartbeat to {} OK", target);
+ auto self = self_weak.lock();
+ if (self) {
+ absl::MutexLock lk(&self->isolated_endpoints_mtx_);
+ if (self->isolated_endpoints_.erase(target)) {
+ SPDLOG_INFO("Removed isolation for endpoint {} after successful
heartbeat", target);
+ }
+ }
};
client_manager_->heartbeat(target, metadata, request,
absl::ToChronoMilliseconds(client_config_.request_timeout), callback);
@@ -594,12 +635,17 @@ void ClientImpl::onRemoteEndpointRemoval(const
std::vector<std::string>& hosts)
void ClientImpl::schedule(const std::string& task_name, const
std::function<void()>& task,
std::chrono::milliseconds delay) {
- client_manager_->getScheduler()->schedule(task, task_name, delay,
std::chrono::milliseconds(0));
+ auto cm = client_manager_;
+ if (!cm) {
+ SPDLOG_WARN("Cannot schedule {}: client manager is null", task_name);
+ return;
+ }
+ cm->getScheduler()->schedule(task, task_name, delay,
std::chrono::milliseconds(0));
}
void ClientImpl::notifyClientTermination() {
- SPDLOG_WARN("Should NOT reach here. Subclass should have overridden this
function.");
- std::abort();
+ // Default: no-op. Subclasses that need to notify the server should override.
+ SPDLOG_DEBUG("notifyClientTermination() not overridden — skipping server
notification");
}
void ClientImpl::notifyClientTermination(const NotifyClientTerminationRequest&
request) {
diff --git a/cpp/source/rocketmq/FifoProducer.cpp
b/cpp/source/rocketmq/FifoProducer.cpp
index b52630b0..32971ad4 100644
--- a/cpp/source/rocketmq/FifoProducer.cpp
+++ b/cpp/source/rocketmq/FifoProducer.cpp
@@ -20,6 +20,8 @@
#include <cstddef>
#include <memory>
+#include <spdlog/spdlog.h>
+
#include "FifoProducerImpl.h"
#include "ProducerImpl.h"
#include "StaticNameServerResolver.h"
@@ -46,8 +48,8 @@ FifoProducerBuilder&
FifoProducerBuilder::withConfiguration(Configuration config
return *this;
}
-FifoProducerBuilder& FifoProducerBuilder::withTopics(const
std::vector<std::string>& topics) {
- producer_impl_->withTopics(topics);
+FifoProducerBuilder& FifoProducerBuilder::withTopics(std::vector<std::string>
topics) {
+ producer_impl_->withTopics(std::move(topics));
return *this;
}
@@ -66,8 +68,15 @@ void FifoProducer::start() {
impl_->internalProducer()->start();
}
-void FifoProducer::send(MessageConstPtr message, SendCallback callback) {
- impl_->send(std::move(message), callback);
+void FifoProducer::send(MessageConstPtr message, SendCallback callback)
noexcept {
+ try {
+ impl_->send(std::move(message), callback);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in FifoProducer::send: {}", e.what());
+ std::error_code ec = std::make_error_code(std::errc::io_error);
+ SendReceipt empty;
+ callback(ec, std::move(empty));
+ }
}
ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/FifoProducerPartition.cpp
b/cpp/source/rocketmq/FifoProducerPartition.cpp
index 67693f0d..8d42af9b 100644
--- a/cpp/source/rocketmq/FifoProducerPartition.cpp
+++ b/cpp/source/rocketmq/FifoProducerPartition.cpp
@@ -44,31 +44,48 @@ void FifoProducerPartition::add(FifoContext&& context) {
void FifoProducerPartition::trySend() {
bool expected = false;
if (inflight_.compare_exchange_strong(expected, true,
std::memory_order_relaxed)) {
- absl::MutexLock lk(&messages_mtx_);
+ MessageConstPtr message;
+ SendCallback send_callback;
- if (messages_.empty()) {
- SPDLOG_DEBUG("There is no more messages to send");
- return;
- }
+ {
+ absl::MutexLock lk(&messages_mtx_);
+ if (messages_.empty()) {
+ SPDLOG_DEBUG("There is no more messages to send");
+ inflight_.store(false, std::memory_order_release);
+ return;
+ }
- FifoContext& ctx = messages_.front();
- MessageConstPtr message = std::move(ctx.message);
- SendCallback send_callback = ctx.callback;
+ FifoContext& ctx = messages_.front();
+ message = std::move(ctx.message);
+ send_callback = ctx.callback;
+ messages_.pop_front();
+ }
+ // Lock released — producer_->send() and its callbacks run without holding
messages_mtx_.
+ // This prevents deadlock when send() fails synchronously and invokes the
callback
+ // on the same thread (onComplete would try to re-acquire messages_mtx_).
std::shared_ptr<FifoProducerPartition> partition = shared_from_this();
- auto fifo_callback = [=](const std::error_code& ec, const SendReceipt&
receipt) mutable {
- partition->onComplete(ec, receipt, send_callback);
+ auto fifo_callback = [=](const std::error_code& ec, SendReceipt&& receipt)
mutable {
+ partition->onComplete(ec, std::move(receipt), send_callback);
};
SPDLOG_DEBUG("Sending FIFO message from {}", name_);
- producer_->send(std::move(message), fifo_callback);
- messages_.pop_front();
- SPDLOG_DEBUG("In addition to the inflight one, there is {} messages
pending in {}", messages_.size(), name_);
+ try {
+ producer_->send(std::move(message), fifo_callback);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in FifoProducerPartition::trySend: {}",
e.what());
+ // Message is lost (consumed by the throwing send call via unique_ptr
move).
+ // Invoke user callback directly to avoid null deref from retrying with
empty SendReceipt.
+ std::error_code ec = std::make_error_code(std::errc::operation_canceled);
+ SendReceipt empty;
+ send_callback(ec, std::move(empty));
+ inflight_.store(false, std::memory_order_release);
+ }
} else {
SPDLOG_DEBUG("There is an inflight message");
}
}
-void FifoProducerPartition::onComplete(const std::error_code& ec, const
SendReceipt& receipt, SendCallback& callback) {
+void FifoProducerPartition::onComplete(const std::error_code& ec,
SendReceipt&& receipt, SendCallback& callback) {
if (ec) {
SPDLOG_INFO("{} completed with a failure: {}", name_, ec.message());
} else {
@@ -76,7 +93,7 @@ void FifoProducerPartition::onComplete(const std::error_code&
ec, const SendRece
}
if (!ec) {
- callback(ec, receipt);
+ callback(ec, std::move(receipt));
// update inflight status
bool expected = true;
if (inflight_.compare_exchange_strong(expected, false,
std::memory_order_relaxed)) {
@@ -88,8 +105,7 @@ void FifoProducerPartition::onComplete(const
std::error_code& ec, const SendRece
}
// Put the message back to the front of the list.
- // receipt is a local temporary in SendContext — const_cast + move is safe
here.
- FifoContext
retry_context(std::move(const_cast<SendReceipt&>(receipt).message), callback);
+ FifoContext retry_context(std::move(receipt.message), callback);
{
absl::MutexLock lk(&messages_mtx_);
messages_.emplace_front(std::move(retry_context));
diff --git a/cpp/source/rocketmq/Producer.cpp b/cpp/source/rocketmq/Producer.cpp
index ebccf60e..0842c1f8 100644
--- a/cpp/source/rocketmq/Producer.cpp
+++ b/cpp/source/rocketmq/Producer.cpp
@@ -20,6 +20,8 @@
#include <system_error>
#include <utility>
+#include <spdlog/spdlog.h>
+
#include "ProducerImpl.h"
#include "StaticNameServerResolver.h"
#include "rocketmq/ErrorCode.h"
@@ -34,42 +36,67 @@ void Producer::start() {
}
SendReceipt Producer::send(MessageConstPtr message, std::error_code& ec)
noexcept {
- if (!message) {
- ec = ErrorCode::BadRequest;
+ try {
+ if (!message) {
+ ec = ErrorCode::BadRequest;
+ return {};
+ }
+
+ return impl_->send(std::move(message), ec);
+ } catch (const std::exception& e) {
+ ec = std::make_error_code(std::errc::io_error);
+ SPDLOG_ERROR("Exception in send: {}", e.what());
return {};
}
-
- return impl_->send(std::move(message), ec);
}
-void Producer::send(MessageConstPtr message, const SendCallback& callback)
noexcept {
- if (!message) {
- std::error_code ec = ErrorCode::BadRequest;
- SendReceipt send_receipt = {};
- callback(ec, send_receipt);
- return;
+void Producer::send(MessageConstPtr message, SendCallback callback) noexcept {
+ try {
+ if (!message) {
+ std::error_code ec = ErrorCode::BadRequest;
+ SendReceipt send_receipt = {};
+ callback(ec, std::move(send_receipt));
+ return;
+ }
+
+ if (!message->group().empty()) {
+ SendReceipt empty;
+ std::error_code ec = ErrorCode::BadRequestAsyncPubFifoMessage;
+ callback(ec, std::move(empty));
+ return;
+ }
+
+ impl_->send(std::move(message), callback);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in async send: {}", e.what());
+ std::error_code ec = std::make_error_code(std::errc::io_error);
+ SendReceipt empty;
+ callback(ec, std::move(empty));
}
-
- if (!message->group().empty()) {
- SendReceipt empty;
- std::error_code ec = ErrorCode::BadRequestAsyncPubFifoMessage;
- callback(ec, empty);
- return;
- }
-
- impl_->send(std::move(message), callback);
}
std::unique_ptr<Transaction> Producer::beginTransaction() {
return impl_->beginTransaction();
}
-SendReceipt Producer::send(MessageConstPtr message, std::error_code& ec,
Transaction& transaction) {
- return impl_->send(std::move(message), ec, transaction);
+SendReceipt Producer::send(MessageConstPtr message, std::error_code& ec,
Transaction& transaction) noexcept {
+ try {
+ return impl_->send(std::move(message), ec, transaction);
+ } catch (const std::exception& e) {
+ ec = std::make_error_code(std::errc::io_error);
+ SPDLOG_ERROR("Exception in transactional send: {}", e.what());
+ return {};
+ }
}
RecallReceipt Producer::recall(std::string& topic, std::string& recall_handle,
std::error_code& ec) noexcept {
- return impl_->recall(topic, recall_handle, ec);
+ try {
+ return impl_->recall(topic, recall_handle, ec);
+ } catch (const std::exception& e) {
+ ec = std::make_error_code(std::errc::io_error);
+ SPDLOG_ERROR("Exception in recall: {}", e.what());
+ return {};
+ }
}
ProducerBuilder Producer::newBuilder() {
@@ -78,7 +105,7 @@ ProducerBuilder Producer::newBuilder() {
ProducerBuilder::ProducerBuilder() : impl_(std::make_shared<ProducerImpl>()){}
-ProducerBuilder& ProducerBuilder::withConfiguration(const Configuration&
configuration) {
+ProducerBuilder& ProducerBuilder::withConfiguration(Configuration
configuration) {
auto name_server_resolver =
std::make_shared<StaticNameServerResolver>(configuration.endpoints());
impl_->withNameServerResolver(std::move(name_server_resolver));
impl_->withResourceNamespace(configuration.resourceNamespace());
@@ -89,13 +116,13 @@ ProducerBuilder& ProducerBuilder::withConfiguration(const
Configuration& configu
return *this;
}
-ProducerBuilder& ProducerBuilder::withTopics(const std::vector<std::string>&
topics) {
- impl_->withTopics(topics);
+ProducerBuilder& ProducerBuilder::withTopics(std::vector<std::string> topics) {
+ impl_->withTopics(std::move(topics));
return *this;
}
-ProducerBuilder& ProducerBuilder::withTransactionChecker(const
TransactionChecker& checker) {
- impl_->transaction_checker_ = checker;
+ProducerBuilder& ProducerBuilder::withTransactionChecker(TransactionChecker
checker) {
+ impl_->transaction_checker_ = std::move(checker);
return *this;
}
diff --git a/cpp/source/rocketmq/ProducerImpl.cpp
b/cpp/source/rocketmq/ProducerImpl.cpp
index 3450faf2..22027ab8 100644
--- a/cpp/source/rocketmq/ProducerImpl.cpp
+++ b/cpp/source/rocketmq/ProducerImpl.cpp
@@ -55,9 +55,9 @@ ProducerImpl::~ProducerImpl() {
void ProducerImpl::start() {
ClientImpl::start();
- State expecting = State::STARTING;
- if (!state_.compare_exchange_strong(expecting, State::STARTED)) {
- SPDLOG_ERROR("Producer started with an unexpected state. Expecting: {},
Actual: {}", State::STARTING,
+ State expected = State::CREATED;
+ if (!state_.compare_exchange_strong(expected, State::STARTED)) {
+ SPDLOG_ERROR("Producer started with unexpected state. Expecting: {},
Actual: {}", State::CREATED,
state_.load(std::memory_order_relaxed));
return;
}
@@ -65,19 +65,19 @@ void ProducerImpl::start() {
client_manager_->addClientObserver(shared_from_this());
}
-void ProducerImpl::shutdown() {
+void ProducerImpl::shutdown() noexcept {
State expected = State::STARTED;
- if (!state_.compare_exchange_strong(expected, State::STOPPING)) {
- SPDLOG_ERROR("Shutdown with unexpected state. Expecting: {}, Actual: {}",
State::STOPPING,
- state_.load(std::memory_order_relaxed));
+ if (!state_.compare_exchange_strong(expected, State::STOPPED)) {
return;
}
- notifyClientTermination();
+ {
+ absl::MutexLock lock(&topic_publish_info_mtx_);
+ topic_publish_info_table_.clear();
+ }
ClientImpl::shutdown();
- assert(State::STOPPED == state_.load());
- SPDLOG_INFO("Producer instance stopped");
+ SPDLOG_INFO("ProducerImpl stopped");
}
void ProducerImpl::notifyClientTermination() {
@@ -138,14 +138,11 @@ void ProducerImpl::wrapSendMessageRequest(const Message&
message, SendMessageReq
// Delivery Timestamp
if (message.deliveryTimestamp().time_since_epoch().count()) {
- auto delivery_timestamp = message.deliveryTimestamp();
- if (delivery_timestamp.time_since_epoch().count()) {
- auto duration = delivery_timestamp.time_since_epoch();
-
system_properties->set_delivery_attempt(std::chrono::duration_cast<std::chrono::milliseconds>(duration).count());
- auto mutable_delivery_timestamp =
system_properties->mutable_delivery_timestamp();
-
mutable_delivery_timestamp->set_seconds(std::chrono::duration_cast<std::chrono::seconds>(duration).count());
-
mutable_delivery_timestamp->set_nanos(std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count()
% 1000000000);
- }
+ auto duration = message.deliveryTimestamp().time_since_epoch();
+
system_properties->set_delivery_attempt(std::chrono::duration_cast<std::chrono::milliseconds>(duration).count());
+ auto mutable_delivery_timestamp =
system_properties->mutable_delivery_timestamp();
+
mutable_delivery_timestamp->set_seconds(std::chrono::duration_cast<std::chrono::seconds>(duration).count());
+
mutable_delivery_timestamp->set_nanos(std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count()
% 1000000000);
}
// Born-time
@@ -233,16 +230,13 @@ SendReceipt ProducerImpl::send(MessageConstPtr message,
std::error_code& ec) noe
// Define callback
auto callback =
- [&, mtx, cv](const std::error_code& code, const SendReceipt& receipt)
mutable {
+ [&, mtx, cv](const std::error_code& code, SendReceipt&& receipt) mutable
{
ec = code;
- // SendReceipt contains a unique_ptr (MessageConstPtr) and is non-copyable.
- // The receipt here is always a local temporary in SendContext, so
const_cast + move is safe.
- auto& mutable_receipt = const_cast<SendReceipt&>(receipt);
- send_receipt.target = std::move(mutable_receipt.target);
- send_receipt.message_id = std::move(mutable_receipt.message_id);
- send_receipt.message = std::move(mutable_receipt.message);
- send_receipt.transaction_id = std::move(mutable_receipt.transaction_id);
- send_receipt.recall_handle = std::move(mutable_receipt.recall_handle);
+ send_receipt.target = std::move(receipt.target);
+ send_receipt.message_id = std::move(receipt.message_id);
+ send_receipt.message = std::move(receipt.message);
+ send_receipt.transaction_id = std::move(receipt.transaction_id);
+ send_receipt.recall_handle = std::move(receipt.recall_handle);
{
absl::MutexLock lk(mtx.get());
completed = true;
@@ -268,7 +262,8 @@ void ProducerImpl::send(MessageConstPtr message,
SendCallback cb) {
if (ec) {
SendReceipt send_receipt;
send_receipt.message = std::move(message);
- cb(ec, send_receipt);
+ cb(ec, std::move(send_receipt));
+ return;
}
std::string topic = message->topic();
@@ -282,7 +277,7 @@ void ProducerImpl::send(MessageConstPtr message,
SendCallback cb) {
if (ec) {
SendReceipt send_receipt;
send_receipt.message = std::move(ptr);
- cb(ec, send_receipt);
+ cb(ec, std::move(send_receipt));
return;
}
@@ -290,7 +285,7 @@ void ProducerImpl::send(MessageConstPtr message,
SendCallback cb) {
std::error_code ec = ErrorCode::NotFound;
SendReceipt send_receipt;
send_receipt.message = std::move(ptr);
- cb(ec, send_receipt);
+ cb(ec, std::move(send_receipt));
return;
}
@@ -300,7 +295,7 @@ void ProducerImpl::send(MessageConstPtr message,
SendCallback cb) {
std::error_code ec = ErrorCode::NotFound;
SendReceipt send_receipt;
send_receipt.message = std::move(ptr);
- cb(ec, send_receipt);
+ cb(ec, std::move(send_receipt));
return;
}
@@ -375,14 +370,14 @@ void ProducerImpl::send0(MessageConstPtr message, const
SendCallback& callback,
validate(*message, ec);
if (ec) {
send_receipt.message = std::move(message);
- callback(ec, send_receipt);
+ callback(ec, std::move(send_receipt));
return;
}
if (list.empty()) {
ec = ErrorCode::NotFound;
send_receipt.message = std::move(message);
- callback(ec, send_receipt);
+ callback(ec, std::move(send_receipt));
return;
}
@@ -682,11 +677,11 @@ void
ProducerImpl::topicsOfInterest(std::vector<std::string> &topics) {
}
}
-void ProducerImpl::withTopics(const std::vector<std::string> &topics) {
+void ProducerImpl::withTopics(std::vector<std::string> topics) {
absl::MutexLock lk(&topics_mtx_);
- for (auto &topic: topics) {
+ for (auto& topic : topics) {
if (std::find(topics_.begin(), topics_.end(), topic) == topics_.end()) {
- topics_.push_back(topic);
+ topics_.push_back(std::move(topic));
}
}
}
diff --git a/cpp/source/rocketmq/PushConsumer.cpp
b/cpp/source/rocketmq/PushConsumer.cpp
index a6dd3d35..726ee673 100644
--- a/cpp/source/rocketmq/PushConsumer.cpp
+++ b/cpp/source/rocketmq/PushConsumer.cpp
@@ -17,18 +17,28 @@
#include <chrono>
#include <memory>
+#include <spdlog/spdlog.h>
+
#include "PushConsumerImpl.h"
#include "StaticNameServerResolver.h"
#include "rocketmq/PushConsumer.h"
ROCKETMQ_NAMESPACE_BEGIN
-void PushConsumer::subscribe(std::string topic, FilterExpression
filter_expression) {
- impl_->subscribe(std::move(topic), filter_expression.content_,
filter_expression.type_);
+void PushConsumer::subscribe(std::string topic, FilterExpression
filter_expression) noexcept {
+ try {
+ impl_->subscribe(std::move(topic), filter_expression.content_,
filter_expression.type_);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in subscribe: {}", e.what());
+ }
}
-void PushConsumer::unsubscribe(const std::string& topic) {
- impl_->unsubscribe(topic);
+void PushConsumer::unsubscribe(const std::string& topic) noexcept {
+ try {
+ impl_->unsubscribe(topic);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in unsubscribe: {}", e.what());
+ }
}
PushConsumerBuilder PushConsumer::newBuilder() {
diff --git a/cpp/source/rocketmq/PushConsumerImpl.cpp
b/cpp/source/rocketmq/PushConsumerImpl.cpp
index bd2a9977..124db9f8 100644
--- a/cpp/source/rocketmq/PushConsumerImpl.cpp
+++ b/cpp/source/rocketmq/PushConsumerImpl.cpp
@@ -54,20 +54,20 @@ void
PushConsumerImpl::topicsOfInterest(std::vector<std::string> &topics) {
}
void PushConsumerImpl::start() {
+ if (!message_listener_) {
+ SPDLOG_ERROR("Required message listener is missing");
+ abort();
+ }
+
ClientImpl::start();
- State expecting = State::STARTING;
- if (!state_.compare_exchange_strong(expecting, State::STARTED)) {
- SPDLOG_ERROR("Unexpected consumer state. Expecting: {}, Actual: {}",
State::STARTING,
+ State expected = State::CREATED;
+ if (!state_.compare_exchange_strong(expected, State::STARTED)) {
+ SPDLOG_ERROR("PushConsumer started with unexpected state. Expecting: {},
Actual: {}", State::CREATED,
state_.load(std::memory_order_relaxed));
return;
}
- if (!message_listener_) {
- SPDLOG_ERROR("Required message listener is missing");
- abort();
- }
-
client_config_.subscriber.group.set_resource_namespace(resourceNamespace());
client_manager_->addClientObserver(shared_from_this());
@@ -109,36 +109,33 @@ void PushConsumerImpl::start() {
const char* PushConsumerImpl::SCAN_ASSIGNMENT_TASK_NAME =
"scan-assignment-task";
const char* PushConsumerImpl::COLLECT_STATS_TASK_NAME = "collect-stats-task";
-void PushConsumerImpl::shutdown() {
- State expecting = State::STARTED;
- if (state_.compare_exchange_strong(expecting, State::STOPPING)) {
- if (scan_assignment_handle_) {
- client_manager_->getScheduler()->cancel(scan_assignment_handle_);
- SPDLOG_DEBUG("Scan assignment periodic task cancelled");
- }
-
- if (collect_stats_handle_) {
- client_manager_->getScheduler()->cancel(collect_stats_handle_);
- SPDLOG_DEBUG("Collect cache stats periodic task cancelled");
- }
+void PushConsumerImpl::shutdown() noexcept {
+ State expected = State::STARTED;
+ if (!state_.compare_exchange_strong(expected, State::STOPPED)) {
+ return;
+ }
- {
- absl::MutexLock lock(&process_queue_table_mtx_);
- process_queue_table_.clear();
- }
+ if (scan_assignment_handle_) {
+ client_manager_->getScheduler()->cancel(scan_assignment_handle_);
+ SPDLOG_DEBUG("Scan assignment periodic task cancelled");
+ }
- if (consume_message_service_) {
- consume_message_service_->shutdown();
- }
+ if (collect_stats_handle_) {
+ client_manager_->getScheduler()->cancel(collect_stats_handle_);
+ SPDLOG_DEBUG("Collect cache stats periodic task cancelled");
+ }
- // Shutdown services started by parent
- ClientImpl::shutdown();
+ {
+ absl::MutexLock lock(&process_queue_table_mtx_);
+ process_queue_table_.clear();
+ }
- SPDLOG_INFO("PushConsumerImpl stopped");
- } else {
- SPDLOG_ERROR("Shutdown with unexpected state. Expecting: {}, Actual: {}",
State::STARTED,
- state_.load(std::memory_order_relaxed));
+ if (consume_message_service_) {
+ consume_message_service_->shutdown();
}
+
+ ClientImpl::shutdown();
+ SPDLOG_INFO("PushConsumerImpl stopped");
}
void PushConsumerImpl::subscribe(const std::string& topic, const std::string&
expression,
diff --git a/cpp/source/rocketmq/SendContext.cpp
b/cpp/source/rocketmq/SendContext.cpp
index 5f6b8fdc..17b8c4b5 100644
--- a/cpp/source/rocketmq/SendContext.cpp
+++ b/cpp/source/rocketmq/SendContext.cpp
@@ -59,7 +59,7 @@ void SendContext::onSuccess(const SendResult& send_result)
noexcept {
send_receipt.transaction_id = send_result.transaction_id;
send_receipt.recall_handle = send_result.recall_handle;
send_receipt.message = std::move(message_);
- callback_(send_result.ec, send_receipt);
+ callback_(send_result.ec, std::move(send_receipt));
}
void SendContext::onFailure(const std::error_code& ec) noexcept {
@@ -90,7 +90,7 @@ void SendContext::onFailure(const std::error_code& ec)
noexcept {
SPDLOG_WARN("Retried {} times, which exceeds the limit: {}",
attempt_times_, producer->maxAttemptTimes());
SendReceipt receipt{};
receipt.message = std::move(message_);
- callback_(ec, receipt);
+ callback_(ec, std::move(receipt));
return;
}
@@ -98,10 +98,19 @@ void SendContext::onFailure(const std::error_code& ec)
noexcept {
SPDLOG_WARN("No alternative hosts to perform additional retries");
SendReceipt receipt{};
receipt.message = std::move(message_);
- callback_(ec, receipt);
+ callback_(ec, std::move(receipt));
return;
}
+ // Isolate the failed endpoint to avoid retrying on the same broken broker
+ {
+ auto& failed_queue = candidates_[(attempt_times_ - 1) %
candidates_.size()];
+ auto target = failed_queue.broker().endpoints().addresses(0).host() + ":" +
+
std::to_string(failed_queue.broker().endpoints().addresses(0).port());
+ producer->isolateEndpoint(target);
+ SPDLOG_INFO("Isolated endpoint {} after send failure", target);
+ }
+
auto message_queue = candidates_[attempt_times_ % candidates_.size()];
request_time_ = std::chrono::steady_clock::now();
diff --git a/cpp/source/rocketmq/SimpleConsumer.cpp
b/cpp/source/rocketmq/SimpleConsumer.cpp
index 152c0537..a791e869 100644
--- a/cpp/source/rocketmq/SimpleConsumer.cpp
+++ b/cpp/source/rocketmq/SimpleConsumer.cpp
@@ -17,6 +17,8 @@
#include "rocketmq/SimpleConsumer.h"
+#include <spdlog/spdlog.h>
+
#include "SimpleConsumerImpl.h"
#include "StaticNameServerResolver.h"
#include "rocketmq/ErrorCode.h"
@@ -37,92 +39,135 @@ void SimpleConsumer::start() {
impl_->start();
}
-void SimpleConsumer::subscribe(std::string topic, FilterExpression
filter_expression) {
- impl_->subscribe(topic, filter_expression);
+void SimpleConsumer::subscribe(std::string topic, FilterExpression
filter_expression) noexcept {
+ try {
+ impl_->subscribe(topic, filter_expression);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in subscribe: {}", e.what());
+ }
}
-void SimpleConsumer::unsubscribe(const std::string& topic) {
- impl_->unsubscribe(topic);
+void SimpleConsumer::unsubscribe(const std::string& topic) noexcept {
+ try {
+ impl_->unsubscribe(topic);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in unsubscribe: {}", e.what());
+ }
}
void SimpleConsumer::receive(std::size_t limit,
std::chrono::milliseconds invisible_duration,
std::error_code& ec,
- std::vector<MessageConstSharedPtr>& messages) {
- auto mtx = std::make_shared<absl::Mutex>();
- auto cv = std::make_shared<absl::CondVar>();
- bool completed = false;
- auto callback = [&, mtx, cv](const std::error_code& code, const
std::vector<MessageConstSharedPtr>& result) {
- {
- absl::MutexLock lk(mtx.get());
- if (code && code != ErrorCode::NoContent) {
- ec = code;
- SPDLOG_WARN("Failed to receive message. Cause: {}", code.message());
+ std::vector<MessageConstSharedPtr>& messages)
noexcept {
+ try {
+ auto mtx = std::make_shared<absl::Mutex>();
+ auto cv = std::make_shared<absl::CondVar>();
+ bool completed = false;
+ auto callback = [&, mtx, cv](const std::error_code& code, const
std::vector<MessageConstSharedPtr>& result) {
+ {
+ absl::MutexLock lk(mtx.get());
+ if (code && code != ErrorCode::NoContent) {
+ ec = code;
+ SPDLOG_WARN("Failed to receive message. Cause: {}", code.message());
+ }
+ completed = true;
+ messages.insert(messages.end(), result.begin(), result.end());
}
- completed = true;
- messages.insert(messages.end(), result.begin(), result.end());
- }
- cv->SignalAll();
- };
+ cv->SignalAll();
+ };
- impl_->receive(limit, invisible_duration, callback);
+ impl_->receive(limit, invisible_duration, callback);
- {
- absl::MutexLock lk(mtx.get());
- while (!completed) {
- cv->Wait(mtx.get());
+ {
+ absl::MutexLock lk(mtx.get());
+ while (!completed) {
+ cv->Wait(mtx.get());
+ }
}
+ } catch (const std::exception& e) {
+ ec = std::make_error_code(std::errc::io_error);
+ SPDLOG_ERROR("Exception in receive: {}", e.what());
}
}
void SimpleConsumer::asyncReceive(std::size_t limit,
std::chrono::milliseconds invisible_duration,
- ReceiveCallback callback) {
- impl_->receive(limit, invisible_duration, callback);
+ ReceiveCallback callback) noexcept {
+ try {
+ impl_->receive(limit, invisible_duration, callback);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in asyncReceive: {}", e.what());
+ std::error_code ec = std::make_error_code(std::errc::io_error);
+ std::vector<MessageConstSharedPtr> empty;
+ callback(ec, empty);
+ }
}
-void SimpleConsumer::ack(const Message& message, std::error_code& ec) {
- impl_->ack(message, ec);
+void SimpleConsumer::ack(const Message& message, std::error_code& ec) noexcept
{
+ try {
+ impl_->ack(message, ec);
+ } catch (const std::exception& e) {
+ ec = std::make_error_code(std::errc::io_error);
+ SPDLOG_ERROR("Exception in ack: {}", e.what());
+ }
}
-void SimpleConsumer::asyncAck(const Message& message, AckCallback callback) {
- impl_->ackAsync(message, callback);
+void SimpleConsumer::asyncAck(const Message& message, AckCallback callback)
noexcept {
+ try {
+ impl_->ackAsync(message, callback);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in asyncAck: {}", e.what());
+ std::error_code ec = std::make_error_code(std::errc::io_error);
+ callback(ec);
+ }
}
void SimpleConsumer::changeInvisibleDuration(const Message& message,
std::string& receipt_handle,
std::chrono::milliseconds
duration,
- std::error_code& ec) {
- auto mtx = std::make_shared<absl::Mutex>();
- auto cv = std::make_shared<absl::CondVar>();
- bool completed = false;
+ std::error_code& ec) noexcept {
+ try {
+ auto mtx = std::make_shared<absl::Mutex>();
+ auto cv = std::make_shared<absl::CondVar>();
+ bool completed = false;
+
+ auto callback =
+ [&, mtx, cv](const std::error_code& code, std::string&
server_receipt_handle) {
+ {
+ absl::MutexLock lk(mtx.get());
+ completed = true;
+ ec = code;
+ if (!ec) {
+ receipt_handle = server_receipt_handle;
+ }
+ }
+ cv->Signal();
+ };
+
+ impl_->changeInvisibleDuration(message, receipt_handle, duration,
callback);
- auto callback =
- [&, mtx, cv](const std::error_code& code, std::string&
server_receipt_handle) {
{
absl::MutexLock lk(mtx.get());
- completed = true;
- ec = code;
- if (!ec) {
- receipt_handle = server_receipt_handle;
+ if (!completed) {
+ cv->Wait(mtx.get());
}
}
- cv->Signal();
- };
-
- impl_->changeInvisibleDuration(message, receipt_handle, duration, callback);
-
- {
- absl::MutexLock lk(mtx.get());
- if (!completed) {
- cv->Wait(mtx.get());
- }
+ } catch (const std::exception& e) {
+ ec = std::make_error_code(std::errc::io_error);
+ SPDLOG_ERROR("Exception in changeInvisibleDuration: {}", e.what());
}
}
void SimpleConsumer::asyncChangeInvisibleDuration(const Message& message,
std::string& receipt_handle,
std::chrono::milliseconds
duration,
-
ChangeInvisibleDurationCallback callback) {
- impl_->changeInvisibleDuration(message, receipt_handle, duration, callback);
+
ChangeInvisibleDurationCallback callback) noexcept {
+ try {
+ impl_->changeInvisibleDuration(message, receipt_handle, duration,
callback);
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in asyncChangeInvisibleDuration: {}", e.what());
+ std::error_code ec = std::make_error_code(std::errc::io_error);
+ std::string handle;
+ callback(ec, handle);
+ }
}
SimpleConsumer SimpleConsumerBuilder::build() {
diff --git a/cpp/source/rocketmq/SimpleConsumerImpl.cpp
b/cpp/source/rocketmq/SimpleConsumerImpl.cpp
index d1508d8b..96bb7b37 100644
--- a/cpp/source/rocketmq/SimpleConsumerImpl.cpp
+++ b/cpp/source/rocketmq/SimpleConsumerImpl.cpp
@@ -76,43 +76,44 @@ void
SimpleConsumerImpl::topicsOfInterest(std::vector<std::string> &topics) {
}
}
-/**
- * @brief Start SimpleConsumer
- *
- * During start, we need synchronously fetch routes and query assignments
- */
void SimpleConsumerImpl::start() {
ClientImpl::start();
- State expected = State::STARTING;
- if (state_.compare_exchange_strong(expected, State::STARTED,
std::memory_order_relaxed)) {
-
client_config_.subscriber.group.set_resource_namespace(resourceNamespace());
- refreshAssignments();
-
- std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
- auto refresh_assignment_task = [consumer]() {
- auto simple_consumer = consumer.lock();
- if (simple_consumer) {
- simple_consumer->refreshAssignments0();
- }
- };
-
- // refer java sdk: set refresh interval to 5 seconds
- // org.apache.rocketmq.client.java.impl.ClientImpl#startUp
- refresh_assignment_task_ = manager()->getScheduler()->schedule(
- refresh_assignment_task, "RefreshAssignmentTask",
- std::chrono::seconds(5), std::chrono::seconds(5));
- client_manager_->addClientObserver(shared_from_this());
+ State expected = State::CREATED;
+ if (!state_.compare_exchange_strong(expected, State::STARTED)) {
+ SPDLOG_ERROR("SimpleConsumer started with unexpected state. Expecting: {},
Actual: {}", State::CREATED,
+ state_.load(std::memory_order_relaxed));
+ return;
}
+
+ client_config_.subscriber.group.set_resource_namespace(resourceNamespace());
+ refreshAssignments();
+
+ std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
+ auto refresh_assignment_task = [consumer]() {
+ auto simple_consumer = consumer.lock();
+ if (simple_consumer) {
+ simple_consumer->refreshAssignments0();
+ }
+ };
+
+ // refer java sdk: set refresh interval to 5 seconds
+ // org.apache.rocketmq.client.java.impl.ClientImpl#startUp
+ refresh_assignment_task_ = manager()->getScheduler()->schedule(
+ refresh_assignment_task, "RefreshAssignmentTask",
+ std::chrono::seconds(5), std::chrono::seconds(5));
+
+ client_manager_->addClientObserver(shared_from_this());
}
-void SimpleConsumerImpl::shutdown() {
+void SimpleConsumerImpl::shutdown() noexcept {
State expected = State::STARTED;
-
- if (state_.compare_exchange_strong(expected, State::STOPPING,
std::memory_order_relaxed)) {
- manager()->getScheduler()->cancel(refresh_assignment_task_);
- ClientImpl::shutdown();
+ if (!state_.compare_exchange_strong(expected, State::STOPPED)) {
+ return;
}
+
+ manager()->getScheduler()->cancel(refresh_assignment_task_);
+ ClientImpl::shutdown();
}
void SimpleConsumerImpl::subscribe(std::string topic, FilterExpression
expression) {
@@ -245,6 +246,10 @@ void SimpleConsumerImpl::refreshAssignment(const
std::string& topic, std::functi
std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
auto callback = [consumer, topic, cb](const std::error_code& ec, const
rmq::QueryAssignmentResponse& response) {
auto simple_consumer = consumer.lock();
+ if (!simple_consumer) {
+ SPDLOG_WARN("SimpleConsumer has been destructed, dropping assignment
response");
+ return;
+ }
const auto& assignments = response.assignments();
if (assignments.empty()) {
cb(ec);
@@ -441,7 +446,7 @@ void SimpleConsumerImpl::ackAsync(const Message& message,
AckCallback callback)
void SimpleConsumerImpl::changeInvisibleDuration(const Message& message,
std::string& receipt_handle,
std::chrono::milliseconds
duration,
- const
ChangeInvisibleDurationCallback callback) {
+
ChangeInvisibleDurationCallback callback) {
Metadata metadata;
Signature::sign(client_config_, metadata);
diff --git a/cpp/source/rocketmq/include/ClientImpl.h
b/cpp/source/rocketmq/include/ClientImpl.h
index 01967384..563d5089 100644
--- a/cpp/source/rocketmq/include/ClientImpl.h
+++ b/cpp/source/rocketmq/include/ClientImpl.h
@@ -55,7 +55,8 @@ public:
virtual void start();
- virtual void shutdown();
+ /// Must be noexcept — called from destructors (Bug 21 fix).
+ virtual void shutdown() noexcept;
void getRouteFor(const std::string& topic, const std::function<void(const
std::error_code&, TopicRouteDataPtr)>& cb)
LOCKS_EXCLUDED(inflight_route_requests_mtx_, topic_route_table_mtx_);
@@ -71,8 +72,7 @@ public:
void heartbeat() override;
bool active() override {
- State state = state_.load(std::memory_order_relaxed);
- return State::STARTING == state || State::STARTED == state;
+ return State::STARTED == state_.load(std::memory_order_relaxed);
}
void onRemoteEndpointRemoval(const std::vector<std::string>& hosts) override
LOCKS_EXCLUDED(isolated_endpoints_mtx_);
@@ -152,6 +152,7 @@ protected:
ClientManagerPtr client_manager_;
std::atomic<State> state_;
+ std::atomic<bool> shutdown_requested_{false};
absl::flat_hash_map<std::string, TopicRouteDataPtr> topic_route_table_
GUARDED_BY(topic_route_table_mtx_);
absl::Mutex topic_route_table_mtx_
ACQUIRED_AFTER(inflight_route_requests_mtx_); // protects topic_route_table_
diff --git a/cpp/source/rocketmq/include/FifoProducerPartition.h
b/cpp/source/rocketmq/include/FifoProducerPartition.h
index 8d0e00d8..199d11f4 100644
--- a/cpp/source/rocketmq/include/FifoProducerPartition.h
+++ b/cpp/source/rocketmq/include/FifoProducerPartition.h
@@ -42,7 +42,7 @@ public:
void trySend() LOCKS_EXCLUDED(messages_mtx_);
- void onComplete(const std::error_code& ec, const SendReceipt& receipt,
SendCallback& callback);
+ void onComplete(const std::error_code& ec, SendReceipt&& receipt,
SendCallback& callback);
private:
std::shared_ptr<ProducerImpl> producer_;
diff --git a/cpp/source/rocketmq/include/ProducerImpl.h
b/cpp/source/rocketmq/include/ProducerImpl.h
index 237c242e..53300ad8 100644
--- a/cpp/source/rocketmq/include/ProducerImpl.h
+++ b/cpp/source/rocketmq/include/ProducerImpl.h
@@ -47,7 +47,7 @@ public:
void start() override;
- void shutdown() override;
+ void shutdown() noexcept override;
/**
* Note we require application to transfer ownership of the message
@@ -128,7 +128,7 @@ public:
void topicsOfInterest(std::vector<std::string> &topics) override
LOCKS_EXCLUDED(topics_mtx_);
- void withTopics(const std::vector<std::string> &topics)
LOCKS_EXCLUDED(topics_mtx_);
+ void withTopics(std::vector<std::string> topics) LOCKS_EXCLUDED(topics_mtx_);
const PublishStats& stats() const {
return stats_;
diff --git a/cpp/source/rocketmq/include/PushConsumerImpl.h
b/cpp/source/rocketmq/include/PushConsumerImpl.h
index 42c3fe49..e8a80838 100644
--- a/cpp/source/rocketmq/include/PushConsumerImpl.h
+++ b/cpp/source/rocketmq/include/PushConsumerImpl.h
@@ -56,7 +56,7 @@ public:
void start() override;
- void shutdown() override;
+ void shutdown() noexcept override;
void subscribe(const std::string& topic,
const std::string& expression,
diff --git a/cpp/source/rocketmq/include/SimpleConsumerImpl.h
b/cpp/source/rocketmq/include/SimpleConsumerImpl.h
index 22850043..e0dc205f 100644
--- a/cpp/source/rocketmq/include/SimpleConsumerImpl.h
+++ b/cpp/source/rocketmq/include/SimpleConsumerImpl.h
@@ -22,7 +22,6 @@
#include "rocketmq/FilterExpression.h"
#include "rocketmq/SimpleConsumer.h"
-using namespace std::chrono;
ROCKETMQ_NAMESPACE_BEGIN
class SimpleConsumerImpl : virtual public ClientImpl, public
std::enable_shared_from_this<SimpleConsumerImpl> {
@@ -41,7 +40,7 @@ public:
void start() override;
- void shutdown() override;
+ void shutdown() noexcept override;
void subscribe(std::string topic, FilterExpression expression)
LOCKS_EXCLUDED(subscriptions_mtx_);
@@ -55,7 +54,7 @@ public:
void changeInvisibleDuration(const Message& message, std::string&
receipt_handle,
std::chrono::milliseconds duration,
- const ChangeInvisibleDurationCallback callback);
+ ChangeInvisibleDurationCallback callback);
void withReceiveMessageTimeout(std::chrono::milliseconds receive_timeout) {
long_polling_duration_ = receive_timeout;
