Improve Error Handling (#23)
Improve error handling
diff --git a/.clang-format b/.clang-format
index bb3e5d7..6345fe1 100644
--- a/.clang-format
+++ b/.clang-format
@@ -7,6 +7,12 @@
SortIncludes: true
IndentWidth: 2
NamespaceIndentation: Inner
+AlwaysBreakTemplateDeclarations: true
+AllowShortCaseLabelsOnASingleLine: false
+AllowShortEnumsOnASingleLine: false
+AllowShortFunctionsOnASingleLine: false
+IndentCaseBlocks: false
+IndentCaseLabels: true
...
---
diff --git a/api/BUILD.bazel b/api/BUILD.bazel
index d971e6d..8693427 100644
--- a/api/BUILD.bazel
+++ b/api/BUILD.bazel
@@ -5,4 +5,7 @@
hdrs = glob(["rocketmq/*.h"]),
strip_include_prefix = "//api",
visibility = ["//visibility:public"],
+ deps = [
+ "@com_google_absl//absl/strings",
+ ],
)
\ No newline at end of file
diff --git a/api/rocketmq/AsyncCallback.h b/api/rocketmq/AsyncCallback.h
index bb431a1..f7d5ae7 100644
--- a/api/rocketmq/AsyncCallback.h
+++ b/api/rocketmq/AsyncCallback.h
@@ -1,35 +1,34 @@
#pragma once
+#include <system_error>
+
#include "MQClientException.h"
#include "PullResult.h"
#include "SendResult.h"
ROCKETMQ_NAMESPACE_BEGIN
-struct AsyncCallback {};
-
-enum class SendCallbackType : int8_t { noAutoDeleteSendCallback = 0, autoDeleteSendCallback = 1 };
-
-using sendCallbackType = SendCallbackType;
+class AsyncCallback {
+public:
+ virtual ~AsyncCallback() = default;
+};
class SendCallback : public AsyncCallback {
public:
- virtual ~SendCallback() = default;
+ ~SendCallback() override = default;
- virtual void onSuccess(SendResult& send_result) = 0;
+ virtual void onSuccess(SendResult &send_result) noexcept = 0;
- virtual void onException(const MQException& e) = 0;
-
- virtual SendCallbackType getSendCallbackType() { return SendCallbackType::noAutoDeleteSendCallback; }
+ virtual void onFailure(const std::error_code &ec) noexcept = 0;
};
class PullCallback : public AsyncCallback {
public:
- virtual ~PullCallback() = default;
+ ~PullCallback() override = default;
- virtual void onSuccess(const PullResult& pull_result) = 0;
+ virtual void onSuccess(const PullResult &pull_result) noexcept = 0;
- virtual void onException(const MQException& e) = 0;
+ virtual void onFailure(const std::error_code &ec) noexcept = 0;
};
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/api/rocketmq/DefaultMQProducer.h b/api/rocketmq/DefaultMQProducer.h
index 94532d1..4110fd2 100644
--- a/api/rocketmq/DefaultMQProducer.h
+++ b/api/rocketmq/DefaultMQProducer.h
@@ -2,10 +2,12 @@
#include <chrono>
#include <memory>
+#include <system_error>
#include <vector>
#include "AsyncCallback.h"
#include "CredentialsProvider.h"
+#include "ErrorCode.h"
#include "LocalTransactionStateChecker.h"
#include "Logger.h"
#include "MQMessage.h"
@@ -42,18 +44,31 @@
*/
void setSendMsgTimeout(std::chrono::milliseconds timeout);
- SendResult send(const MQMessage& message, const std::string& message_group);
+ SendResult send(MQMessage& message, const std::string& message_group);
/**
* Send message in synchronous manner.
* @param message Message to send.
* @param filter_active_broker Do NOT rely on this parameter. it has been deprecated.
*/
- SendResult send(const MQMessage& message, bool filter_active_broker = false);
- SendResult send(const MQMessage& message, const MQMessageQueue& message_queue);
- SendResult send(const MQMessage& message, MessageQueueSelector* selector, void* arg);
+ SendResult send(const MQMessage& message, bool filter_active_broker = true);
- SendResult send(const MQMessage& message, MessageQueueSelector* selector, void* arg, int retry_times,
+ SendResult send(const MQMessage& message, std::error_code& ec) noexcept;
+
+ SendResult send(MQMessage& message, const MQMessageQueue& message_queue);
+ SendResult send(MQMessage& message, MessageQueueSelector* selector, void* arg);
+
+ /**
+ * @brief This function is deprecated.
+ *
+ * @param message
+ * @param selector
+ * @param arg
+ * @param retry_times retry_times is ignored with respect to member retry_times setting.
+ * @param select_active_broker
+ * @return SendResult
+ */
+ SendResult send(MQMessage& message, MessageQueueSelector* selector, void* arg, int retry_times,
bool select_active_broker = false);
/**
@@ -63,8 +78,8 @@
* @param select_active_broker Do NOT rely on this parameter. it has been deprecated.
*/
void send(const MQMessage& message, SendCallback* send_callback, bool select_active_broker = false);
- void send(const MQMessage& message, const MQMessageQueue& message_queue, SendCallback* send_callback);
- void send(const MQMessage& message, MessageQueueSelector* selector, void* arg, SendCallback* send_callback);
+ void send(MQMessage& message, const MQMessageQueue& message_queue, SendCallback* send_callback);
+ void send(MQMessage& message, MessageQueueSelector* selector, void* arg, SendCallback* send_callback);
/**
* send message in Oneway(The implementation is simply ignore the result of send message in synchronous).
@@ -72,8 +87,8 @@
* @param select_active_broker Do NOT rely on this parameter. it has been deprecated.
*/
void sendOneway(const MQMessage& message, bool select_active_broker = false);
- void sendOneway(const MQMessage& message, const MQMessageQueue& message_queue);
- void sendOneway(const MQMessage& message, MessageQueueSelector* selector, void* arg);
+ void sendOneway(MQMessage& message, const MQMessageQueue& message_queue);
+ void sendOneway(MQMessage& message, MessageQueueSelector* selector, void* arg);
void setLocalTransactionStateChecker(LocalTransactionStateCheckerPtr checker);
diff --git a/api/rocketmq/ErrorCategory.h b/api/rocketmq/ErrorCategory.h
new file mode 100644
index 0000000..316db57
--- /dev/null
+++ b/api/rocketmq/ErrorCategory.h
@@ -0,0 +1,22 @@
+#pragma once
+
+#include "ErrorCode.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+class ErrorCategory : public std::error_category {
+public:
+ static const ErrorCategory& instance() {
+ static ErrorCategory instance;
+ return instance;
+ }
+
+ const char* name() const noexcept override { return "RocketMQ"; }
+
+ std::string message(int code) const override;
+
+private:
+ ErrorCategory() = default;
+};
+
+ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/api/rocketmq/ErrorCode.h b/api/rocketmq/ErrorCode.h
index fa6cdf0..0e09dd6 100644
--- a/api/rocketmq/ErrorCode.h
+++ b/api/rocketmq/ErrorCode.h
@@ -2,26 +2,152 @@
#include "RocketMQ.h"
+#include <cstdint>
+#include <system_error>
+#include <type_traits>
+
ROCKETMQ_NAMESPACE_BEGIN
-constexpr int NO_TOPIC_ROUTE_INFO = -1;
+enum class ErrorCode : int {
+ Success = 0,
-constexpr int FAILED_TO_SELECT_MESSAGE_QUEUE = -2;
+ /**
+ * @brief Client state not as expected. Call Producer#start() first.
+ *
+ */
+ IllegalState = 1,
-constexpr int FAILED_TO_RESOLVE_BROKER_ADDRESS_FROM_TOPIC_ROUTE = -3;
+ /**
+ * @brief Bad configuration. For example, negative max-attempt-times.
+ *
+ */
+ BadConfiguration = 300,
-constexpr int FAILED_TO_SEND_MESSAGE = -4;
+ /**
+ * @brief The server cannot process the request due to apprent client-side
+ * error. For example, topic contains invalid character or is excessively
+ * long.
+ *
+ */
+ BadRequest = 400,
-constexpr int FAILED_TO_POP_MESSAGE_ASYNCHRONOUSLY = -5;
+ /**
+ * @brief Authentication failed. Possibly caused by invalid credentials.
+ *
+ */
+ Unauthorized = 401,
-constexpr int ILLEGAL_STATE = -6;
+ /**
+ * @brief Credentials are understood by server but authenticated user does not
+ * have privilege to perform the requested action.
+ *
+ */
+ Forbidden = 403,
-constexpr int MESSAGE_ILLEGAL = -7;
+ /**
+ * @brief Topic not found, which should be created through console or
+ * administration API before hand.
+ *
+ */
+ NotFound = 404,
-constexpr int BAD_CONFIGURATION = -8;
+ /**
+ * @brief Timeout when connecting, reading from or writing to brokers.
+ *
+ */
+ RequestTimeout = 408,
-constexpr int MESSAGE_QUEUE_ILLEGAL = -9;
+ /**
+ * @brief Message body is too large.
+ *
+ */
+ PayloadTooLarge = 413,
-constexpr int ERR_INVALID_MAX_ATTEMPT_TIME = -10;
+ /**
+ * @brief When trying to perform an action whose dependent procedure state is
+ * not right, this code will be used.
+ * 1. Acknowledge a message that is not previously received;
+ * 2. Commit/Rollback a transactional message that does not exist;
+ * 3. Commit an offset which is greater than maximum of partition;
+ */
+ PreconditionRequired = 428,
-ROCKETMQ_NAMESPACE_END
\ No newline at end of file
+ /**
+ * @brief Quota exchausted. The user has sent too many requests in a given
+ * amount of time.
+ *
+ */
+ TooManyRequest = 429,
+
+ /**
+ * @brief The server is unwilling to process the request because either an
+ * individual header field, or all the header fields collectively, are too
+ * large
+ *
+ */
+ HeaderFieldsTooLarge = 431,
+
+ /**
+ * @brief A server operator has received a legal demand to deny access to a
+ * resource or to a set of resources that includes the requested resource.
+ *
+ */
+ UnavailableForLegalReasons = 451,
+
+ /**
+ * @brief Server side interval error
+ *
+ */
+ InternalServerError = 500,
+
+ /**
+ * @brief The server either does not recognize the request method, or it lacks
+ * the ability to fulfil the request.
+ *
+ */
+ NotImplemented = 501,
+
+ /**
+ * @brief The server was acting as a gateway or proxy and received an invalid
+ * response from the upstream server.
+ *
+ */
+ BadGateway = 502,
+
+ /**
+ * @brief The server cannot handle the request (because it is overloaded or
+ * down for maintenance). Generally, this is a temporary state.
+ *
+ */
+ ServiceUnavailable = 503,
+
+ /**
+ * @brief The server was acting as a gateway or proxy and did not receive a
+ * timely response from the upstream server.
+ *
+ */
+ GatewayTimeout = 504,
+
+ /**
+ * @brief The server does not support the protocol version used in the
+ * request.
+ *
+ */
+ ProtocolVersionNotSupported = 505,
+
+ /**
+ * @brief The server is unable to store the representation needed to complete
+ * the request.
+ *
+ */
+ InsufficientStorage = 507,
+};
+
+std::error_code make_error_code(ErrorCode code);
+
+ROCKETMQ_NAMESPACE_END
+
+namespace std {
+template <>
+struct is_error_code_enum<ROCKETMQ_NAMESPACE::ErrorCode> : true_type {};
+} // namespace std
\ No newline at end of file
diff --git a/api/rocketmq/MQMessage.h b/api/rocketmq/MQMessage.h
index 23f4169..30dac59 100644
--- a/api/rocketmq/MQMessage.h
+++ b/api/rocketmq/MQMessage.h
@@ -7,6 +7,9 @@
#include <unordered_map>
#include <vector>
+#include "absl/strings/string_view.h"
+
+#include "MQMessageQueue.h"
#include "MessageType.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -18,64 +21,82 @@
class MQMessage {
public:
MQMessage();
- MQMessage(const std::string& topic, const std::string& body);
- MQMessage(const std::string& topic, const std::string& tags, const std::string& body);
- MQMessage(const std::string& topic, const std::string& tags, const std::string& keys, const std::string& body);
+ MQMessage(const std::string &topic, const std::string &body);
+ MQMessage(const std::string &topic, const std::string &tags,
+ const std::string &body);
+ MQMessage(const std::string &topic, const std::string &tags,
+ const std::string &keys, const std::string &body);
virtual ~MQMessage();
- MQMessage(const MQMessage& other);
- MQMessage& operator=(const MQMessage& other);
+ MQMessage(const MQMessage &other);
+ MQMessage &operator=(const MQMessage &other);
- const std::string& getMsgId() const;
+ const std::string &getMsgId() const;
- void setProperty(const std::string& name, const std::string& value);
- std::string getProperty(const std::string& name) const;
+ void setProperty(const std::string &name, const std::string &value);
+ std::string getProperty(const std::string &name) const;
- const std::string& getTopic() const;
- void setTopic(const std::string& topic);
- void setTopic(const char* data, int len);
+ const std::string &getTopic() const;
+ void setTopic(const std::string &topic);
+ void setTopic(const char *data, int len);
std::string getTags() const;
- void setTags(const std::string& tags);
+ void setTags(const std::string &tags);
- const std::vector<std::string>& getKeys() const;
+ const std::vector<std::string> &getKeys() const;
/**
* @brief Add a unique key for the message
- * TODO: a message may be associated with multiple keys. setKey, actually mean attach the given key to the message.
- * Better rename it.
+ * TODO: a message may be associated with multiple keys. setKey, actually mean
+ * attach the given key to the message. Better rename it.
* @param key Unique key in perspective of bussiness logic.
*/
- void setKey(const std::string& key);
- void setKeys(const std::vector<std::string>& keys);
+ void setKey(const std::string &key);
+ void setKeys(const std::vector<std::string> &keys);
int getDelayTimeLevel() const;
void setDelayTimeLevel(int level);
- const std::string& traceContext() const;
- void traceContext(const std::string& trace_context);
+ const std::string &traceContext() const;
+ void traceContext(const std::string &trace_context);
std::string getBornHost() const;
std::chrono::system_clock::time_point deliveryTimestamp() const;
- const std::string& getBody() const;
- void setBody(const char* data, int len);
- void setBody(const std::string& body);
+ const std::string &getBody() const;
+ void setBody(const char *data, int len);
+ void setBody(const std::string &body);
uint32_t bodyLength() const;
- const std::map<std::string, std::string>& getProperties() const;
- void setProperties(const std::map<std::string, std::string>& properties);
+ const std::map<std::string, std::string> &getProperties() const;
+ void setProperties(const std::map<std::string, std::string> &properties);
void messageType(MessageType message_type);
MessageType messageType() const;
+ void bindMessageGroup(absl::string_view message_group) {
+ message_group_ = std::string(message_group.data(), message_group.length());
+ }
+
+ void bindMessageQueue(const MQMessageQueue &message_queue) {
+ message_queue_ = message_queue;
+ }
+
+ const std::string &messageGroup() const { return message_group_; }
+
+ const MQMessageQueue &messageQueue() const { return message_queue_; }
+
protected:
- MessageImpl* impl_;
+ MessageImpl *impl_;
friend class MessageAccessor;
+
+private:
+ std::string message_group_;
+ MQMessageQueue message_queue_;
};
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/api/rocketmq/MQMessageQueue.h b/api/rocketmq/MQMessageQueue.h
index d7d9070..4beb347 100644
--- a/api/rocketmq/MQMessageQueue.h
+++ b/api/rocketmq/MQMessageQueue.h
@@ -13,18 +13,19 @@
MQMessageQueue() = default;
MQMessageQueue(std::string topic, std::string broker_name, int queue_id)
- : topic_(std::move(topic)), broker_name_(std::move(broker_name)), queue_id_(queue_id) {}
+ : topic_(std::move(topic)), broker_name_(std::move(broker_name)),
+ queue_id_(queue_id) {}
- MQMessageQueue(const MQMessageQueue& other) { this->operator=(other); }
+ MQMessageQueue(const MQMessageQueue &other) { this->operator=(other); }
- MQMessageQueue(MQMessageQueue&& other) noexcept {
+ MQMessageQueue(MQMessageQueue &&other) noexcept {
topic_ = std::move(other.topic_);
broker_name_ = std::move(other.broker_name_);
queue_id_ = other.queue_id_;
service_address_ = other.service_address_;
}
- MQMessageQueue& operator=(const MQMessageQueue& other) {
+ MQMessageQueue &operator=(const MQMessageQueue &other) {
if (this == &other) {
return *this;
}
@@ -36,27 +37,31 @@
return *this;
}
- const std::string& getTopic() const { return topic_; }
+ const std::string &getTopic() const { return topic_; }
- void setTopic(const std::string& topic) { topic_ = topic; }
+ void setTopic(const std::string &topic) { topic_ = topic; }
- const std::string& getBrokerName() const { return broker_name_; }
+ const std::string &getBrokerName() const { return broker_name_; }
- void setBrokerName(const std::string& broker_name) { broker_name_ = broker_name; }
+ void setBrokerName(const std::string &broker_name) {
+ broker_name_ = broker_name;
+ }
int getQueueId() const { return queue_id_; }
void setQueueId(int queue_id) { queue_id_ = queue_id; }
- bool operator!=(const MQMessageQueue& rhs) const {
- return topic_ != rhs.topic_ || broker_name_ != rhs.broker_name_ || queue_id_ != rhs.queue_id_;
+ bool operator!=(const MQMessageQueue &rhs) const {
+ return topic_ != rhs.topic_ || broker_name_ != rhs.broker_name_ ||
+ queue_id_ != rhs.queue_id_;
}
- bool operator==(const MQMessageQueue& mq) const {
- return topic_ == mq.topic_ && broker_name_ == mq.broker_name_ && queue_id_ == mq.queue_id_;
+ bool operator==(const MQMessageQueue &mq) const {
+ return topic_ == mq.topic_ && broker_name_ == mq.broker_name_ &&
+ queue_id_ == mq.queue_id_;
}
- bool operator<(const MQMessageQueue& mq) const {
+ bool operator<(const MQMessageQueue &mq) const {
if (topic_ != mq.topic_) {
return topic_ < mq.topic_;
}
@@ -68,7 +73,11 @@
return queue_id_ < mq.queue_id_;
}
- int compareTo(const MQMessageQueue& mq) const {
+ operator bool() const {
+ return !topic_.empty() && !broker_name_.empty() && queue_id_ >= 0;
+ }
+
+ int compareTo(const MQMessageQueue &mq) const {
if (this == &mq) {
return 0;
}
@@ -84,21 +93,22 @@
return 0;
}
- std::string simpleName() const { return topic_ + "_" + broker_name_ + "_" + std::to_string(queue_id_); }
+ std::string simpleName() const {
+ return topic_ + "_" + broker_name_ + "_" + std::to_string(queue_id_);
+ }
std::string toString() const {
std::stringstream ss;
- ss << "MessageQueue [topic=" << topic_ << ", brokerName=" << broker_name_ << ", queueId=" << queue_id_ << "]";
+ ss << "MessageQueue [topic=" << topic_ << ", brokerName=" << broker_name_
+ << ", queueId=" << queue_id_ << "]";
return ss.str();
}
- template <typename H> friend H AbslHashValue(H h, const MQMessageQueue& mq) {
+ template <typename H> friend H AbslHashValue(H h, const MQMessageQueue &mq) {
return H::combine(std::move(h), mq.topic_, mq.broker_name_, mq.queue_id_);
}
- const std::string& serviceAddress() const {
- return service_address_;
- }
+ const std::string &serviceAddress() const { return service_address_; }
void serviceAddress(std::string service_address) {
service_address_ = std::move(service_address);
diff --git a/api/rocketmq/SendResult.h b/api/rocketmq/SendResult.h
index 961d229..554827c 100644
--- a/api/rocketmq/SendResult.h
+++ b/api/rocketmq/SendResult.h
@@ -82,6 +82,8 @@
trace_context_ = std::move(trace_context);
}
+ operator bool() { return !message_id_.empty(); }
+
private:
SendStatus send_status_{SendStatus::SEND_OK};
std::string message_id_;
diff --git a/example/rocketmq/ExampleAsyncProducer.cpp b/example/rocketmq/ExampleAsyncProducer.cpp
index a40dd02..78e23e5 100644
--- a/example/rocketmq/ExampleAsyncProducer.cpp
+++ b/example/rocketmq/ExampleAsyncProducer.cpp
@@ -1,4 +1,5 @@
#include "rocketmq/DefaultMQProducer.h"
+#include "rocketmq/ErrorCode.h"
#include <algorithm>
#include <array>
#include <atomic>
@@ -6,6 +7,7 @@
#include <iostream>
#include <mutex>
#include <random>
+#include <system_error>
using namespace rocketmq;
@@ -41,7 +43,8 @@
return result;
}
-template <int PARTITION> class RateLimiter {
+template <int PARTITION>
+class RateLimiter {
public:
explicit RateLimiter(int permit) : permits_{0}, interval_(1000 / PARTITION), stopped_(false) {
int avg = permit / PARTITION;
@@ -137,9 +140,9 @@
public:
SampleSendCallback(std::atomic_int& counter, std::atomic_int& error) : counter_(counter), error_(error) {}
- void onSuccess(SendResult& send_result) override { counter_.fetch_add(1, std::memory_order_relaxed); }
+ void onSuccess(SendResult& send_result) noexcept override { counter_.fetch_add(1, std::memory_order_relaxed); }
- void onException(const MQException& e) override { error_.fetch_add(1, std::memory_order_relaxed); }
+ void onFailure(const std::error_code& ec) noexcept override { error_.fetch_add(1, std::memory_order_relaxed); }
private:
std::atomic_int& counter_;
diff --git a/src/main/cpp/base/ErrorCategory.cpp b/src/main/cpp/base/ErrorCategory.cpp
new file mode 100644
index 0000000..9d40725
--- /dev/null
+++ b/src/main/cpp/base/ErrorCategory.cpp
@@ -0,0 +1,89 @@
+#include "rocketmq/ErrorCategory.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+std::string ErrorCategory::message(int code) const {
+ ErrorCode ec = static_cast<ErrorCode>(code);
+ switch (ec) {
+ case ErrorCode::Success:
+ return "Success";
+
+ case ErrorCode::IllegalState:
+ return "Client state illegal. Forgot to call start()?";
+
+ case ErrorCode::BadConfiguration:
+ return "Bad configuration.";
+
+ case ErrorCode::BadRequest:
+ return "Message is ill-formed. Check validity of your topic, tag, "
+ "etc";
+
+ case ErrorCode::Unauthorized:
+ return "Authentication failed. Possibly caused by invalid credentials.";
+
+ case ErrorCode::Forbidden:
+ return "Authenticated user does not have privilege to perform the "
+ "requested action";
+
+ case ErrorCode::NotFound:
+ return "Topic not found, which should be created through console or "
+ "administration API before hand.";
+
+ case ErrorCode::RequestTimeout:
+ return "Timeout when connecting, reading from or writing to brokers.";
+
+ case ErrorCode::PayloadTooLarge:
+ return "Message body is too large.";
+
+ case ErrorCode::PreconditionRequired:
+ return "State of dependent procedure is not right";
+
+ case ErrorCode::TooManyRequest:
+ return "Quota exchausted. The user has sent too many requests in a given "
+ "amount of time.";
+
+ case ErrorCode::UnavailableForLegalReasons:
+ return "A server operator has received a legal demand to deny access to "
+ "a resource or to a set of resources that "
+ "includes the requested resource.";
+
+ case ErrorCode::HeaderFieldsTooLarge:
+ return "The server is unwilling to process the request because either an "
+ "individual header field, or all the header fields collectively, "
+ "are too large";
+
+ case ErrorCode::InternalServerError:
+ return "Server side interval error";
+
+ case ErrorCode::NotImplemented:
+ return "The server either does not recognize the request method, or it "
+ "lacks the ability to fulfil the request.";
+
+ case ErrorCode::BadGateway:
+ return "The server was acting as a gateway or proxy and received an "
+ "invalid response from the upstream server.";
+
+ case ErrorCode::ServiceUnavailable:
+ return "The server cannot handle the request (because it is overloaded "
+ "or down for maintenance). Generally, this "
+ "is a temporary state.";
+
+ case ErrorCode::GatewayTimeout:
+ return "The server was acting as a gateway or proxy and did not receive "
+ "a timely response from the upstream "
+ "server.";
+
+ case ErrorCode::ProtocolVersionNotSupported:
+ return "The server does not support the protocol version used in the "
+ "request.";
+
+ case ErrorCode::InsufficientStorage:
+ return "The server is unable to store the representation needed to "
+ "complete the request.";
+
+ default:
+ return "Not-Implemented";
+ }
+}
+
+ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/base/ErrorCode.cpp b/src/main/cpp/base/ErrorCode.cpp
new file mode 100644
index 0000000..b6e44dc
--- /dev/null
+++ b/src/main/cpp/base/ErrorCode.cpp
@@ -0,0 +1,10 @@
+#include "rocketmq/ErrorCategory.h"
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+std::error_code make_error_code(ErrorCode code) {
+ const ErrorCategory& instance = ErrorCategory::instance();
+ return {static_cast<int>(code), instance};
+}
+
+ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/client/ClientManagerImpl.cpp b/src/main/cpp/client/ClientManagerImpl.cpp
index a86ff5e..679ce3f 100644
--- a/src/main/cpp/client/ClientManagerImpl.cpp
+++ b/src/main/cpp/client/ClientManagerImpl.cpp
@@ -1,5 +1,14 @@
#include "ClientManagerImpl.h"
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <system_error>
+#include <utility>
+#include <vector>
+
+#include "google/rpc/code.pb.h"
+
#include "InvocationContext.h"
#include "LogInterceptor.h"
#include "LogInterceptorFactory.h"
@@ -16,11 +25,6 @@
#include "grpcpp/create_channel.h"
#include "rocketmq/ErrorCode.h"
#include "rocketmq/MQMessageExt.h"
-#include <chrono>
-#include <google/rpc/code.pb.h>
-#include <memory>
-#include <utility>
-#include <vector>
ROCKETMQ_NAMESPACE_BEGIN
@@ -155,18 +159,14 @@
void ClientManagerImpl::healthCheck(
const std::string& target_host, const Metadata& metadata, const HealthCheckRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(const std::string&, const InvocationContext<HealthCheckResponse>*)>& cb) {
- {
- absl::MutexLock lk(&rpc_clients_mtx_);
- if (!rpc_clients_.contains(target_host)) {
- SPDLOG_WARN("Try to perform health check for {}, which is unknown to client manager", target_host);
- cb(target_host, nullptr);
- return;
- }
- }
-
+ const std::function<void(const std::error_code&, const InvocationContext<HealthCheckResponse>*)>& cb) {
+ std::error_code ec;
auto client = getRpcClient(target_host);
- assert(client);
+ if (!client) {
+ ec = ErrorCode::RequestTimeout;
+ cb(ec, nullptr);
+ return;
+ }
auto invocation_context = new InvocationContext<HealthCheckResponse>();
invocation_context->remote_address = target_host;
@@ -176,7 +176,41 @@
invocation_context->context.AddMetadata(entry.first, entry.second);
}
- auto callback = [cb](const InvocationContext<HealthCheckResponse>* ctx) { cb(ctx->remote_address, ctx); };
+ auto callback = [cb](const InvocationContext<HealthCheckResponse>* ctx) {
+ std::error_code ec;
+ if (!ctx->status.ok()) {
+ ec = ErrorCode::RequestTimeout;
+ cb(ec, ctx);
+ return;
+ }
+
+ const auto& common = ctx->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ cb(ec, ctx);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ ec = ErrorCode::Unauthorized;
+ cb(ec, ctx);
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ ec = ErrorCode::Forbidden;
+ cb(ec, ctx);
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}", common.status().message());
+ ec = ErrorCode::InternalServerError;
+ cb(ec, ctx);
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplemented: please upgrade SDK to latest release");
+ ec = ErrorCode::NotImplemented;
+ cb(ec, ctx);
+ } break;
+ }
+ };
invocation_context->callback = callback;
client->asyncHealthCheck(request, invocation_context);
@@ -190,7 +224,7 @@
return;
}
- cleanOfflineRpcClients();
+ auto&& rpc_clients_removed = cleanOfflineRpcClients();
std::vector<std::shared_ptr<Client>> clients;
{
@@ -203,13 +237,19 @@
}
}
+ if (!rpc_clients_removed.empty()) {
+ for (auto& client : clients) {
+ client->onRemoteEndpointRemoval(rpc_clients_removed);
+ }
+ }
+
for (auto& client : clients) {
client->healthCheck();
}
SPDLOG_DEBUG("Health check completed");
}
-void ClientManagerImpl::cleanOfflineRpcClients() {
+std::vector<std::string> ClientManagerImpl::cleanOfflineRpcClients() {
absl::flat_hash_set<std::string> hosts;
{
absl::MutexLock lk(&clients_mtx_);
@@ -222,23 +262,27 @@
}
}
+ std::vector<std::string> removed;
{
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)) {
SPDLOG_INFO("Removed RPC client whose peer is offline. RemoteHost={}", host);
+ removed.push_back(host);
rpc_clients_.erase(it++);
} else {
it++;
}
}
}
+
+ return removed;
}
void ClientManagerImpl::heartbeat(const std::string& target_host, const Metadata& metadata,
const HeartbeatRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const HeartbeatResponse&)>& cb) {
+ const std::function<void(const std::error_code&, const HeartbeatResponse&)>& cb) {
auto client = getRpcClient(target_host, true);
if (!client) {
return;
@@ -251,22 +295,47 @@
}
auto callback = [cb](const InvocationContext<HeartbeatResponse>* invocation_context) {
- if (invocation_context->status.ok()) {
- if (google::rpc::Code::OK == invocation_context->response.common().status().code()) {
- SPDLOG_DEBUG("Send heartbeat to target_host={}, gRPC status OK", invocation_context->remote_address);
- cb(true, invocation_context->response);
- } else {
- SPDLOG_WARN("Server[{}] failed to process heartbeat. Reason: {}", invocation_context->remote_address,
- invocation_context->response.common().DebugString());
- cb(false, invocation_context->response);
- }
- } else {
- SPDLOG_WARN("Failed to send heartbeat to target_host={}. GRPC code: {}, message : {}",
+ if (!invocation_context->status.ok()) {
+ SPDLOG_WARN("Failed to send heartbeat to target_host={}. gRPC code: {}, message: {}",
invocation_context->remote_address, invocation_context->status.error_code(),
invocation_context->status.error_message());
- cb(false, invocation_context->response);
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb(ec, invocation_context->response);
+ return;
+ }
+
+ const auto& common = invocation_context->response.common();
+ std::error_code ec;
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ ec = ErrorCode::Unauthorized;
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ ec = ErrorCode::Forbidden;
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}", common.status().message());
+ ec = ErrorCode::BadRequest;
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}", common.status().message());
+ ec = ErrorCode::InternalServerError;
+ cb(ec, invocation_context->response);
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplemented: Please upgrade SDK to latest release");
+ } break;
}
};
+
invocation_context->callback = callback;
invocation_context->context.set_deadline(std::chrono::system_clock::now() + timeout);
client->asyncHeartbeat(request, invocation_context);
@@ -296,7 +365,6 @@
}
void ClientManagerImpl::pollCompletionQueue() {
-
while (State::STARTED == state_.load(std::memory_order_relaxed) ||
State::STARTING == state_.load(std::memory_order_relaxed)) {
bool ok = false;
@@ -328,40 +396,65 @@
}
const std::string& topic = request.message().topic().name();
- auto completion_callback = [topic, cb, this](const InvocationContext<SendMessageResponse>* invocation_context) {
- if (invocation_context->status.ok() &&
- google::rpc::Code::OK == invocation_context->response.common().status().code()) {
- SendResult send_result;
- send_result.setSendStatus(SendStatus::SEND_OK);
- send_result.setMsgId(invocation_context->response.message_id());
- send_result.setTransactionId(invocation_context->response.transaction_id());
- if (State::STARTED == state_.load(std::memory_order_relaxed)) {
- cb->onSuccess(send_result);
- } else {
- SPDLOG_INFO("Client instance has stopped, state={}. Message[MessageId={}] ignored",
- state_.load(std::memory_order_relaxed), send_result.getMsgId());
- }
- } else {
- if (!invocation_context->status.ok()) {
- SPDLOG_WARN("Failed to send message to {} due to gRPC error. gRPC code: {}, gRPC error message: {}",
- invocation_context->remote_address, invocation_context->status.error_code(),
- invocation_context->status.error_message());
- }
- std::string msg;
- msg.append("gRPC code: ")
- .append(std::to_string(invocation_context->status.error_code()))
- .append(", gRPC message: ")
- .append(invocation_context->status.error_message())
- .append(", code: ")
- .append(std::to_string(invocation_context->response.common().status().code()))
- .append(", remark: ")
- .append(invocation_context->response.common().DebugString());
- MQException e(msg, FAILED_TO_SEND_MESSAGE, __FILE__, __LINE__);
- if (State::STARTED == state_.load(std::memory_order_relaxed)) {
- cb->onException(e);
- } else {
- SPDLOG_WARN("Client instance has stopped, state={}. Ignore exception raised while sending message: {}",
- state_.load(std::memory_order_relaxed), e.what());
+ std::weak_ptr<ClientManager> client_manager(shared_from_this());
+ auto completion_callback = [topic, cb,
+ client_manager](const InvocationContext<SendMessageResponse>* invocation_context) {
+ ClientManagerPtr client_manager_ptr = client_manager.lock();
+ if (!client_manager_ptr) {
+ return;
+ }
+
+ if (State::STARTED != client_manager_ptr->state()) {
+ // TODO: Would this leak some memroy?
+ return;
+ }
+
+ const auto& common = invocation_context->response.common();
+
+ if (!invocation_context->status.ok()) {
+ SPDLOG_WARN("Failed to send message to {} due to gRPC error. gRPC code: {}, gRPC error message: {}",
+ invocation_context->remote_address, invocation_context->status.error_code(),
+ invocation_context->status.error_message());
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb->onFailure(ec);
+ return;
+ }
+
+ if (invocation_context->status.ok()) {
+ switch (invocation_context->response.common().status().code()) {
+ case google::rpc::Code::OK: {
+ SendResult send_result;
+ send_result.setSendStatus(SendStatus::SEND_OK);
+ send_result.setMsgId(invocation_context->response.message_id());
+ send_result.setTransactionId(invocation_context->response.transaction_id());
+ cb->onSuccess(send_result);
+ } break;
+
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}", common.status().message());
+ std::error_code ec = ErrorCode::BadRequest;
+ cb->onFailure(ec);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ std::error_code ec = ErrorCode::Unauthorized;
+ cb->onFailure(ec);
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ std::error_code ec = ErrorCode::Forbidden;
+ cb->onFailure(ec);
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}", common.status().message());
+ std::error_code ec = ErrorCode::InternalServerError;
+ cb->onFailure(ec);
+ } break;
+ default: {
+ SPDLOG_WARN("Unsupported status code. Check and upgrade SDK to the latest");
+ std::error_code ec = ErrorCode::NotImplemented;
+ cb->onFailure(ec);
+ } break;
}
}
};
@@ -447,12 +540,13 @@
void ClientManagerImpl::resolveRoute(const std::string& target_host, const Metadata& metadata,
const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr&)>& cb) {
+ const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>& cb) {
RpcClientSharedPtr client = getRpcClient(target_host, false);
if (!client) {
SPDLOG_WARN("Failed to create RPC client for name server[host={}]", target_host);
- cb(false, nullptr);
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb(ec, nullptr);
return;
}
@@ -467,94 +561,144 @@
if (!invocation_context->status.ok()) {
SPDLOG_WARN("Failed to send query route request to server[host={}]. Reason: {}",
invocation_context->remote_address, invocation_context->status.error_message());
- cb(false, nullptr);
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb(ec, nullptr);
return;
}
- if (google::rpc::Code::OK != invocation_context->response.common().status().code()) {
- SPDLOG_WARN("Server[host={}] failed to process query route request. Reason: {}",
- invocation_context->remote_address, invocation_context->response.common().DebugString());
- cb(false, nullptr);
- return;
+ std::error_code ec;
+ const auto& common = invocation_context->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ auto& partitions = invocation_context->response.partitions();
+ std::vector<Partition> topic_partitions;
+ for (const auto& partition : partitions) {
+ Topic t(partition.topic().resource_namespace(), partition.topic().name());
+
+ auto& broker = partition.broker();
+ AddressScheme scheme = AddressScheme::IPv4;
+ switch (broker.endpoints().scheme()) {
+ case rmq::AddressScheme::IPv4:
+ scheme = AddressScheme::IPv4;
+ break;
+ case rmq::AddressScheme::IPv6:
+ scheme = AddressScheme::IPv6;
+ break;
+ case rmq::AddressScheme::DOMAIN_NAME:
+ scheme = AddressScheme::DOMAIN_NAME;
+ break;
+ default:
+ break;
+ }
+
+ std::vector<Address> addresses;
+ for (const auto& address : broker.endpoints().addresses()) {
+ addresses.emplace_back(Address{address.host(), address.port()});
+ }
+ ServiceAddress service_address(scheme, addresses);
+ Broker b(partition.broker().name(), partition.broker().id(), service_address);
+
+ Permission permission = Permission::READ_WRITE;
+ switch (partition.permission()) {
+ case rmq::Permission::READ:
+ permission = Permission::READ;
+ break;
+
+ case rmq::Permission::WRITE:
+ permission = Permission::WRITE;
+ break;
+ case rmq::Permission::READ_WRITE:
+ permission = Permission::READ_WRITE;
+ break;
+ default:
+ break;
+ }
+ Partition topic_partition(t, partition.id(), permission, std::move(b));
+ topic_partitions.emplace_back(std::move(topic_partition));
+ }
+ auto ptr =
+ std::make_shared<TopicRouteData>(std::move(topic_partitions), invocation_context->response.DebugString());
+ cb(ec, ptr);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ ec = ErrorCode::Unauthorized;
+ cb(ec, nullptr);
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ ec = ErrorCode::Forbidden;
+ cb(ec, nullptr);
+ } break;
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}", common.status().message());
+ ec = ErrorCode::BadRequest;
+ cb(ec, nullptr);
+ } break;
+ case google::rpc::Code::NOT_FOUND: {
+ SPDLOG_WARN("NotFound: {}", common.status().message());
+ ec = ErrorCode::NotFound;
+ cb(ec, nullptr);
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}", common.status().message());
+ ec = ErrorCode::InternalServerError;
+ cb(ec, nullptr);
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplement: Please upgrade to latest SDK release");
+ ec = ErrorCode::NotImplemented;
+ cb(ec, nullptr);
+ } break;
}
-
- auto& partitions = invocation_context->response.partitions();
-
- std::vector<Partition> topic_partitions;
- for (const auto& partition : partitions) {
- Topic t(partition.topic().resource_namespace(), partition.topic().name());
-
- auto& broker = partition.broker();
- AddressScheme scheme = AddressScheme::IPv4;
- switch (broker.endpoints().scheme()) {
- case rmq::AddressScheme::IPv4:
- scheme = AddressScheme::IPv4;
- break;
- case rmq::AddressScheme::IPv6:
- scheme = AddressScheme::IPv6;
- break;
- case rmq::AddressScheme::DOMAIN_NAME:
- scheme = AddressScheme::DOMAIN_NAME;
- break;
- default:
- break;
- }
-
- std::vector<Address> addresses;
- for (const auto& address : broker.endpoints().addresses()) {
- addresses.emplace_back(Address{address.host(), address.port()});
- }
- ServiceAddress service_address(scheme, addresses);
- Broker b(partition.broker().name(), partition.broker().id(), service_address);
-
- Permission permission = Permission::READ_WRITE;
- switch (partition.permission()) {
- case rmq::Permission::READ:
- permission = Permission::READ;
- break;
-
- case rmq::Permission::WRITE:
- permission = Permission::WRITE;
- break;
- case rmq::Permission::READ_WRITE:
- permission = Permission::READ_WRITE;
- break;
- default:
- break;
- }
- Partition topic_partition(t, partition.id(), permission, std::move(b));
- topic_partitions.emplace_back(std::move(topic_partition));
- }
-
- auto ptr =
- std::make_shared<TopicRouteData>(std::move(topic_partitions), invocation_context->response.DebugString());
- cb(true, ptr);
};
invocation_context->callback = callback;
client->asyncQueryRoute(request, invocation_context);
}
-void ClientManagerImpl::queryAssignment(const std::string& target, const Metadata& metadata,
- const QueryAssignmentRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryAssignmentResponse&)>& cb) {
+void ClientManagerImpl::queryAssignment(
+ const std::string& target, const Metadata& metadata, const QueryAssignmentRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code&, const QueryAssignmentResponse&)>& cb) {
SPDLOG_DEBUG("Prepare to send query assignment request to broker[address={}]", target);
std::shared_ptr<RpcClient> client = getRpcClient(target);
auto callback = [&, cb](const InvocationContext<QueryAssignmentResponse>* invocation_context) {
if (!invocation_context->status.ok()) {
SPDLOG_WARN("Failed to query assignment. Reason: {}", invocation_context->status.error_message());
- cb(false, invocation_context->response);
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb(ec, invocation_context->response);
return;
}
- if (google::rpc::Code::OK != invocation_context->response.common().status().code()) {
- SPDLOG_WARN("Server[host={}] failed to process query assignment request. Reason: {}",
- invocation_context->remote_address, invocation_context->response.common().DebugString());
- cb(false, invocation_context->response);
- return;
+ const auto& common = invocation_context->response.common();
+ std::error_code ec;
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ SPDLOG_DEBUG("Query assignment OK");
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ ec = ErrorCode::Unauthorized;
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ ec = ErrorCode::Forbidden;
+ } break;
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}", common.status().message());
+ ec = ErrorCode::BadRequest;
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}", common.status().message());
+ ec = ErrorCode::InternalServerError;
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplemented: please upgrade SDK to latest release");
+ ec = ErrorCode::NotImplemented;
+ } break;
}
-
- cb(true, invocation_context->response);
+ cb(ec, invocation_context->response);
};
auto invocation_context = new InvocationContext<QueryAssignmentResponse>();
@@ -585,17 +729,57 @@
auto callback = [this, cb](const InvocationContext<ReceiveMessageResponse>* invocation_context) {
if (invocation_context->status.ok()) {
SPDLOG_DEBUG("Received pop response through gRPC from brokerAddress={}", invocation_context->remote_address);
- ReceiveMessageResult receive_result;
- this->processPopResult(invocation_context->context, invocation_context->response, receive_result,
- invocation_context->remote_address);
- cb->onSuccess(receive_result);
+ const auto& common = invocation_context->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ ReceiveMessageResult receive_result;
+ this->processPopResult(invocation_context->context, invocation_context->response, receive_result,
+ invocation_context->remote_address);
+ cb->onSuccess(receive_result);
+ } break;
+
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ std::error_code ec = ErrorCode::Unauthorized;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ std::error_code ec = ErrorCode::Forbidden;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}", common.status().message());
+ std::error_code ec = ErrorCode::BadRequest;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::DEADLINE_EXCEEDED: {
+ SPDLOG_WARN("DeadlineExceeded: {}", common.status().message());
+ std::error_code ec = ErrorCode::GatewayTimeout;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("IntervalServerError: {}", common.status().message());
+ std::error_code ec = ErrorCode::InternalServerError;
+ cb->onFailure(ec);
+ } break;
+ default: {
+ SPDLOG_WARN("Unsupported code. Please upgrade to use the latest release");
+ std::error_code ec = ErrorCode::NotImplemented;
+ cb->onFailure(ec);
+ } break;
+ }
+
} else {
SPDLOG_WARN("Failed to pop messages through GRPC from {}, gRPC code: {}, gRPC error message: {}",
invocation_context->remote_address, invocation_context->status.error_code(),
invocation_context->status.error_message());
- MQException e(invocation_context->status.error_message(), FAILED_TO_POP_MESSAGE_ASYNCHRONOUSLY, __FILE__,
- __LINE__);
- cb->onException(e);
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb->onFailure(ec);
}
};
invocation_context->callback = callback;
@@ -608,25 +792,31 @@
// process response to result
ReceiveMessageStatus status;
switch (response.common().status().code()) {
- case google::rpc::Code::OK:
- status = ReceiveMessageStatus::OK;
- break;
- case google::rpc::Code::RESOURCE_EXHAUSTED:
- status = ReceiveMessageStatus::RESOURCE_EXHAUSTED;
- SPDLOG_WARN("Too many pop requests in broker. Long polling is full in the broker side. BrokerAddress={}",
- target_host);
- break;
- case google::rpc::Code::DEADLINE_EXCEEDED:
- status = ReceiveMessageStatus::DEADLINE_EXCEEDED;
- break;
- case google::rpc::Code::NOT_FOUND:
- status = ReceiveMessageStatus::NOT_FOUND;
- break;
- default:
- SPDLOG_WARN("Pop response indicates server-side error. BrokerAddress={}, Reason={}", target_host,
- response.common().DebugString());
- status = ReceiveMessageStatus::INTERNAL;
- break;
+ case google::rpc::Code::OK: {
+ status = ReceiveMessageStatus::OK;
+ break;
+ }
+ case google::rpc::Code::RESOURCE_EXHAUSTED: {
+ status = ReceiveMessageStatus::RESOURCE_EXHAUSTED;
+ SPDLOG_WARN("Too many pop requests in broker. Long polling is full in the broker side. BrokerAddress={}",
+ target_host);
+ break;
+ }
+
+ case google::rpc::Code::DEADLINE_EXCEEDED: {
+ status = ReceiveMessageStatus::DEADLINE_EXCEEDED;
+ break;
+ }
+ case google::rpc::Code::NOT_FOUND: {
+ status = ReceiveMessageStatus::NOT_FOUND;
+ break;
+ }
+ default: {
+ SPDLOG_WARN("Pop response indicates server-side error. BrokerAddress={}, Reason={}", target_host,
+ response.common().DebugString());
+ status = ReceiveMessageStatus::INTERNAL;
+ break;
+ }
}
result.sourceHost(target_host);
@@ -667,34 +857,42 @@
}
result.sourceHost(target_host);
switch (response.common().status().code()) {
- case google::rpc::Code::OK: {
- assert(!response.messages().empty());
- result.status_ = ReceiveMessageStatus::OK;
- for (const auto& item : response.messages()) {
- MQMessageExt message;
- if (!wrapMessage(item, message)) {
- result.status_ = ReceiveMessageStatus::DATA_CORRUPTED;
- return;
+ case google::rpc::Code::OK: {
+ assert(!response.messages().empty());
+ result.status_ = ReceiveMessageStatus::OK;
+ for (const auto& item : response.messages()) {
+ MQMessageExt message;
+ if (!wrapMessage(item, message)) {
+ result.status_ = ReceiveMessageStatus::DATA_CORRUPTED;
+ return;
+ }
+ result.messages_.emplace_back(message);
}
- result.messages_.emplace_back(message);
+ break;
}
- } break;
- case google::rpc::Code::DEADLINE_EXCEEDED:
- result.status_ = ReceiveMessageStatus::DEADLINE_EXCEEDED;
- break;
+ case google::rpc::Code::DEADLINE_EXCEEDED: {
+ result.status_ = ReceiveMessageStatus::DEADLINE_EXCEEDED;
+ break;
+ }
- case google::rpc::Code::RESOURCE_EXHAUSTED:
- result.status_ = ReceiveMessageStatus::RESOURCE_EXHAUSTED;
- break;
+ case google::rpc::Code::RESOURCE_EXHAUSTED: {
+ result.status_ = ReceiveMessageStatus::RESOURCE_EXHAUSTED;
+ break;
+ }
- case google::rpc::Code::OUT_OF_RANGE:
- result.status_ = ReceiveMessageStatus::OUT_OF_RANGE;
- result.next_offset_ = response.next_offset();
- break;
+ case google::rpc::Code::OUT_OF_RANGE: {
+ result.status_ = ReceiveMessageStatus::OUT_OF_RANGE;
+ result.next_offset_ = response.next_offset();
+ break;
+ }
}
}
+State ClientManagerImpl::state() const {
+ return state_.load(std::memory_order_relaxed);
+}
+
bool ClientManagerImpl::wrapMessage(const rmq::Message& item, MQMessageExt& message_ext) {
assert(item.topic().resource_namespace() == resource_namespace_);
@@ -728,57 +926,57 @@
body_digest_match = true;
} else {
switch (digest.type()) {
- case rmq::DigestType::CRC32: {
- std::string checksum;
- bool success = MixAll::crc32(item.body(), checksum);
- if (success) {
- body_digest_match = (digest.checksum() == checksum);
- if (body_digest_match) {
- SPDLOG_DEBUG("Message body CRC32 checksum validation passed.");
+ case rmq::DigestType::CRC32: {
+ std::string checksum;
+ bool success = MixAll::crc32(item.body(), checksum);
+ if (success) {
+ body_digest_match = (digest.checksum() == checksum);
+ if (body_digest_match) {
+ SPDLOG_DEBUG("Message body CRC32 checksum validation passed.");
+ } else {
+ SPDLOG_WARN("Body CRC32 checksum validation failed. Actual: {}, expect: {}", checksum, digest.checksum());
+ }
} else {
- SPDLOG_WARN("Body CRC32 checksum validation failed. Actual: {}, expect: {}", checksum, digest.checksum());
+ SPDLOG_WARN("Failed to calculate CRC32 checksum. Skip.");
}
- } else {
- SPDLOG_WARN("Failed to calculate CRC32 checksum. Skip.");
+ break;
}
- break;
- }
- case rmq::DigestType::MD5: {
- std::string checksum;
- bool success = MixAll::md5(item.body(), checksum);
- if (success) {
- body_digest_match = (digest.checksum() == checksum);
- if (body_digest_match) {
- SPDLOG_DEBUG("MD5 checksum validation passed.");
+ case rmq::DigestType::MD5: {
+ std::string checksum;
+ bool success = MixAll::md5(item.body(), checksum);
+ if (success) {
+ body_digest_match = (digest.checksum() == checksum);
+ if (body_digest_match) {
+ SPDLOG_DEBUG("MD5 checksum validation passed.");
+ } else {
+ SPDLOG_WARN("Body MD5 checksum validation failed. Expect: {}, Actual: {}", digest.checksum(), checksum);
+ }
} else {
- SPDLOG_WARN("Body MD5 checksum validation failed. Expect: {}, Actual: {}", digest.checksum(), checksum);
+ SPDLOG_WARN("Failed to calculate MD5 digest. Skip.");
+ body_digest_match = true;
}
- } else {
- SPDLOG_WARN("Failed to calculate MD5 digest. Skip.");
+ break;
+ }
+ case rmq::DigestType::SHA1: {
+ std::string checksum;
+ bool success = MixAll::sha1(item.body(), checksum);
+ if (success) {
+ body_digest_match = (checksum == digest.checksum());
+ if (body_digest_match) {
+ SPDLOG_DEBUG("SHA1 checksum validation passed");
+ } else {
+ SPDLOG_WARN("Body SHA1 checksum validation failed. Expect: {}, Actual: {}", digest.checksum(), checksum);
+ }
+ } else {
+ SPDLOG_WARN("Failed to calculate SHA1 digest. Skip.");
+ }
+ break;
+ }
+ default: {
+ SPDLOG_WARN("Unsupported message body digest algorithm");
body_digest_match = true;
+ break;
}
- break;
- }
- case rmq::DigestType::SHA1: {
- std::string checksum;
- bool success = MixAll::sha1(item.body(), checksum);
- if (success) {
- body_digest_match = (checksum == digest.checksum());
- if (body_digest_match) {
- SPDLOG_DEBUG("SHA1 checksum validation passed");
- } else {
- SPDLOG_WARN("Body SHA1 checksum validation failed. Expect: {}, Actual: {}", digest.checksum(), checksum);
- }
- } else {
- SPDLOG_WARN("Failed to calculate SHA1 digest. Skip.");
- }
- break;
- }
- default: {
- SPDLOG_WARN("Unsupported message body digest algorithm");
- body_digest_match = true;
- break;
- }
}
}
@@ -790,20 +988,20 @@
// Body encoding
switch (system_attributes.body_encoding()) {
- case rmq::Encoding::GZIP: {
- std::string uncompressed;
- UtilAll::uncompress(item.body(), uncompressed);
- message_ext.setBody(uncompressed);
- break;
- }
- case rmq::Encoding::IDENTITY: {
- message_ext.setBody(item.body());
- break;
- }
- default: {
- SPDLOG_WARN("Unsupported encoding algorithm");
- break;
- }
+ case rmq::Encoding::GZIP: {
+ std::string uncompressed;
+ UtilAll::uncompress(item.body(), uncompressed);
+ message_ext.setBody(uncompressed);
+ break;
+ }
+ case rmq::Encoding::IDENTITY: {
+ message_ext.setBody(item.body());
+ break;
+ }
+ default: {
+ SPDLOG_WARN("Unsupported encoding algorithm");
+ break;
+ }
}
timeval tv{};
@@ -811,22 +1009,22 @@
// Message-type
MessageType message_type;
switch (system_attributes.message_type()) {
- case rmq::MessageType::NORMAL:
- message_type = MessageType::NORMAL;
- break;
- case rmq::MessageType::FIFO:
- message_type = MessageType::FIFO;
- break;
- case rmq::MessageType::DELAY:
- message_type = MessageType::DELAY;
- break;
- case rmq::MessageType::TRANSACTION:
- message_type = MessageType::TRANSACTION;
- break;
- default:
- SPDLOG_WARN("Unknown message type. Treat it as normal message");
- message_type = MessageType::NORMAL;
- break;
+ case rmq::MessageType::NORMAL:
+ message_type = MessageType::NORMAL;
+ break;
+ case rmq::MessageType::FIFO:
+ message_type = MessageType::FIFO;
+ break;
+ case rmq::MessageType::DELAY:
+ message_type = MessageType::DELAY;
+ break;
+ case rmq::MessageType::TRANSACTION:
+ message_type = MessageType::TRANSACTION;
+ break;
+ default:
+ SPDLOG_WARN("Unknown message type. Treat it as normal message");
+ message_type = MessageType::NORMAL;
+ break;
}
MessageAccessor::setMessageType(message_ext, message_type);
@@ -853,20 +1051,20 @@
// Process one-of: delivery-timestamp and delay-level.
switch (system_attributes.timed_delivery_case()) {
- case rmq::SystemAttribute::TimedDeliveryCase::kDelayLevel: {
- message_ext.setDelayTimeLevel(system_attributes.delay_level());
- break;
- }
+ case rmq::SystemAttribute::TimedDeliveryCase::kDelayLevel: {
+ message_ext.setDelayTimeLevel(system_attributes.delay_level());
+ break;
+ }
- case rmq::SystemAttribute::TimedDeliveryCase::kDeliveryTimestamp: {
- tv.tv_sec = system_attributes.delivery_timestamp().seconds();
- tv.tv_usec = system_attributes.delivery_timestamp().nanos();
- MessageAccessor::setDeliveryTimestamp(message_ext, absl::TimeFromTimeval(tv));
- break;
- }
+ case rmq::SystemAttribute::TimedDeliveryCase::kDeliveryTimestamp: {
+ tv.tv_sec = system_attributes.delivery_timestamp().seconds();
+ tv.tv_usec = system_attributes.delivery_timestamp().nanos();
+ MessageAccessor::setDeliveryTimestamp(message_ext, absl::TimeFromTimeval(tv));
+ break;
+ }
- default:
- break;
+ default:
+ break;
}
// Partition-id
@@ -908,10 +1106,12 @@
return true;
}
-Scheduler& ClientManagerImpl::getScheduler() { return scheduler_; }
+Scheduler& ClientManagerImpl::getScheduler() {
+ return scheduler_;
+}
void ClientManagerImpl::ack(const std::string& target, const Metadata& metadata, const AckMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& cb) {
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& cb) {
std::string target_host(target.data(), target.length());
SPDLOG_DEBUG("Prepare to ack message against {} asynchronously. AckMessageRequest: {}", target_host,
request.DebugString());
@@ -927,12 +1127,40 @@
// TODO: Use capture by move and pass-by-value paradigm when C++ 14 is available.
auto callback = [request, cb](const InvocationContext<AckMessageResponse>* invocation_context) {
- if (invocation_context->status.ok() &&
- google::rpc::Code::OK == invocation_context->response.common().status().code()) {
- cb(true);
- } else {
- cb(false);
+ std::error_code ec;
+ if (!invocation_context->status.ok()) {
+ ec = ErrorCode::RequestTimeout;
+ cb(ec);
+ return;
}
+
+ const auto& common = invocation_context->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ SPDLOG_DEBUG("Ack OK. host={}", invocation_context->remote_address);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Unauthorized;
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Forbidden;
+ } break;
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::BadRequest;
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::InternalServerError;
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplement: please upgrade SDK to latest release. host={}", invocation_context->remote_address);
+ ec = ErrorCode::NotImplemented;
+ } break;
+ }
+ cb(ec);
};
invocation_context->callback = callback;
client->asyncAck(request, invocation_context);
@@ -940,7 +1168,7 @@
void ClientManagerImpl::nack(const std::string& target_host, const Metadata& metadata,
const NackMessageRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool)>& completion_callback) {
+ const std::function<void(const std::error_code&)>& completion_callback) {
RpcClientSharedPtr client = getRpcClient(target_host);
assert(client);
auto invocation_context = new InvocationContext<NackMessageResponse>();
@@ -952,25 +1180,58 @@
}
auto callback = [completion_callback](const InvocationContext<NackMessageResponse>* invocation_context) {
- if (invocation_context->status.ok() &&
- google::rpc::Code::OK == invocation_context->response.common().status().code()) {
- completion_callback(true);
- } else {
- completion_callback(false);
+ if (!invocation_context->status.ok()) {
+ SPDLOG_WARN("Failed to write Nack request to wire. gRPC-code: {}, gRPC-message: {}",
+ invocation_context->status.error_code(), invocation_context->status.error_message());
+ std::error_code ec = ErrorCode::RequestTimeout;
+ completion_callback(ec);
+ return;
}
+
+ std::error_code ec;
+ const auto& common = invocation_context->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ SPDLOG_DEBUG("Nack to {} OK", invocation_context->remote_address);
+ break;
+ };
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Unauthorized;
+ break;
+ }
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Forbidden;
+ break;
+ }
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::InternalServerError;
+ break;
+ }
+ default: {
+ SPDLOG_WARN("NotImplemented: Please upgrade to latest SDK, host={}", invocation_context->remote_address);
+ ec = ErrorCode::NotImplemented;
+ break;
+ }
+ }
+ completion_callback(ec);
};
invocation_context->callback = callback;
client->asyncNack(request, invocation_context);
}
-void ClientManagerImpl::endTransaction(const std::string& target_host, const Metadata& metadata,
- const EndTransactionRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const EndTransactionResponse&)>& cb) {
+void ClientManagerImpl::endTransaction(
+ const std::string& target_host, const Metadata& metadata, const EndTransactionRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code&, const EndTransactionResponse&)>& cb) {
RpcClientSharedPtr client = getRpcClient(target_host);
if (!client) {
SPDLOG_WARN("No RPC client for {}", target_host);
EndTransactionResponse response;
- cb(false, response);
+ std::error_code ec = ErrorCode::BadRequest;
+ cb(ec, response);
return;
}
@@ -987,17 +1248,43 @@
invocation_context->context.set_deadline(deadline);
auto callback = [target_host, cb](const InvocationContext<EndTransactionResponse>* invocation_context) {
- if (!invocation_context->status.ok() ||
- google::rpc::Code::OK != invocation_context->response.common().status().code()) {
- SPDLOG_WARN("Failed to endTransaction. TargetHost={}, gRPC statusCode={}, errorMessage={}", target_host.data(),
- invocation_context->status.error_message(), invocation_context->status.error_message());
- cb(false, invocation_context->response);
+ std::error_code ec;
+ if (!invocation_context->status.ok()) {
+ SPDLOG_WARN("Failed to write EndTransaction to wire. gRPC-code: {}, gRPC-message: {}, host={}",
+ invocation_context->status.error_code(), invocation_context->status.error_message(),
+ invocation_context->remote_address);
+ ec = ErrorCode::BadRequest;
+ cb(ec, invocation_context->response);
return;
}
- SPDLOG_DEBUG("endTransaction completed OK. Response: {}", invocation_context->response.DebugString());
- cb(true, invocation_context->response);
+ const auto& common = invocation_context->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ SPDLOG_DEBUG("endTransaction completed OK. Response: {}, host={}", invocation_context->response.DebugString(),
+ invocation_context->remote_address);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Unauthorized;
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Forbidden;
+ } break;
+ case google::rpc::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::InternalServerError;
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplemented: please upgrade SDK to latest release. {}, host={}", common.status().message(),
+ invocation_context->remote_address);
+ ec = ErrorCode::NotImplemented;
+ }
+ }
+ cb(ec, invocation_context->response);
};
+
invocation_context->callback = callback;
client->asyncEndTransaction(request, invocation_context);
}
@@ -1042,10 +1329,14 @@
void ClientManagerImpl::queryOffset(const std::string& target_host, const Metadata& metadata,
const QueryOffsetRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryOffsetResponse&)>& cb) {
+ const std::function<void(const std::error_code&, const QueryOffsetResponse&)>& cb) {
auto client = getRpcClient(target_host);
+ std::error_code ec;
if (!client) {
SPDLOG_WARN("Failed to get/create RPC client for {}", target_host);
+ ec = ErrorCode::RequestTimeout;
+ QueryOffsetResponse response;
+ cb(ec, response);
return;
}
@@ -1053,21 +1344,45 @@
invocation_context->remote_address = target_host;
invocation_context->context.set_deadline(std::chrono::system_clock::now() + timeout);
auto callback = [cb](const InvocationContext<QueryOffsetResponse>* invocation_context) {
+ std::error_code ec;
+
if (!invocation_context->status.ok()) {
- SPDLOG_WARN("Failed to send query offset request to {}. Reason: {}", invocation_context->remote_address,
- invocation_context->status.error_message());
- cb(false, invocation_context->response);
+ SPDLOG_WARN("Failed to write QueryOffset request to wire. gRPC-code: {}, gRPC-message: {}, host={}",
+ invocation_context->status.error_code(), invocation_context->status.error_message(),
+ invocation_context->remote_address);
+ ec = ErrorCode::RequestTimeout;
+ cb(ec, invocation_context->response);
return;
}
- if (google::rpc::Code::OK != invocation_context->response.common().status().code()) {
- SPDLOG_WARN("Server[host={}] failed to process query offset request. Reason: {}",
- invocation_context->remote_address, invocation_context->response.common().DebugString());
- cb(false, invocation_context->response);
+ const auto& common = invocation_context->response.common();
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ SPDLOG_DEBUG("Query offset from server[host={}] OK", invocation_context->remote_address);
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Unauthorized;
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::Forbidden;
+ cb(ec, invocation_context->response);
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}, host={}", common.status().message(), invocation_context->remote_address);
+ ec = ErrorCode::InternalServerError;
+ cb(ec, invocation_context->response);
+ } break;
+ default: {
+ SPDLOG_WARN("NotImplemented: please upgrade SDK to the latest release. host={}",
+ invocation_context->remote_address);
+ ec = ErrorCode::NotImplemented;
+ cb(ec, invocation_context->response);
+ }
}
-
- SPDLOG_DEBUG("Query offset from server[host={}] OK", invocation_context->remote_address);
- cb(true, invocation_context->response);
};
invocation_context->callback = callback;
client->asyncQueryOffset(request, invocation_context);
@@ -1141,6 +1456,8 @@
NotifyClientTerminationResponse response;
grpc::Status status = client->notifyClientTermination(&context, request, &response);
if (!status.ok()) {
+ SPDLOG_WARN("Failed to write NotifyClientTermination request to wire. gRPC-code: {}, gRPC-message: {}, host={}",
+ status.error_code(), status.error_message(), target_host);
return false;
}
diff --git a/src/main/cpp/client/LogInterceptor.cpp b/src/main/cpp/client/LogInterceptor.cpp
index 3497ddd..869fc95 100644
--- a/src/main/cpp/client/LogInterceptor.cpp
+++ b/src/main/cpp/client/LogInterceptor.cpp
@@ -34,8 +34,12 @@
response_headers.insert({absl::string_view(it.first.data(), it.first.length()),
absl::string_view(it.second.data(), it.second.length())});
}
- SPDLOG_DEBUG("[Inbound]Response Headers of {}:\n{}", client_rpc_info_->method(),
- absl::StrJoin(response_headers, "\n", absl::PairFormatter(" --> ")));
+ if (!response_headers.empty()) {
+ SPDLOG_DEBUG("[Inbound]Response Headers of {}:\n{}", client_rpc_info_->method(),
+ absl::StrJoin(response_headers, "\n", absl::PairFormatter(" --> ")));
+ } else {
+ SPDLOG_DEBUG("[Inbound]Response metadata of {} is empty", client_rpc_info_->method());
+ }
}
}
diff --git a/src/main/cpp/client/include/Client.h b/src/main/cpp/client/include/Client.h
index d029598..9a14e8c 100644
--- a/src/main/cpp/client/include/Client.h
+++ b/src/main/cpp/client/include/Client.h
@@ -18,6 +18,8 @@
virtual bool active() = 0;
+ virtual void onRemoteEndpointRemoval(const std::vector<std::string>&) = 0;
+
/**
* For endpoints that are marked as inactive due to one or multiple business operation failure, this function is to
* initiate health-check RPCs; Once the health-check passes, they are conceptually add back to serve further business
diff --git a/src/main/cpp/client/include/ClientManager.h b/src/main/cpp/client/include/ClientManager.h
index b98a6d3..28d0ef1 100644
--- a/src/main/cpp/client/include/ClientManager.h
+++ b/src/main/cpp/client/include/ClientManager.h
@@ -2,6 +2,7 @@
#include <chrono>
#include <memory>
+#include <system_error>
#include "Client.h"
#include "ReceiveMessageCallback.h"
@@ -10,6 +11,7 @@
#include "TopAddressing.h"
#include "TopicRouteData.h"
#include "rocketmq/MQMessageExt.h"
+#include "rocketmq/State.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -29,11 +31,11 @@
virtual void resolveRoute(const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) = 0;
+ const std::function<void(const std::error_code&, const TopicRouteDataPtr& ptr)>& cb) = 0;
virtual void heartbeat(const std::string& target_host, const Metadata& metadata, const HeartbeatRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const HeartbeatResponse&)>& cb) = 0;
+ const std::function<void(const std::error_code&, const HeartbeatResponse&)>& cb) = 0;
virtual void multiplexingCall(const std::string& target, const Metadata& metadata, const MultiplexingRequest& request,
std::chrono::milliseconds timeout,
@@ -42,10 +44,10 @@
virtual bool wrapMessage(const rmq::Message& item, MQMessageExt& message_ext) = 0;
virtual void ack(const std::string& target_host, const Metadata& metadata, const AckMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& cb) = 0;
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& cb) = 0;
virtual void nack(const std::string& target_host, const Metadata& metadata, const NackMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& callback) = 0;
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& callback) = 0;
virtual void forwardMessageToDeadLetterQueue(
const std::string& target_host, const Metadata& metadata, const ForwardMessageToDeadLetterQueueRequest& request,
@@ -54,22 +56,23 @@
virtual void endTransaction(const std::string& target_host, const Metadata& metadata,
const EndTransactionRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const EndTransactionResponse&)>& cb) = 0;
+ const std::function<void(const std::error_code&, const EndTransactionResponse&)>& cb) = 0;
virtual void queryOffset(const std::string& target_host, const Metadata& metadata, const QueryOffsetRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryOffsetResponse&)>& cb) = 0;
+ const std::function<void(const std::error_code&, const QueryOffsetResponse&)>& cb) = 0;
virtual void
healthCheck(const std::string& target_host, const Metadata& metadata, const HealthCheckRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(const std::string&, const InvocationContext<HealthCheckResponse>*)>& cb) = 0;
+ const std::function<void(const std::error_code&, const InvocationContext<HealthCheckResponse>*)>& cb) = 0;
virtual void addClientObserver(std::weak_ptr<Client> client) = 0;
- virtual void queryAssignment(const std::string& target, const Metadata& metadata,
- const QueryAssignmentRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryAssignmentResponse&)>& cb) = 0;
+ virtual void
+ queryAssignment(const std::string& target, const Metadata& metadata, const QueryAssignmentRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code&, const QueryAssignmentResponse&)>& cb) = 0;
virtual void receiveMessage(const std::string& target, const Metadata& metadata, const ReceiveMessageRequest& request,
std::chrono::milliseconds timeout, const std::shared_ptr<ReceiveMessageCallback>& cb) = 0;
@@ -87,6 +90,8 @@
virtual bool notifyClientTermination(const std::string& target_host, const Metadata& metadata,
const NotifyClientTerminationRequest& request,
std::chrono::milliseconds timeout) = 0;
+
+ virtual State state() const = 0;
};
using ClientManagerPtr = std::shared_ptr<ClientManager>;
diff --git a/src/main/cpp/client/include/ClientManagerImpl.h b/src/main/cpp/client/include/ClientManagerImpl.h
index ba56c47..2cb179b 100644
--- a/src/main/cpp/client/include/ClientManagerImpl.h
+++ b/src/main/cpp/client/include/ClientManagerImpl.h
@@ -6,6 +6,7 @@
#include <functional>
#include <future>
#include <string>
+#include <system_error>
#include <vector>
#include "absl/base/thread_annotations.h"
@@ -65,7 +66,7 @@
*/
void resolveRoute(const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) override
+ const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>& cb) override
LOCKS_EXCLUDED(rpc_clients_mtx_);
void doHealthCheck() LOCKS_EXCLUDED(clients_mtx_);
@@ -74,16 +75,15 @@
* If inactive RPC clients refer to remote hosts that are absent from topic_route_table_, we need to purge them
* immediately.
*/
- void cleanOfflineRpcClients() LOCKS_EXCLUDED(clients_mtx_, rpc_clients_mtx_);
+ std::vector<std::string> cleanOfflineRpcClients() LOCKS_EXCLUDED(clients_mtx_, rpc_clients_mtx_);
/**
* Execute health-check on behalf of the client.
*/
- void
- healthCheck(const std::string& target_host, const Metadata& metadata, const HealthCheckRequest& request,
- std::chrono::milliseconds timeout,
- const std::function<void(const std::string&, const InvocationContext<HealthCheckResponse>*)>& cb) override
- LOCKS_EXCLUDED(rpc_clients_mtx_);
+ void healthCheck(const std::string& target_host, const Metadata& metadata, const HealthCheckRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code&, const InvocationContext<HealthCheckResponse>*)>& cb)
+ override LOCKS_EXCLUDED(rpc_clients_mtx_);
bool send(const std::string& target_host, const Metadata& metadata, SendMessageRequest& request,
SendCallback* cb) override LOCKS_EXCLUDED(rpc_clients_mtx_);
@@ -114,7 +114,7 @@
void queryAssignment(const std::string& target, const Metadata& metadata, const QueryAssignmentRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryAssignmentResponse&)>& cb) override;
+ const std::function<void(const std::error_code&, const QueryAssignmentResponse&)>& cb) override;
void receiveMessage(const std::string& target, const Metadata& metadata, const ReceiveMessageRequest& request,
std::chrono::milliseconds timeout, const std::shared_ptr<ReceiveMessageCallback>& cb) override
@@ -137,10 +137,10 @@
* @param request Ack message request.
*/
void ack(const std::string& target_host, const Metadata& metadata, const AckMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& cb) override;
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& cb) override;
void nack(const std::string& target_host, const Metadata& metadata, const NackMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& callback) override;
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& callback) override;
void forwardMessageToDeadLetterQueue(
const std::string& target_host, const Metadata& metadata, const ForwardMessageToDeadLetterQueueRequest& request,
@@ -163,7 +163,7 @@
*/
void endTransaction(const std::string& target_host, const Metadata& metadata, const EndTransactionRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const EndTransactionResponse&)>& cb) override;
+ const std::function<void(const std::error_code&, const EndTransactionResponse&)>& cb) override;
void multiplexingCall(const std::string& target, const Metadata& metadata, const MultiplexingRequest& request,
std::chrono::milliseconds timeout,
@@ -171,7 +171,7 @@
void queryOffset(const std::string& target_host, const Metadata& metadata, const QueryOffsetRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryOffsetResponse&)>& cb) override;
+ const std::function<void(const std::error_code&, const QueryOffsetResponse&)>& cb) override;
void pullMessage(const std::string& target_host, const Metadata& metadata, const PullMessageRequest& request,
std::chrono::milliseconds timeout,
@@ -181,15 +181,19 @@
const NotifyClientTerminationRequest& request,
std::chrono::milliseconds timeout) override;
- void trace(bool trace) { trace_ = trace; }
+ void trace(bool trace) {
+ trace_ = trace;
+ }
void heartbeat(const std::string& target_host, const Metadata& metadata, const HeartbeatRequest& request,
std::chrono::milliseconds timeout,
- const std::function<void(bool, const HeartbeatResponse&)>& cb) override;
+ const std::function<void(const std::error_code&, const HeartbeatResponse&)>& cb) override;
void processPullResult(const grpc::ClientContext& client_context, const PullMessageResponse& response,
ReceiveMessageResult& result, const std::string& target_host) override;
+ State state() const override;
+
private:
void processPopResult(const grpc::ClientContext& client_context, const ReceiveMessageResponse& response,
ReceiveMessageResult& result, const std::string& target_host);
diff --git a/src/main/cpp/client/include/ReceiveMessageCallback.h b/src/main/cpp/client/include/ReceiveMessageCallback.h
index 738b8a7..e4088a6 100644
--- a/src/main/cpp/client/include/ReceiveMessageCallback.h
+++ b/src/main/cpp/client/include/ReceiveMessageCallback.h
@@ -2,12 +2,18 @@
#include "ReceiveMessageResult.h"
#include "rocketmq/AsyncCallback.h"
+#include "rocketmq/ErrorCode.h"
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
+
class ReceiveMessageCallback : public AsyncCallback {
public:
- virtual ~ReceiveMessageCallback() = default;
- virtual void onSuccess(ReceiveMessageResult& result) = 0;
- virtual void onException(MQException& e) = 0;
+ ~ReceiveMessageCallback() override = default;
+
+ virtual void onSuccess(ReceiveMessageResult &result) = 0;
+
+ virtual void onFailure(const std::error_code &ec) = 0;
};
+
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/client/mocks/include/ClientManagerMock.h b/src/main/cpp/client/mocks/include/ClientManagerMock.h
index 7abce6e..f5e5bc2 100644
--- a/src/main/cpp/client/mocks/include/ClientManagerMock.h
+++ b/src/main/cpp/client/mocks/include/ClientManagerMock.h
@@ -3,6 +3,7 @@
#include "ClientManager.h"
#include "gmock/gmock.h"
#include <chrono>
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
@@ -18,12 +19,12 @@
MOCK_METHOD(void, resolveRoute,
(const std::string&, const Metadata&, const QueryRouteRequest&, std::chrono::milliseconds,
- (const std::function<void(bool, const TopicRouteDataPtr&)>&)),
+ (const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>&)),
(override));
MOCK_METHOD(void, heartbeat,
(const std::string&, const Metadata&, const HeartbeatRequest&, std::chrono::milliseconds,
- (const std::function<void(bool, const HeartbeatResponse&)>&)),
+ (const std::function<void(const std::error_code&, const HeartbeatResponse&)>&)),
(override));
MOCK_METHOD(void, multiplexingCall,
@@ -35,12 +36,12 @@
MOCK_METHOD(void, ack,
(const std::string&, const Metadata&, const AckMessageRequest&, std::chrono::milliseconds,
- (const std::function<void(bool)>&)),
+ (const std::function<void(const std::error_code&)>&)),
(override));
MOCK_METHOD(void, nack,
(const std::string&, const Metadata&, const NackMessageRequest&, std::chrono::milliseconds,
- (const std::function<void(bool)>&)),
+ (const std::function<void(const std::error_code&)>&)),
(override));
MOCK_METHOD(void, forwardMessageToDeadLetterQueue,
@@ -51,24 +52,24 @@
MOCK_METHOD(void, endTransaction,
(const std::string&, const Metadata&, const EndTransactionRequest&, std::chrono::milliseconds,
- (const std::function<void(bool, const EndTransactionResponse&)>&)),
+ (const std::function<void(const std::error_code&, const EndTransactionResponse&)>&)),
(override));
MOCK_METHOD(void, queryOffset,
(const std::string&, const Metadata&, const QueryOffsetRequest&, std::chrono::milliseconds,
- (const std::function<void(bool, const QueryOffsetResponse&)>&)),
+ (const std::function<void(const std::error_code&, const QueryOffsetResponse&)>&)),
(override));
MOCK_METHOD(void, healthCheck,
(const std::string&, const Metadata&, const HealthCheckRequest&, std::chrono::milliseconds,
- (const std::function<void(const std::string&, const InvocationContext<HealthCheckResponse>*)>&)),
+ (const std::function<void(const std::error_code&, const InvocationContext<HealthCheckResponse>*)>&)),
(override));
MOCK_METHOD(void, addClientObserver, (std::weak_ptr<Client>), (override));
MOCK_METHOD(void, queryAssignment,
(const std::string& target, const Metadata&, const QueryAssignmentRequest&, std::chrono::milliseconds,
- (const std::function<void(bool, const QueryAssignmentResponse&)>&)),
+ (const std::function<void(const std::error_code&, const QueryAssignmentResponse&)>&)),
(override));
MOCK_METHOD(void, receiveMessage,
@@ -90,6 +91,8 @@
MOCK_METHOD(bool, notifyClientTermination,
(const std::string&, const Metadata&, const NotifyClientTerminationRequest&, std::chrono::milliseconds),
(override));
+
+ MOCK_METHOD(State, state, (), (const override));
};
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/client/mocks/include/ClientMock.h b/src/main/cpp/client/mocks/include/ClientMock.h
index 7ad69b0..dfd7a66 100644
--- a/src/main/cpp/client/mocks/include/ClientMock.h
+++ b/src/main/cpp/client/mocks/include/ClientMock.h
@@ -13,6 +13,8 @@
MOCK_METHOD(bool, active, (), (override));
+ MOCK_METHOD(void, onRemoteEndpointRemoval, (const std::vector<std::string>&), (override));
+
MOCK_METHOD(void, healthCheck, (), (override));
MOCK_METHOD(void, schedule, (const std::string&, const std::function<void()>&, std::chrono::milliseconds),
diff --git a/src/main/cpp/client/mocks/include/ReceiveMessageCallbackMock.h b/src/main/cpp/client/mocks/include/ReceiveMessageCallbackMock.h
index cb7016c..7f40903 100644
--- a/src/main/cpp/client/mocks/include/ReceiveMessageCallbackMock.h
+++ b/src/main/cpp/client/mocks/include/ReceiveMessageCallbackMock.h
@@ -1,14 +1,16 @@
#pragma once
-#include "ReceiveMessageCallback.h"
+
#include "gmock/gmock.h"
+#include "ReceiveMessageCallback.h"
+
ROCKETMQ_NAMESPACE_BEGIN
class ReceiveMessageCallbackMock : public ReceiveMessageCallback {
public:
- MOCK_METHOD(void, onSuccess, (ReceiveMessageResult&), (override));
+ MOCK_METHOD(void, onSuccess, (ReceiveMessageResult &), (override));
- MOCK_METHOD(void, onException, (MQException&), (override));
+ MOCK_METHOD(void, onFailure, (const std::error_code &), (override));
};
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/rocketmq/AsyncReceiveMessageCallback.cpp b/src/main/cpp/rocketmq/AsyncReceiveMessageCallback.cpp
index 654b4e6..9bd1d52 100644
--- a/src/main/cpp/rocketmq/AsyncReceiveMessageCallback.cpp
+++ b/src/main/cpp/rocketmq/AsyncReceiveMessageCallback.cpp
@@ -3,6 +3,7 @@
#include "ConsumeMessageType.h"
#include "LoggerImpl.h"
#include "PushConsumer.h"
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
@@ -98,10 +99,10 @@
}
}
-void AsyncReceiveMessageCallback::onException(MQException& e) {
+void AsyncReceiveMessageCallback::onFailure(const std::error_code& ec) {
auto process_queue_ptr = process_queue_.lock();
if (process_queue_ptr) {
- SPDLOG_WARN("pop message error:{}, pop message later. Queue={}", e.what(), process_queue_ptr->simpleName());
+ SPDLOG_WARN("pop message error:{}, pop message later. Queue={}", ec.message(), process_queue_ptr->simpleName());
// pop message later
receiveMessageLater();
}
diff --git a/src/main/cpp/rocketmq/AwaitPullCallback.cpp b/src/main/cpp/rocketmq/AwaitPullCallback.cpp
index 99ca912..6c942a8 100644
--- a/src/main/cpp/rocketmq/AwaitPullCallback.cpp
+++ b/src/main/cpp/rocketmq/AwaitPullCallback.cpp
@@ -2,19 +2,18 @@
ROCKETMQ_NAMESPACE_BEGIN
-void AwaitPullCallback::onSuccess(const PullResult& pull_result) {
+void AwaitPullCallback::onSuccess(const PullResult& pull_result) noexcept {
absl::MutexLock lk(&mtx_);
- has_failure_ = false;
completed_ = true;
// TODO: optimize out messages copy here.
pull_result_ = pull_result;
cv_.SignalAll();
}
-void AwaitPullCallback::onException(const MQException& e) {
+void AwaitPullCallback::onFailure(const std::error_code& ec) noexcept {
absl::MutexLock lk(&mtx_);
- has_failure_ = true;
completed_ = true;
+ ec_ = ec;
cv_.SignalAll();
}
@@ -24,7 +23,7 @@
while (!completed_) {
cv_.Wait(&mtx_);
}
- return !has_failure_;
+ return !hasFailure();
}
}
diff --git a/src/main/cpp/rocketmq/ClientImpl.cpp b/src/main/cpp/rocketmq/ClientImpl.cpp
index df1b5ae..60a4049 100644
--- a/src/main/cpp/rocketmq/ClientImpl.cpp
+++ b/src/main/cpp/rocketmq/ClientImpl.cpp
@@ -2,9 +2,11 @@
#include <chrono>
#include <cstdint>
#include <cstdlib>
+#include <functional>
#include <iterator>
#include <memory>
#include <string>
+#include <system_error>
#include <utility>
#include "absl/strings/str_join.h"
@@ -85,7 +87,8 @@
}
}
-void ClientImpl::getRouteFor(const std::string& topic, const std::function<void(TopicRouteDataPtr)>& cb) {
+void ClientImpl::getRouteFor(const std::string& topic,
+ const std::function<void(const std::error_code&, TopicRouteDataPtr)>& cb) {
TopicRouteDataPtr route = nullptr;
{
absl::MutexLock lock(&topic_route_table_mtx_);
@@ -95,7 +98,8 @@
}
if (route) {
- cb(route);
+ std::error_code ec;
+ cb(ec, route);
return;
}
@@ -116,7 +120,7 @@
SPDLOG_DEBUG("Would reuse prior route request for topic={}", topic);
return;
} else {
- std::vector<std::function<void(const TopicRouteDataPtr&)>> inflight{cb};
+ std::vector<std::function<void(const std::error_code&, const TopicRouteDataPtr&)>> inflight{cb};
inflight_route_requests_.insert({topic, inflight});
SPDLOG_INFO("Create inflight route query cache for topic={}", topic);
}
@@ -124,9 +128,11 @@
}
if (!query_backend && route) {
- cb(route);
+ std::error_code ec;
+ cb(ec, route);
} else {
- fetchRouteFor(topic, std::bind(&ClientImpl::onTopicRouteReady, this, topic, std::placeholders::_1));
+ fetchRouteFor(topic,
+ std::bind(&ClientImpl::onTopicRouteReady, this, topic, std::placeholders::_1, std::placeholders::_2));
}
}
@@ -163,26 +169,27 @@
}
}
-void ClientImpl::fetchRouteFor(const std::string& topic, const std::function<void(const TopicRouteDataPtr&)>& cb) {
+void ClientImpl::fetchRouteFor(const std::string& topic,
+ const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>& cb) {
std::string name_server = name_server_resolver_->current();
if (name_server.empty()) {
SPDLOG_WARN("No name server available");
return;
}
- auto callback = [this, topic, name_server, cb](bool ok, const TopicRouteDataPtr& route) {
- if (!ok || !route) {
+ auto callback = [this, topic, name_server, cb](const std::error_code& ec, const TopicRouteDataPtr& route) {
+ if (ec) {
SPDLOG_WARN("Failed to resolve route for topic={} from {}", topic, name_server);
std::string name_server_changed = name_server_resolver_->next();
if (!name_server_changed.empty()) {
SPDLOG_INFO("Change current name server from {} to {}", name_server, name_server_changed);
}
- cb(nullptr);
+ cb(ec, nullptr);
return;
}
SPDLOG_DEBUG("Apply callback of fetchRouteFor({}) since a valid route is fetched", topic);
- cb(route);
+ cb(ec, route);
};
QueryRouteRequest request;
@@ -212,14 +219,10 @@
if (!topics.empty()) {
for (const auto& topic : topics) {
- fetchRouteFor(topic, std::bind(&ClientImpl::updateRouteCache, this, topic, std::placeholders::_1));
+ fetchRouteFor(
+ topic, std::bind(&ClientImpl::updateRouteCache, this, topic, std::placeholders::_1, std::placeholders::_2));
}
}
-
-#ifdef ENABLE_TRACING
- updateTraceProvider();
-#endif
-
SPDLOG_DEBUG("Topic route info updated");
}
@@ -238,9 +241,9 @@
Signature::sign(this, metadata);
for (const auto& target : hosts) {
- auto callback = [target](bool ok, const HeartbeatResponse& response) {
- if (!ok) {
- SPDLOG_WARN("Failed to send heartbeat request to {}", target);
+ auto callback = [target](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);
@@ -249,15 +252,16 @@
}
}
-void ClientImpl::onTopicRouteReady(const std::string& topic, const TopicRouteDataPtr& route) {
+void ClientImpl::onTopicRouteReady(const std::string& topic, const std::error_code& ec,
+ const TopicRouteDataPtr& route) {
if (route) {
SPDLOG_DEBUG("Received route data for topic={}", topic);
}
- updateRouteCache(topic, route);
+ updateRouteCache(topic, ec, route);
// Take all pending callbacks
- std::vector<std::function<void(const TopicRouteDataPtr&)>> pending_requests;
+ std::vector<std::function<void(const std::error_code&, const TopicRouteDataPtr&)>> pending_requests;
{
absl::MutexLock lk(&inflight_route_requests_mtx_);
assert(inflight_route_requests_.contains(topic));
@@ -268,13 +272,13 @@
SPDLOG_DEBUG("Apply cached callbacks with acquired route data for topic={}", topic);
for (const auto& cb : pending_requests) {
- cb(route);
+ cb(ec, route);
}
}
-void ClientImpl::updateRouteCache(const std::string& topic, const TopicRouteDataPtr& route) {
- if (!route || route->partitions().empty()) {
- SPDLOG_WARN("Yuck! route for {} is invalid", topic);
+void ClientImpl::updateRouteCache(const std::string& topic, const std::error_code& ec, const TopicRouteDataPtr& route) {
+ if (ec || !route || route->partitions().empty()) {
+ SPDLOG_WARN("Yuck! route for {} is invalid. Cause: {}", topic, ec.message());
return;
}
@@ -385,6 +389,18 @@
}
}
+void ClientImpl::onRemoteEndpointRemoval(const std::vector<std::string>& hosts) {
+ absl::MutexLock lk(&isolated_endpoints_mtx_);
+ for (auto it = isolated_endpoints_.begin(); it != isolated_endpoints_.end();) {
+ if (hosts.end() != std::find_if(hosts.begin(), hosts.end(), [&](const std::string& item) { return *it == item; })) {
+ SPDLOG_INFO("Drop isolated-endoint[{}] as it has been removed from route table", *it);
+ isolated_endpoints_.erase(it++);
+ } else {
+ it++;
+ }
+ }
+}
+
void ClientImpl::healthCheck() {
std::vector<std::string> endpoints;
{
@@ -395,14 +411,14 @@
}
std::weak_ptr<ClientImpl> base(self());
- auto callback = [base](const std::string& endpoint,
- const InvocationContext<HealthCheckResponse>* invocation_context) {
+ auto callback = [base](const std::error_code& ec, const InvocationContext<HealthCheckResponse>* invocation_context) {
std::shared_ptr<ClientImpl> ptr = base.lock();
- if (ptr) {
- ptr->onHealthCheckResponse(endpoint, invocation_context);
- } else {
+ if (!ptr) {
SPDLOG_INFO("BaseImpl has been destructed");
+ return;
}
+
+ ptr->onHealthCheckResponse(ec, invocation_context);
};
for (const auto& endpoint : endpoints) {
@@ -418,31 +434,16 @@
client_manager_->getScheduler().schedule(task, task_name, delay, std::chrono::milliseconds(0));
}
-void ClientImpl::onHealthCheckResponse(const std::string& endpoint, const InvocationContext<HealthCheckResponse>* ctx) {
- if (!ctx) {
- SPDLOG_WARN("ClientInstance does not have RPC client for {}. It might have been offline and thus cleaned",
- endpoint);
- {
- absl::MutexLock lk(&isolated_endpoints_mtx_);
- isolated_endpoints_.erase(endpoint);
- }
+void ClientImpl::onHealthCheckResponse(const std::error_code& ec, const InvocationContext<HealthCheckResponse>* ctx) {
+ if (ec) {
+ SPDLOG_WARN("Health check to server[host={}] failed. Cause: {}", ec.message());
return;
}
- assert(endpoint == ctx->remote_address);
-
- if (ctx->status.ok()) {
- if (google::rpc::Code::OK == ctx->response.common().status().code()) {
- SPDLOG_INFO("Health check to server[host={}] passed. Move it back to active node pool", endpoint);
- absl::MutexLock lk(&isolated_endpoints_mtx_);
- isolated_endpoints_.erase(endpoint);
- } else {
- SPDLOG_INFO("Health check to server[host={}] failed due to application layer reason: {}",
- ctx->response.common().DebugString());
- }
- } else {
- SPDLOG_INFO("Health check to server[host={}] failed due to transport layer reason: {}", endpoint,
- ctx->status.error_message());
+ SPDLOG_INFO("Health check to server[host={}] passed. Remove it from isolated endpoint pool", ctx->remote_address);
+ {
+ absl::MutexLock lk(&isolated_endpoints_mtx_);
+ isolated_endpoints_.erase(ctx->remote_address);
}
}
diff --git a/src/main/cpp/rocketmq/ConsumeFifoMessageService.cpp b/src/main/cpp/rocketmq/ConsumeFifoMessageService.cpp
index 1049055..4a9ef06 100644
--- a/src/main/cpp/rocketmq/ConsumeFifoMessageService.cpp
+++ b/src/main/cpp/rocketmq/ConsumeFifoMessageService.cpp
@@ -1,16 +1,19 @@
+#include <chrono>
+#include <functional>
+#include <limits>
+#include <system_error>
+
#include "ConsumeMessageService.h"
#include "MessageAccessor.h"
#include "ProcessQueue.h"
#include "PushConsumerImpl.h"
-#include <chrono>
-#include <functional>
-#include <limits>
ROCKETMQ_NAMESPACE_BEGIN
ConsumeFifoMessageService ::ConsumeFifoMessageService(std::weak_ptr<PushConsumer> consumer, int thread_count,
MessageListener* message_listener)
- : ConsumeMessageService(std::move(consumer), thread_count, message_listener) {}
+ : ConsumeMessageService(std::move(consumer), thread_count, message_listener) {
+}
void ConsumeFifoMessageService::start() {
ConsumeMessageService::start();
@@ -78,7 +81,9 @@
}
}
-MessageListenerType ConsumeFifoMessageService::messageListenerType() { return MessageListenerType::FIFO; }
+MessageListenerType ConsumeFifoMessageService::messageListenerType() {
+ return MessageListenerType::FIFO;
+}
void ConsumeFifoMessageService::consumeTask(const ProcessQueueWeakPtr& process_queue, MQMessageExt& message) {
ProcessQueueSharedPtr process_queue_ptr = process_queue.lock();
@@ -145,31 +150,32 @@
}
}
-void ConsumeFifoMessageService::onAck(const ProcessQueueWeakPtr& process_queue, const MQMessageExt& message, bool ok) {
+void ConsumeFifoMessageService::onAck(const ProcessQueueWeakPtr& process_queue, const MQMessageExt& message,
+ const std::error_code& ec) {
auto process_queue_ptr = process_queue.lock();
if (!process_queue_ptr) {
SPDLOG_WARN("ProcessQueue has destructed.");
return;
}
- if (ok) {
- SPDLOG_DEBUG("Acknowledge FIFO message[MessageQueue={}, MsgId={}] OK", process_queue_ptr->simpleName(),
- message.getMsgId());
- process_queue_ptr->unbindFifoConsumeTask();
- signalDispatcher();
- } else {
- SPDLOG_WARN("Failed to acknowledge FIFO message[MessageQueue={}, MsgId={}]", process_queue_ptr->simpleName(),
- message.getMsgId());
+
+ if (ec) {
+ SPDLOG_WARN("Failed to acknowledge FIFO message[MessageQueue={}, MsgId={}]. Cause: {}",
+ process_queue_ptr->simpleName(), message.getMsgId(), ec.message());
auto consumer = consumer_.lock();
if (!consumer) {
SPDLOG_WARN("Consumer instance has destructed");
return;
}
-
auto task = std::bind(&ConsumeFifoMessageService::scheduleAckTask, this, process_queue, message);
int32_t duration = 100;
consumer->schedule("Ack-FIFO-Message-On-Failure", task, std::chrono::milliseconds(duration));
SPDLOG_INFO("Scheduled to ack message[Topic={}, MessageId={}] in {}ms", message.getTopic(), message.getMsgId(),
duration);
+ } else {
+ SPDLOG_DEBUG("Acknowledge FIFO message[MessageQueue={}, MsgId={}] OK", process_queue_ptr->simpleName(),
+ message.getMsgId());
+ process_queue_ptr->unbindFifoConsumeTask();
+ signalDispatcher();
}
}
diff --git a/src/main/cpp/rocketmq/ConsumeStandardMessageService.cpp b/src/main/cpp/rocketmq/ConsumeStandardMessageService.cpp
index d74274f..e73674e 100644
--- a/src/main/cpp/rocketmq/ConsumeStandardMessageService.cpp
+++ b/src/main/cpp/rocketmq/ConsumeStandardMessageService.cpp
@@ -1,5 +1,6 @@
#include <limits>
#include <string>
+#include <system_error>
#include <utility>
#include "absl/memory/memory.h"
@@ -25,7 +26,8 @@
ConsumeStandardMessageService::ConsumeStandardMessageService(std::weak_ptr<PushConsumer> consumer, int thread_count,
MessageListener* message_listener_ptr)
- : ConsumeMessageService(std::move(consumer), thread_count, message_listener_ptr) {}
+ : ConsumeMessageService(std::move(consumer), thread_count, message_listener_ptr) {
+}
void ConsumeStandardMessageService::start() {
ConsumeMessageService::start();
@@ -88,7 +90,9 @@
}
}
-MessageListenerType ConsumeStandardMessageService::messageListenerType() { return MessageListenerType::STANDARD; }
+MessageListenerType ConsumeStandardMessageService::messageListenerType() {
+ return MessageListenerType::STANDARD;
+}
void ConsumeStandardMessageService::consumeTask(const ProcessQueueWeakPtr& process_queue,
const std::vector<MQMessageExt>& msgs) {
@@ -197,12 +201,12 @@
{
for (auto& span : spans) {
switch (status) {
- case ConsumeMessageResult::SUCCESS:
- span.SetStatus(opencensus::trace::StatusCode::OK);
- break;
- case ConsumeMessageResult::FAILURE:
- span.SetStatus(opencensus::trace::StatusCode::UNKNOWN);
- break;
+ case ConsumeMessageResult::SUCCESS:
+ span.SetStatus(opencensus::trace::StatusCode::OK);
+ break;
+ case ConsumeMessageResult::FAILURE:
+ span.SetStatus(opencensus::trace::StatusCode::UNKNOWN);
+ break;
}
span.End();
}
@@ -219,26 +223,26 @@
process_queue_ptr->release(msg.getBody().size(), msg.getQueueOffset());
if (status == ConsumeMessageResult::SUCCESS) {
- auto callback = [process_queue_ptr, message_id](bool ok) {
- if (ok) {
+ auto callback = [process_queue_ptr, message_id](const std::error_code& ec) {
+ if (ec) {
+ SPDLOG_WARN("Failed to acknowledge message[MessageQueue={}, MsgId={}]. Cause: {}",
+ process_queue_ptr->simpleName(), message_id, ec.message());
+ } else {
SPDLOG_DEBUG("Acknowledge message[MessageQueue={}, MsgId={}] OK", process_queue_ptr->simpleName(),
message_id);
- } else {
- SPDLOG_WARN("Failed to acknowledge message[MessageQueue={}, MsgId={}]", process_queue_ptr->simpleName(),
- message_id);
}
};
consumer->ack(msg, callback);
} else {
- auto callback = [process_queue_ptr, message_id](bool ok) {
- if (ok) {
- SPDLOG_DEBUG("Nack message[MessageQueue={}, MsgId={}] OK", process_queue_ptr->simpleName(), message_id);
- } else {
- SPDLOG_INFO(
- "Failed to negative acknowledge message[MessageQueue={}, MsgId={}]. Message will be re-consumed "
- "after default invisible time",
- process_queue_ptr->simpleName(), message_id);
+ auto callback = [process_queue_ptr, message_id](const std::error_code& ec) {
+ if (ec) {
+ SPDLOG_WARN("Failed to negative acknowledge message[MessageQueue={}, MsgId={}]. Cause: {} Message will be "
+ "re-consumed after default invisible time",
+ process_queue_ptr->simpleName(), message_id, ec.message());
+ return;
}
+
+ SPDLOG_DEBUG("Nack message[MessageQueue={}, MsgId={}] OK", process_queue_ptr->simpleName(), message_id);
};
consumer->nack(msg, callback);
}
diff --git a/src/main/cpp/rocketmq/DefaultMQProducer.cpp b/src/main/cpp/rocketmq/DefaultMQProducer.cpp
index f72e64f..2780dca 100644
--- a/src/main/cpp/rocketmq/DefaultMQProducer.cpp
+++ b/src/main/cpp/rocketmq/DefaultMQProducer.cpp
@@ -2,6 +2,8 @@
#include <chrono>
#include <memory>
+#include <system_error>
+#include <utility>
#include "absl/strings/str_split.h"
@@ -24,8 +26,14 @@
return absl::ToChronoMilliseconds(impl_->getIoTimeout());
}
-SendResult DefaultMQProducer::send(const MQMessage& message, const std::string& message_group) {
- return impl_->send(message, message_group);
+SendResult DefaultMQProducer::send(MQMessage& message, const std::string& message_group) {
+ message.bindMessageGroup(message_group);
+ std::error_code ec;
+ auto&& send_result = impl_->send(message, ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+ return std::move(send_result);
}
void DefaultMQProducer::setSendMsgTimeout(std::chrono::milliseconds timeout) {
@@ -50,43 +58,109 @@
bool DefaultMQProducer::isTracingEnabled() { return impl_->isTracingEnabled(); }
-SendResult DefaultMQProducer::send(const rocketmq::MQMessage& message, bool filter_active_broker) {
- return impl_->send(message);
+SendResult DefaultMQProducer::send(const MQMessage& message, bool filter_active_broker) {
+ std::error_code ec;
+ auto&& send_result = impl_->send(message, ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+ return std::move(send_result);
}
-SendResult DefaultMQProducer::send(const MQMessage& msg, const MQMessageQueue& mq) { return impl_->send(msg, mq); }
-
-SendResult DefaultMQProducer::send(const MQMessage& msg, MessageQueueSelector* selector, void* arg) {
- return impl_->send(msg, selector, arg);
+SendResult DefaultMQProducer::send(const MQMessage& message, std::error_code& ec) noexcept {
+ return impl_->send(message, ec);
}
-SendResult DefaultMQProducer::send(const MQMessage& message, MessageQueueSelector* selector, void* arg, int retry_times,
+SendResult DefaultMQProducer::send(MQMessage& msg, const MQMessageQueue& mq) {
+ msg.bindMessageQueue(mq);
+ std::error_code ec;
+ auto&& send_result = impl_->send(msg, ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+ return std::move(send_result);
+}
+
+SendResult DefaultMQProducer::send(MQMessage& msg, MessageQueueSelector* selector, void* arg) {
+ std::error_code ec;
+ auto&& list = impl_->listMessageQueue(msg.getTopic(), ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+
+ auto&& message_queue = selector->select(list, msg, arg);
+ msg.bindMessageQueue(message_queue);
+
+ auto&& send_result = impl_->send(msg, ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+ return std::move(send_result);
+}
+
+SendResult DefaultMQProducer::send(MQMessage& message, MessageQueueSelector* selector, void* arg, int retry_times,
bool select_active_broker) {
- return impl_->send(message, selector, arg, retry_times);
+ return send(message, selector, arg);
}
void DefaultMQProducer::send(const MQMessage& message, SendCallback* send_callback, bool select_active_broker) {
impl_->send(message, send_callback);
}
-void DefaultMQProducer::send(const MQMessage& message, const MQMessageQueue& message_queue,
+void DefaultMQProducer::send(MQMessage& message, const MQMessageQueue& message_queue, SendCallback* send_callback) {
+ message.bindMessageQueue(message_queue);
+ impl_->send(message, send_callback);
+}
+
+void DefaultMQProducer::send(MQMessage& message, MessageQueueSelector* selector, void* arg,
SendCallback* send_callback) {
- impl_->send(message, message_queue, send_callback);
+ std::error_code ec;
+
+ // TODO: make querying route async
+ auto&& list = impl_->listMessageQueue(message.getTopic(), ec);
+ if (ec) {
+ send_callback->onFailure(ec);
+ }
+
+ if (list.empty()) {
+ send_callback->onFailure(ErrorCode::ServiceUnavailable);
+ }
+
+ auto&& message_queue = selector->select(list, message, arg);
+ message.bindMessageQueue(message_queue);
+
+ impl_->send(message, send_callback);
}
-void DefaultMQProducer::send(const MQMessage& message, MessageQueueSelector* selector, void* arg,
- SendCallback* send_callback) {
- impl_->send(message, selector, arg, send_callback);
+void DefaultMQProducer::sendOneway(const MQMessage& message, bool select_active_broker) {
+ std::error_code ec;
+ impl_->sendOneway(message, ec);
}
-void DefaultMQProducer::sendOneway(const MQMessage& message, bool select_active_broker) { impl_->sendOneway(message); }
-
-void DefaultMQProducer::sendOneway(const MQMessage& message, const MQMessageQueue& message_queue) {
- impl_->sendOneway(message, message_queue);
+void DefaultMQProducer::sendOneway(MQMessage& message, const MQMessageQueue& message_queue) {
+ message.bindMessageQueue(message_queue);
+ std::error_code ec;
+ impl_->sendOneway(message, ec);
+ if (ec) {
+ SPDLOG_INFO("Failed to send message in one-way: {}", ec.message());
+ }
}
-void DefaultMQProducer::sendOneway(const MQMessage& message, MessageQueueSelector* selector, void* arg) {
- impl_->sendOneway(message, selector, arg);
+void DefaultMQProducer::sendOneway(MQMessage& message, MessageQueueSelector* selector, void* arg) {
+ std::error_code ec;
+ auto&& list = impl_->listMessageQueue(message.getTopic(), ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+
+ if (list.empty()) {
+ ec = ErrorCode::ServiceUnavailable;
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+
+ auto&& message_queue = selector->select(list, message, arg);
+ message.bindMessageQueue(message_queue);
+ impl_->sendOneway(message, ec);
}
void DefaultMQProducer::setLocalTransactionStateChecker(LocalTransactionStateCheckerPtr checker) {
@@ -98,7 +172,12 @@
int DefaultMQProducer::getMaxAttemptTimes() const { return impl_->maxAttemptTimes(); }
std::vector<MQMessageQueue> DefaultMQProducer::getTopicMessageQueueInfo(const std::string& topic) {
- return impl_->getTopicMessageQueueInfo(topic);
+ std::error_code ec;
+ auto&& list = impl_->listMessageQueue(topic, ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
+ return std::move(list);
}
void DefaultMQProducer::setUnitName(std::string unit_name) { impl_->setUnitName(std::move(unit_name)); }
@@ -120,7 +199,11 @@
void DefaultMQProducer::setRegion(const std::string& region) { impl_->region(region); }
TransactionPtr DefaultMQProducer::prepare(MQMessage& message) {
- auto transaction = impl_->prepare(message);
+ std::error_code ec;
+ auto transaction = impl_->prepare(message, ec);
+ if (ec) {
+ THROW_MQ_EXCEPTION(MQClientException, ec.message(), ec.value());
+ }
return transaction;
}
diff --git a/src/main/cpp/rocketmq/ProcessQueueImpl.cpp b/src/main/cpp/rocketmq/ProcessQueueImpl.cpp
index f5c455e..da7c97d 100644
--- a/src/main/cpp/rocketmq/ProcessQueueImpl.cpp
+++ b/src/main/cpp/rocketmq/ProcessQueueImpl.cpp
@@ -3,6 +3,7 @@
#include <atomic>
#include <chrono>
#include <memory>
+#include <system_error>
#include <utility>
#include "ClientManagerImpl.h"
@@ -77,12 +78,12 @@
auto policy = consumer->receiveMessageAction();
switch (policy) {
- case ReceiveMessageAction::POLLING:
- popMessage();
- break;
- case ReceiveMessageAction::PULL:
- pullMessage();
- break;
+ case ReceiveMessageAction::POLLING:
+ popMessage();
+ break;
+ case ReceiveMessageAction::PULL:
+ pullMessage();
+ break;
}
}
@@ -115,29 +116,61 @@
auto timeout = consumer->getLongPollingTimeout();
auto callback = [this](const InvocationContext<PullMessageResponse>* invocation_context) {
- if (!invocation_context) {
- MQException e("Transport layer failure", -1, __FILE__, __LINE__);
- receive_callback_->onException(e);
- return;
- }
+ auto status = invocation_context->status;
+ if (status.ok()) {
+ const auto& common = invocation_context->response.common();
- if (invocation_context->status.ok()) {
- if (google::rpc::Code::OK == invocation_context->response.common().status().code()) {
- ReceiveMessageResult result;
- client_manager_->processPullResult(invocation_context->context, invocation_context->response, result,
- invocation_context->remote_address);
- receive_callback_->onSuccess(result);
- } else {
- auto status = invocation_context->response.common().status();
- MQException e(status.message(), status.code(), __FILE__, __LINE__);
- receive_callback_->onException(e);
+ switch (common.status().code()) {
+ case google::rpc::Code::OK: {
+ ReceiveMessageResult result;
+ client_manager_->processPullResult(invocation_context->context, invocation_context->response, result,
+ invocation_context->remote_address);
+ receive_callback_->onSuccess(result);
+ } break;
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", common.status().message());
+ std::error_code ec = ErrorCode::Forbidden;
+ receive_callback_->onFailure(ec);
+ } break;
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", common.status().message());
+ std::error_code ec = ErrorCode::Unauthorized;
+ receive_callback_->onFailure(ec);
+ } break;
+ case google::rpc::Code::DEADLINE_EXCEEDED: {
+ SPDLOG_WARN("DeadlineExceeded: {}", common.status().message());
+ std::error_code ec = ErrorCode::GatewayTimeout;
+ receive_callback_->onFailure(ec);
+ } break;
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("InvalidArgument: {}", common.status().message());
+ std::error_code ec = ErrorCode::BadRequest;
+ receive_callback_->onFailure(ec);
+ } break;
+ case google::rpc::Code::FAILED_PRECONDITION: {
+ SPDLOG_WARN("FailedPrecondition: {}", common.status().message());
+ std::error_code ec = ErrorCode::PreconditionRequired;
+ receive_callback_->onFailure(ec);
+ } break;
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("InternalServerError: {}", common.status().message());
+ std::error_code ec = ErrorCode::InternalServerError;
+ receive_callback_->onFailure(ec);
+ } break;
+ default: {
+ SPDLOG_WARN("Unimplemented: Please upgrade to use latest SDK release");
+ std::error_code ec = ErrorCode::NotImplemented;
+ receive_callback_->onFailure(ec);
+ } break;
}
} else {
- MQException e(invocation_context->status.error_message(), invocation_context->status.error_code(), __FILE__,
- __LINE__);
- receive_callback_->onException(e);
+ SPDLOG_WARN("Failed to receive valid gRPC response from server. gRPC-status[code={}, message={}]",
+ status.error_code(), status.error_message());
+ std::error_code ec = ErrorCode::RequestTimeout;
+ receive_callback_->onFailure(ec);
}
};
+
client_manager_->pullMessage(message_queue_.serviceAddress(), metadata, request, absl::ToChronoMilliseconds(timeout),
callback);
}
@@ -161,13 +194,14 @@
const std::string& msg_id = message.getMsgId();
if (!filter_expression_.accept(message)) {
const std::string& topic = message.getTopic();
- auto callback = [topic, msg_id](bool ok) {
- if (ok) {
+ auto callback = [topic, msg_id](const std::error_code& ec) {
+ if (ec) {
+ SPDLOG_WARN(
+ "Failed to ack message[Topic={}, MsgId={}] directly as it fails to pass filter expression. Cause: {}",
+ topic, msg_id, ec.message());
+ } else {
SPDLOG_DEBUG("Ack message[Topic={}, MsgId={}] directly as it fails to pass filter expression", topic,
msg_id);
- } else {
- SPDLOG_WARN("Failed to ack message[Topic={}, MsgId={}] directly as it fails to pass filter expression",
- topic, msg_id);
}
};
consumer->ack(message, callback);
@@ -248,14 +282,14 @@
if (optional.has_value()) {
auto expression = optional.value();
switch (expression.type_) {
- case TAG:
- filter_expression->set_type(rmq::FilterType::TAG);
- filter_expression->set_expression(expression.content_);
- break;
- case SQL92:
- filter_expression->set_type(rmq::FilterType::SQL);
- filter_expression->set_expression(expression.content_);
- break;
+ case TAG:
+ filter_expression->set_type(rmq::FilterType::TAG);
+ filter_expression->set_expression(expression.content_);
+ break;
+ case SQL92:
+ filter_expression->set_type(rmq::FilterType::SQL);
+ filter_expression->set_expression(expression.content_);
+ break;
}
} else {
filter_expression->set_type(rmq::FilterType::TAG);
@@ -306,12 +340,20 @@
wrapFilterExpression(request.mutable_filter_expression());
}
-std::weak_ptr<PushConsumer> ProcessQueueImpl::getConsumer() { return consumer_; }
+std::weak_ptr<PushConsumer> ProcessQueueImpl::getConsumer() {
+ return consumer_;
+}
-std::shared_ptr<ClientManager> ProcessQueueImpl::getClientManager() { return client_manager_; }
+std::shared_ptr<ClientManager> ProcessQueueImpl::getClientManager() {
+ return client_manager_;
+}
-MQMessageQueue ProcessQueueImpl::getMQMessageQueue() { return message_queue_; }
+MQMessageQueue ProcessQueueImpl::getMQMessageQueue() {
+ return message_queue_;
+}
-const FilterExpression& ProcessQueueImpl::getFilterExpression() const { return filter_expression_; }
+const FilterExpression& ProcessQueueImpl::getFilterExpression() const {
+ return filter_expression_;
+}
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/rocketmq/ProducerImpl.cpp b/src/main/cpp/rocketmq/ProducerImpl.cpp
index 9f174fd..85f5344 100644
--- a/src/main/cpp/rocketmq/ProducerImpl.cpp
+++ b/src/main/cpp/rocketmq/ProducerImpl.cpp
@@ -1,6 +1,9 @@
#include "ProducerImpl.h"
#include <atomic>
+#include <limits>
+#include <system_error>
+#include <utility>
#include "absl/strings/str_join.h"
#include "opencensus/trace/propagation/trace_context.h"
@@ -20,7 +23,6 @@
#include "UniqueIdGenerator.h"
#include "UtilAll.h"
#include "rocketmq/ErrorCode.h"
-#include "rocketmq/MQClientException.h"
#include "rocketmq/MQMessage.h"
#include "rocketmq/MQMessageQueue.h"
#include "rocketmq/Transaction.h"
@@ -55,9 +57,9 @@
bool ProducerImpl::isRunning() const { return State::STARTED == state_.load(std::memory_order_relaxed); }
-void ProducerImpl::ensureRunning() const {
+void ProducerImpl::ensureRunning(std::error_code& ec) const noexcept {
if (!isRunning()) {
- THROW_MQ_EXCEPTION(MQClientException, "Invoke #start() first", ILLEGAL_STATE);
+ ec = ErrorCode::IllegalState;
}
}
@@ -118,17 +120,51 @@
return message_id;
}
-SendResult ProducerImpl::send(const MQMessage& message) {
- ensureRunning();
+SendResult ProducerImpl::send(const MQMessage& message, std::error_code& ec) noexcept {
+ ensureRunning(ec);
+ if (ec) {
+ return {};
+ }
+
auto topic_publish_info = getPublishInfo(message.getTopic());
if (!topic_publish_info) {
- THROW_MQ_EXCEPTION(MQClientException, "No topic route available", NO_TOPIC_ROUTE_INFO);
+ ec = ErrorCode::NotFound;
+ return {};
}
std::vector<MQMessageQueue> message_queue_list;
- takeMessageQueuesRoundRobin(topic_publish_info, message_queue_list, max_attempt_times_);
+
+ if (!message.messageGroup().empty() || message.messageQueue()) {
+ auto&& list = listMessageQueue(message.getTopic(), ec);
+ if (ec) {
+ return {};
+ }
+
+ if (list.empty()) {
+ ec = ErrorCode::ServiceUnavailable;
+ return {};
+ }
+
+ if (!message.messageGroup().empty()) {
+ std::size_t hash_code = std::hash<std::string>{}(message.messageGroup());
+ hash_code = hash_code & std::numeric_limits<std::size_t>::max();
+ std::size_t r = hash_code % list.size();
+ message_queue_list.push_back(list[r]);
+ } else {
+ for (const auto& entry : list) {
+ if (entry == message.messageQueue()) {
+ message_queue_list.push_back(entry);
+ break;
+ }
+ }
+ }
+ } else {
+ takeMessageQueuesRoundRobin(topic_publish_info, message_queue_list, max_attempt_times_);
+ }
+
if (message_queue_list.empty()) {
- THROW_MQ_EXCEPTION(MQClientException, "No topic route available", NO_TOPIC_ROUTE_INFO);
+ ec = ErrorCode::ServiceUnavailable;
+ return {};
}
AwaitSendCallback callback;
@@ -138,138 +174,56 @@
if (callback) {
return callback.sendResult();
}
- THROW_MQ_EXCEPTION(MQClientException, callback.errorMessage(), FAILED_TO_SEND_MESSAGE);
-}
-SendResult ProducerImpl::send(const MQMessage& message, const std::string& message_group) {
- MessageGroupQueueSelector selector(message_group);
- return send(message, &selector, nullptr);
-}
-
-SendResult ProducerImpl::send(const MQMessage& message, const MQMessageQueue& message_queue) {
- ensureRunning();
- std::vector<MQMessageQueue> message_queue_list{withServiceAddress(message_queue)};
- AwaitSendCallback callback;
- send0(message, &callback, message_queue_list, max_attempt_times_);
- callback.await();
- if (callback) {
- return callback.sendResult();
- }
- THROW_MQ_EXCEPTION(MQClientException, callback.errorMessage(), FAILED_TO_SEND_MESSAGE);
-}
-
-SendResult ProducerImpl::send(const MQMessage& message, MessageQueueSelector* selector, void* arg) {
- ensureRunning();
- std::vector<MQMessageQueue> message_queue_list;
- executeMessageQueueSelector(message, selector, arg, message_queue_list);
- if (message_queue_list.empty()) {
- THROW_MQ_EXCEPTION(MQClientException, "No topic route available", NO_TOPIC_ROUTE_INFO);
- }
- AwaitSendCallback callback;
- send0(message, &callback, message_queue_list, max_attempt_times_);
- callback.await();
- if (callback) {
- return callback.sendResult();
- }
- THROW_MQ_EXCEPTION(MQClientException, callback.errorMessage(), FAILED_TO_SEND_MESSAGE);
-}
-
-SendResult ProducerImpl::send(const MQMessage& message, MessageQueueSelector* selector, void* arg, int max_attempts) {
- ensureRunning();
- std::vector<MQMessageQueue> message_queue_list;
- executeMessageQueueSelector(message, selector, arg, message_queue_list);
- if (message_queue_list.empty()) {
- THROW_MQ_EXCEPTION(MQClientException, "No topic route available", NO_TOPIC_ROUTE_INFO);
- }
- AwaitSendCallback callback;
- send0(message, &callback, message_queue_list, max_attempts);
- callback.await();
- if (callback) {
- return callback.sendResult();
- }
- THROW_MQ_EXCEPTION(MQClientException, callback.errorMessage(), FAILED_TO_SEND_MESSAGE);
+ ec = callback.errorCode();
+ return {};
}
void ProducerImpl::send(const MQMessage& message, SendCallback* cb) {
- ensureRunning();
- auto callback = [this, message, cb](const TopicPublishInfoPtr& publish_info) {
- if (!publish_info) {
- MQClientException e("Failed to acquire topic route data", NO_TOPIC_ROUTE_INFO, __FILE__, __LINE__);
- cb->onException(e);
+ std::error_code ec;
+ ensureRunning(ec);
+ if (ec) {
+ cb->onFailure(ec);
+ }
+
+ auto callback = [this, message, cb](const std::error_code& ec, const TopicPublishInfoPtr& publish_info) {
+ if (ec) {
+ cb->onFailure(ec);
return;
}
std::vector<MQMessageQueue> message_queue_list;
- takeMessageQueuesRoundRobin(publish_info, message_queue_list, max_attempt_times_);
+ if (!message.messageGroup().empty() || message.messageQueue()) {
+ auto&& list = publish_info->getMessageQueueList();
+ if (!message.messageGroup().empty()) {
+ std::size_t hash_code = std::hash<std::string>{}(message.messageGroup());
+ hash_code = hash_code & std::numeric_limits<std::size_t>::max();
+ std::size_t r = hash_code % list.size();
+ message_queue_list.push_back(list[r]);
+ } else {
+ for (const auto& entry : list) {
+ if (entry == message.messageQueue()) {
+ message_queue_list.push_back(entry);
+ break;
+ }
+ }
+ }
+ } else {
+ takeMessageQueuesRoundRobin(publish_info, message_queue_list, max_attempt_times_);
+ }
+
+ if (message_queue_list.empty()) {
+ cb->onFailure(ErrorCode::ServiceUnavailable);
+ return;
+ }
+
send0(message, cb, message_queue_list, max_attempt_times_);
};
asyncPublishInfo(message.getTopic(), callback);
}
-void ProducerImpl::send(const MQMessage& message, const MQMessageQueue& message_queue, SendCallback* callback) {
- ensureRunning();
- std::vector<MQMessageQueue> message_queue_list{withServiceAddress(message_queue)};
- send0(message, callback, message_queue_list, max_attempt_times_);
-}
-
-void ProducerImpl::send(const MQMessage& message, MessageQueueSelector* selector, void* arg, SendCallback* callback) {
- ensureRunning();
-
- auto cb = [this, message, selector, callback, arg](const TopicPublishInfoPtr& ptr) {
- if (!ptr) {
- MQClientException e("Failed to acquire topic route", NO_TOPIC_ROUTE_INFO, __FILE__, __LINE__);
- callback->onException(e);
- return;
- }
-
- MQMessageQueue queue = selector->select(ptr->getMessageQueueList(), message, arg);
- std::vector<MQMessageQueue> message_queue_list{queue};
-
- send0(message, callback, message_queue_list, max_attempt_times_);
- };
-
- asyncPublishInfo(message.getTopic(), cb);
-}
-
-void ProducerImpl::sendOneway(const MQMessage& message) {
- ensureRunning();
- auto callback = [this, message](const TopicPublishInfoPtr& ptr) {
- if (!ptr) {
- SPDLOG_WARN("Failed acquire topic publish info for {}", message.getTopic());
- return;
- }
- MQMessageQueue message_queue;
- absl::flat_hash_set<std::string> isolated;
- isolatedEndpoints(isolated);
- ptr->selectOneActiveMessageQueue(isolated, message_queue);
- std::vector<MQMessageQueue> list{message_queue};
- send0(message, onewaySendCallback(), list, 1);
- };
- asyncPublishInfo(message.getTopic(), callback);
-}
-
-void ProducerImpl::sendOneway(const MQMessage& message, const MQMessageQueue& message_queue) {
- ensureRunning();
- std::vector<MQMessageQueue> list{withServiceAddress(message_queue)};
- send0(message, onewaySendCallback(), list, 1);
-}
-
-void ProducerImpl::sendOneway(const MQMessage& message, MessageQueueSelector* selector, void* arg) {
- ensureRunning();
-
- auto callback = [this, message, selector, arg](const TopicPublishInfoPtr& ptr) {
- if (!ptr) {
- SPDLOG_WARN("No topic route for {}", message.getTopic());
- return;
- }
- MQMessageQueue queue = selector->select(ptr->getMessageQueueList(), message, arg);
- std::vector<MQMessageQueue> list{queue};
- send0(message, onewaySendCallback(), list, 1);
- };
-
- asyncPublishInfo(message.getTopic(), callback);
-}
+void ProducerImpl::sendOneway(const MQMessage& message, std::error_code& ec) { send(message, ec); }
void ProducerImpl::setLocalTransactionStateChecker(LocalTransactionStateCheckerPtr checker) {
transaction_state_checker_ = std::move(checker);
@@ -279,9 +233,8 @@
const std::string& target = callback->messageQueue().serviceAddress();
if (target.empty()) {
SPDLOG_WARN("Failed to resolve broker address from MessageQueue");
- MQClientException e("Failed to resolve broker address", FAILED_TO_RESOLVE_BROKER_ADDRESS_FROM_TOPIC_ROUTE, __FILE__,
- __LINE__);
- callback->onException(e);
+ std::error_code ec = ErrorCode::BadGateway;
+ callback->onFailure(ec);
return;
}
@@ -333,20 +286,20 @@
assert(callback);
if (!validate(message)) {
- MQClientException e("Message is illegal", MESSAGE_ILLEGAL, __FILE__, __LINE__);
- callback->onException(e);
+ std::error_code ec = ErrorCode::BadRequest;
+ callback->onFailure(ec);
return;
}
if (list.empty()) {
- MQClientException e("Topic route not found", NO_TOPIC_ROUTE_INFO, __FILE__, __LINE__);
- callback->onException(e);
+ std::error_code ec = ErrorCode::NotFound;
+ callback->onFailure(ec);
return;
}
if (max_attempt_times <= 0) {
- MQClientException e("Retry times illegal", ERR_INVALID_MAX_ATTEMPT_TIME, __FILE__, __LINE__);
- callback->onException(e);
+ std::error_code ec = ErrorCode::BadConfiguration;
+ callback->onFailure(ec);
return;
}
MQMessageQueue message_queue = list[0];
@@ -403,26 +356,16 @@
absl::Mutex mtx;
absl::CondVar cv;
- auto cb = [&, span](bool rpc_ok, const EndTransactionResponse& response) {
+ auto cb = [&, span](const std::error_code& ec, const EndTransactionResponse& response) {
completed = true;
- if (!rpc_ok) {
+ if (ec) {
{
span.SetStatus(opencensus::trace::StatusCode::ABORTED);
- span.AddAnnotation("gRPC tier failure");
+ span.AddAnnotation(ec.message());
span.End();
}
-
- SPDLOG_WARN("Failed to send {} transaction request to {}", action, target);
+ SPDLOG_WARN("Failed to send {} transaction request to {}. Cause: ", action, target, ec.message());
success = false;
- } else if (response.common().status().code() != google::rpc::Code::OK) {
- {
- span.SetStatus(opencensus::trace::ABORTED);
- span.AddAnnotation(response.common().status().DebugString());
- span.End();
- }
- success = false;
- SPDLOG_WARN("Server[host={}] failed to {} transaction. Reason: {}", target, action,
- response.common().DebugString());
} else {
{
span.SetStatus(opencensus::trace::StatusCode::OK);
@@ -450,17 +393,21 @@
endpoints.insert(isolated_endpoints_.begin(), isolated_endpoints_.end());
}
-MQMessageQueue ProducerImpl::withServiceAddress(const MQMessageQueue& message_queue) {
+MQMessageQueue ProducerImpl::withServiceAddress(const MQMessageQueue& message_queue, std::error_code& ec) {
if (!message_queue.serviceAddress().empty()) {
return message_queue;
}
if (message_queue.getTopic().empty() || message_queue.getBrokerName().empty() || message_queue.getQueueId() < 0) {
- MQClientException e("Message queue is illegal", MESSAGE_QUEUE_ILLEGAL, __FILE__, __LINE__);
- throw e;
+ ec = ErrorCode::BadRequest;
+ return {};
}
- std::vector<MQMessageQueue> list = getTopicMessageQueueInfo(message_queue.getTopic());
+ std::vector<MQMessageQueue> list = listMessageQueue(message_queue.getTopic(), ec);
+ if (ec) {
+ return {};
+ }
+
for (const auto& item : list) {
if (item == message_queue) {
return item;
@@ -468,8 +415,8 @@
}
if (list.empty()) {
- MQClientException e("No topic route available", NO_TOPIC_ROUTE_INFO, __FILE__, __LINE__);
- throw e;
+ ec = ErrorCode::NotFound;
+ return {};
} else {
return *list.begin();
}
@@ -485,17 +432,16 @@
isolated_endpoints_.insert(target);
}
-std::unique_ptr<TransactionImpl> ProducerImpl::prepare(MQMessage& message) {
- try {
- message.messageType(MessageType::TRANSACTION);
- SendResult send_result = send(message);
- return std::unique_ptr<TransactionImpl>(new TransactionImpl(
- send_result.getMsgId(), send_result.getTransactionId(), send_result.getMessageQueue().serviceAddress(),
- send_result.traceContext(), ProducerImpl::shared_from_this()));
- } catch (const MQClientException& e) {
- SPDLOG_ERROR("Failed to send transaction message. Cause: {}", e.what());
+std::unique_ptr<TransactionImpl> ProducerImpl::prepare(MQMessage& message, std::error_code& ec) {
+ message.messageType(MessageType::TRANSACTION);
+ SendResult send_result = send(message, ec);
+ if (ec) {
return nullptr;
}
+
+ return std::unique_ptr<TransactionImpl>(new TransactionImpl(
+ send_result.getMsgId(), send_result.getTransactionId(), send_result.getMessageQueue().serviceAddress(),
+ send_result.traceContext(), ProducerImpl::shared_from_this()));
}
bool ProducerImpl::commit(const std::string& message_id, const std::string& transaction_id,
@@ -509,7 +455,7 @@
}
void ProducerImpl::asyncPublishInfo(const std::string& topic,
- const std::function<void(const TopicPublishInfoPtr&)>& cb) {
+ const std::function<void(const std::error_code&, const TopicPublishInfoPtr&)>& cb) {
TopicPublishInfoPtr ptr;
{
absl::MutexLock lock(&topic_publish_info_mtx_);
@@ -517,12 +463,13 @@
ptr = topic_publish_info_table_.at(topic);
}
}
+ std::error_code ec;
if (ptr) {
- cb(ptr);
+ cb(ec, ptr);
} else {
- auto callback = [this, topic, cb](const TopicRouteDataPtr& route) {
- if (!route) {
- cb(nullptr);
+ auto callback = [this, topic, cb](const std::error_code& ec, const TopicRouteDataPtr& route) {
+ if (ec) {
+ cb(ec, nullptr);
return;
}
@@ -531,8 +478,9 @@
absl::MutexLock lk(&topic_publish_info_mtx_);
topic_publish_info_table_.insert_or_assign(topic, publish_info);
}
- cb(publish_info);
+ cb(ec, publish_info);
};
+
getRouteFor(topic, callback);
}
}
@@ -542,9 +490,11 @@
absl::Mutex mtx;
absl::CondVar cv;
TopicPublishInfoPtr topic_publish_info;
- auto cb = [&](const TopicPublishInfoPtr& ptr) {
+ std::error_code error_code;
+ auto cb = [&](const std::error_code& ec, const TopicPublishInfoPtr& ptr) {
absl::MutexLock lk(&mtx);
topic_publish_info = ptr;
+ error_code = ec;
complete = true;
cv.SignalAll();
};
@@ -555,37 +505,11 @@
absl::MutexLock lk(&mtx);
cv.Wait(&mtx);
}
+
+ // TODO: propogate error_code to caller
return topic_publish_info;
}
-bool ProducerImpl::executeMessageQueueSelector(const MQMessage& message, MessageQueueSelector* selector, void* arg,
- std::vector<MQMessageQueue>& result) {
- TopicPublishInfoPtr publish_info;
- absl::Mutex mtx;
- absl::CondVar cv;
- bool completed = false;
- auto callback = [&](const TopicPublishInfoPtr& ptr) {
- absl::MutexLock lk(&mtx);
- publish_info = ptr;
- completed = true;
- cv.SignalAll();
- };
- asyncPublishInfo(message.getTopic(), callback);
-
- while (!completed) {
- absl::MutexLock lk(&mtx);
- cv.Wait(&mtx);
- }
-
- if (!publish_info) {
- THROW_MQ_EXCEPTION(MQClientException, "Failed to acquire topic route data", NO_TOPIC_ROUTE_INFO);
- }
-
- MQMessageQueue queue = selector->select(publish_info->getMessageQueueList(), message, arg);
- result.emplace_back(std::move(queue));
- return true;
-}
-
void ProducerImpl::takeMessageQueuesRoundRobin(const TopicPublishInfoPtr& publish_info,
std::vector<MQMessageQueue>& message_queues, int number) {
assert(publish_info);
@@ -594,17 +518,19 @@
publish_info->takeMessageQueues(isolated, message_queues, number);
}
-std::vector<MQMessageQueue> ProducerImpl::getTopicMessageQueueInfo(const std::string& topic) {
+std::vector<MQMessageQueue> ProducerImpl::listMessageQueue(const std::string& topic, std::error_code& ec) {
absl::Mutex mtx;
absl::CondVar cv;
bool completed = false;
TopicPublishInfoPtr ptr;
- auto await_callback = [&](const TopicPublishInfoPtr& publish_info) {
+ auto await_callback = [&](const std::error_code& error_code, const TopicPublishInfoPtr& publish_info) {
absl::MutexLock lk(&mtx);
ptr = publish_info;
+ ec = error_code;
completed = true;
cv.SignalAll();
};
+
asyncPublishInfo(topic, await_callback);
while (!completed) {
@@ -612,8 +538,8 @@
cv.Wait(&mtx);
}
- if (!ptr) {
- THROW_MQ_EXCEPTION(MQClientException, "Failed to acquire topic publish_info", NO_TOPIC_ROUTE_INFO);
+ if (ec) {
+ return {};
}
return ptr->getMessageQueueList();
diff --git a/src/main/cpp/rocketmq/PullConsumerImpl.cpp b/src/main/cpp/rocketmq/PullConsumerImpl.cpp
index df45146..daec943 100644
--- a/src/main/cpp/rocketmq/PullConsumerImpl.cpp
+++ b/src/main/cpp/rocketmq/PullConsumerImpl.cpp
@@ -2,8 +2,12 @@
#include "ClientManagerFactory.h"
#include "InvocationContext.h"
#include "Signature.h"
-#include "rocketmq/MessageModel.h"
#include "apache/rocketmq/v1/definition.pb.h"
+#include "rocketmq/ErrorCode.h"
+#include "rocketmq/MQClientException.h"
+#include "rocketmq/MessageModel.h"
+#include <exception>
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
@@ -43,15 +47,21 @@
return promise->get_future();
}
}
- auto callback = [promise](const TopicRouteDataPtr& route) {
- if (route) {
- std::vector<MQMessageQueue> message_queues;
- for (const auto& partition : route->partitions()) {
- message_queues.emplace_back(partition.asMessageQueue());
- }
- promise->set_value(message_queues);
+
+ auto callback = [promise](const std::error_code& ec, const TopicRouteDataPtr& route) {
+ if (ec) {
+ MQClientException e(ec.message(), ec.value(), __FILE__, __LINE__);
+ promise->set_exception(std::make_exception_ptr(e));
+ return;
}
+
+ std::vector<MQMessageQueue> message_queues;
+ for (const auto& partition : route->partitions()) {
+ message_queues.emplace_back(partition.asMessageQueue());
+ }
+ promise->set_value(message_queues);
};
+
getRouteFor(topic, callback);
return promise->get_future();
}
@@ -85,13 +95,13 @@
// TODO: Use std::unique_ptr if C++14 is adopted.
auto promise_ptr = std::make_shared<std::promise<int64_t>>();
- auto callback = [promise_ptr](bool ok, const QueryOffsetResponse& response) {
- if (ok) {
- promise_ptr->set_value(response.offset());
+ auto callback = [promise_ptr](const std::error_code& ec, const QueryOffsetResponse& response) {
+ if (ec) {
+ MQClientException e(ec.message(), ec.value(), __FILE__, __LINE__);
+ promise_ptr->set_exception(std::make_exception_ptr(e));
return;
}
- MQClientException e("Failed to query offset", -1, __FILE__, __LINE__);
- promise_ptr->set_exception(std::make_exception_ptr(e));
+ promise_ptr->set_value(response.offset());
};
client_manager_->queryOffset(query.message_queue.serviceAddress(), metadata, request,
@@ -119,28 +129,63 @@
auto callback = [this, target_host, cb](const InvocationContext<PullMessageResponse>* invocation_context) {
if (!invocation_context || !invocation_context->status.ok()) {
- MQClientException exception(fmt::format("Server[{}] is not reachable", target_host), -1, __FILE__, __LINE__);
- cb->onException(exception);
+ std::error_code ec = ErrorCode::RequestTimeout;
+ cb->onFailure(ec);
return;
}
auto response = invocation_context->response;
auto biz_status = response.common().status();
- if (google::rpc::Code::OK != biz_status.code()) {
- MQClientException exception(response.common().status().message(), biz_status.code(), __FILE__, __LINE__);
- cb->onException(exception);
- return;
- }
- std::vector<MQMessageExt> messages;
- for (const auto& item : response.messages()) {
- MQMessageExt message_ext;
- if (client_manager_->wrapMessage(item, message_ext)) {
- messages.emplace_back(message_ext);
+ switch (biz_status.code()) {
+ case google::rpc::Code::OK: {
+ std::vector<MQMessageExt> messages;
+ for (const auto& item : response.messages()) {
+ MQMessageExt message_ext;
+ if (client_manager_->wrapMessage(item, message_ext)) {
+ messages.emplace_back(message_ext);
+ }
}
+ PullResult pull_result(response.min_offset(), response.max_offset(), response.next_offset(), std::move(messages));
+ cb->onSuccess(pull_result);
+ } break;
+
+ case google::rpc::Code::PERMISSION_DENIED: {
+ SPDLOG_WARN("PermissionDenied: {}", response.common().status().message());
+ std::error_code ec = ErrorCode::Forbidden;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::UNAUTHENTICATED: {
+ SPDLOG_WARN("Unauthenticated: {}", response.common().status().message());
+ std::error_code ec = ErrorCode::Unauthorized;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::DEADLINE_EXCEEDED: {
+ SPDLOG_WARN("GatewayTimeout: {}", response.common().status().message());
+ std::error_code ec = ErrorCode::GatewayTimeout;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::INVALID_ARGUMENT: {
+ SPDLOG_WARN("BadRequest: {}", response.common().status().message());
+ std::error_code ec = ErrorCode::BadRequest;
+ cb->onFailure(ec);
+ } break;
+
+ case google::rpc::Code::INTERNAL: {
+ SPDLOG_WARN("ServerIntervalError: {}", response.common().status().message());
+ std::error_code ec = ErrorCode::InternalServerError;
+ cb->onFailure(ec);
+ } break;
+
+ default: {
+ SPDLOG_WARN("Unsupported response code. Please upgrade to lastest SDK");
+ std::error_code ec = ErrorCode::NotImplemented;
+ cb->onFailure(ec);
+ } break;
}
- PullResult pull_result(response.min_offset(), response.max_offset(), response.next_offset(), std::move(messages));
- cb->onSuccess(pull_result);
};
absl::flat_hash_map<std::string, std::string> metadata;
diff --git a/src/main/cpp/rocketmq/PushConsumerImpl.cpp b/src/main/cpp/rocketmq/PushConsumerImpl.cpp
index 99550b1..c0508c1 100644
--- a/src/main/cpp/rocketmq/PushConsumerImpl.cpp
+++ b/src/main/cpp/rocketmq/PushConsumerImpl.cpp
@@ -1,5 +1,12 @@
#include "PushConsumerImpl.h"
+#include <cassert>
+#include <chrono>
+#include <cstdlib>
+#include <system_error>
+
+#include "apache/rocketmq/v1/definition.pb.h"
+
#include "AsyncReceiveMessageCallback.h"
#include "ClientManagerFactory.h"
#include "MessageAccessor.h"
@@ -9,16 +16,15 @@
#include "Signature.h"
#include "rocketmq/MQClientException.h"
#include "rocketmq/MessageModel.h"
-#include <apache/rocketmq/v1/definition.pb.h>
-#include <cassert>
-#include <chrono>
-#include <cstdlib>
ROCKETMQ_NAMESPACE_BEGIN
-PushConsumerImpl::PushConsumerImpl(absl::string_view group_name) : ClientImpl(group_name) {}
+PushConsumerImpl::PushConsumerImpl(absl::string_view group_name) : ClientImpl(group_name) {
+}
-PushConsumerImpl::~PushConsumerImpl() { SPDLOG_DEBUG("DefaultMQPushConsumerImpl is destructed"); }
+PushConsumerImpl::~PushConsumerImpl() {
+ SPDLOG_DEBUG("DefaultMQPushConsumerImpl is destructed");
+}
void PushConsumerImpl::start() {
ClientImpl::start();
@@ -152,11 +158,12 @@
std::string topic = entry.first;
const auto& filter_expression = entry.second;
SPDLOG_DEBUG("Scan assignments for {}", topic);
- auto callback = [this, topic, filter_expression](const TopicAssignmentPtr& assignments) {
- if (assignments && !assignments->assignmentList().empty()) {
+ auto callback = [this, topic, filter_expression](const std::error_code& ec,
+ const TopicAssignmentPtr& assignments) {
+ if (ec) {
+ SPDLOG_WARN("Failed to acquire assignments for topic={} from load balancer. Cause: {}", topic, ec.message());
+ } else if (assignments && !assignments->assignmentList().empty()) {
syncProcessQueue(topic, assignments, filter_expression);
- } else {
- SPDLOG_WARN("Failed to acquire assignments for topic={} from load balancer for the first time", topic);
}
};
queryAssignment(topic, callback);
@@ -193,15 +200,15 @@
request.set_client_id(client_id);
}
-void PushConsumerImpl::queryAssignment(const std::string& topic,
- const std::function<void(const TopicAssignmentPtr&)>& cb) {
+void PushConsumerImpl::queryAssignment(
+ const std::string& topic, const std::function<void(const std::error_code&, const TopicAssignmentPtr&)>& cb) {
- auto callback = [this, topic, cb](const TopicRouteDataPtr& topic_route) {
+ auto callback = [this, topic, cb](const std::error_code& ec, const TopicRouteDataPtr& topic_route) {
TopicAssignmentPtr topic_assignment;
if (MessageModel::BROADCASTING == message_model_) {
- if (!topic_route) {
- SPDLOG_WARN("Failed to get valid route entries for topic={}", topic);
- cb(topic_assignment);
+ if (ec) {
+ SPDLOG_WARN("Failed to get valid route entries for topic={}. Cause: {}", topic, ec.message());
+ cb(ec, topic_assignment);
}
std::vector<Assignment> assignments;
@@ -210,7 +217,7 @@
assignments.emplace_back(Assignment(partition.asMessageQueue()));
}
topic_assignment = std::make_shared<TopicAssignment>(std::move(assignments));
- cb(topic_assignment);
+ cb(ec, topic_assignment);
return;
}
@@ -226,16 +233,16 @@
absl::flat_hash_map<std::string, std::string> metadata;
Signature::sign(this, metadata);
-
- auto assignment_callback = [this, cb, topic, broker_host](bool ok, const QueryAssignmentResponse& response) {
- if (ok) {
+ auto assignment_callback = [this, cb, topic, broker_host](const std::error_code& ec,
+ const QueryAssignmentResponse& response) {
+ if (ec) {
+ SPDLOG_WARN("Failed to acquire queue assignment of topic={} from brokerAddress={}", topic, broker_host);
+ cb(ec, nullptr);
+ } else {
SPDLOG_DEBUG("Query topic assignment OK. Topic={}, group={}, assignment-size={}", topic, group_name_,
response.assignments().size());
SPDLOG_TRACE("Query assignment response for {} is: {}", topic, response.DebugString());
- cb(std::make_shared<TopicAssignment>(response));
- } else {
- SPDLOG_WARN("Failed to acquire queue assignment of topic={} from brokerAddress={}", topic, broker_host);
- cb(nullptr);
+ cb(ec, std::make_shared<TopicAssignment>(response));
}
};
@@ -348,42 +355,45 @@
}
switch (receive_message_policy_) {
- case ReceiveMessageAction::PULL: {
- int64_t offset = -1;
- if (!offset_store_ || !offset_store_->readOffset(message_queue, offset)) {
- // Query latest offset from server.
- QueryOffsetRequest request;
- request.mutable_partition()->mutable_topic()->set_resource_namespace(resource_namespace_);
- request.mutable_partition()->mutable_topic()->set_name(message_queue.getTopic());
- request.mutable_partition()->set_id(message_queue.getQueueId());
- request.mutable_partition()->mutable_broker()->set_name(message_queue.getBrokerName());
- request.set_policy(rmq::QueryOffsetPolicy::END);
- absl::flat_hash_map<std::string, std::string> metadata;
- Signature::sign(this, metadata);
- auto callback = [broker_host, message_queue, process_queue_ptr](bool ok, const QueryOffsetResponse& response) {
- if (ok) {
- assert(response.offset() >= 0);
- process_queue_ptr->nextOffset(response.offset());
- process_queue_ptr->receiveMessage();
- } else {
- SPDLOG_WARN("Failed to acquire latest offset for partition[{}] from server[host={}]",
- message_queue.simpleName(), broker_host);
- }
- };
- client_manager_->queryOffset(broker_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), callback);
+ case ReceiveMessageAction::PULL: {
+ int64_t offset = -1;
+ if (!offset_store_ || !offset_store_->readOffset(message_queue, offset)) {
+ // Query latest offset from server.
+ QueryOffsetRequest request;
+ request.mutable_partition()->mutable_topic()->set_resource_namespace(resource_namespace_);
+ request.mutable_partition()->mutable_topic()->set_name(message_queue.getTopic());
+ request.mutable_partition()->set_id(message_queue.getQueueId());
+ request.mutable_partition()->mutable_broker()->set_name(message_queue.getBrokerName());
+ request.set_policy(rmq::QueryOffsetPolicy::END);
+ absl::flat_hash_map<std::string, std::string> metadata;
+ Signature::sign(this, metadata);
+ auto callback = [broker_host, message_queue, process_queue_ptr](const std::error_code& ec,
+ const QueryOffsetResponse& response) {
+ if (ec) {
+ SPDLOG_WARN("Failed to acquire latest offset for partition[{}] from server[host={}]. Cause: {}",
+ message_queue.simpleName(), broker_host, ec.message());
+ } else {
+ assert(response.offset() >= 0);
+ process_queue_ptr->nextOffset(response.offset());
+ process_queue_ptr->receiveMessage();
+ }
+ };
+ client_manager_->queryOffset(broker_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), callback);
+ }
+ break;
}
- break;
- }
- case ReceiveMessageAction::POLLING:
- process_queue_ptr->receiveMessage();
- break;
+ case ReceiveMessageAction::POLLING:
+ process_queue_ptr->receiveMessage();
+ break;
}
return true;
}
-std::shared_ptr<ConsumeMessageService> PushConsumerImpl::getConsumeMessageService() { return consume_message_service_; }
+std::shared_ptr<ConsumeMessageService> PushConsumerImpl::getConsumeMessageService() {
+ return consume_message_service_;
+}
-void PushConsumerImpl::ack(const MQMessageExt& msg, const std::function<void(bool)>& callback) {
+void PushConsumerImpl::ack(const MQMessageExt& msg, const std::function<void(const std::error_code&)>& callback) {
const std::string& target_host = MessageAccessor::targetEndpoint(msg);
assert(!target_host.empty());
SPDLOG_DEBUG("Prepare to send ack to broker. BrokerAddress={}, topic={}, queueId={}, msgId={}", target_host,
@@ -395,7 +405,7 @@
client_manager_->ack(target_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), callback);
}
-void PushConsumerImpl::nack(const MQMessageExt& msg, const std::function<void(bool)>& callback) {
+void PushConsumerImpl::nack(const MQMessageExt& msg, const std::function<void(const std::error_code&)>& callback) {
std::string target_host = MessageAccessor::targetEndpoint(msg);
absl::flat_hash_map<std::string, std::string> metadata;
@@ -452,7 +462,9 @@
request.set_receipt_handle(msg.receiptHandle());
}
-uint32_t PushConsumerImpl::consumeThreadPoolSize() const { return consume_thread_pool_size_; }
+uint32_t PushConsumerImpl::consumeThreadPoolSize() const {
+ return consume_thread_pool_size_;
+}
void PushConsumerImpl::consumeThreadPoolSize(int thread_pool_size) {
if (thread_pool_size >= 1) {
@@ -460,7 +472,9 @@
}
}
-uint32_t PushConsumerImpl::consumeBatchSize() const { return consume_batch_size_; }
+uint32_t PushConsumerImpl::consumeBatchSize() const {
+ return consume_batch_size_;
+}
void PushConsumerImpl::consumeBatchSize(uint32_t consume_batch_size) {
@@ -492,15 +506,6 @@
}
}
-#ifdef ENABLE_TRACING
-nostd::shared_ptr<trace::Tracer> DefaultMQPushConsumerImpl::getTracer() {
- if (nullptr == client_manager_) {
- return nostd::shared_ptr<trace::Tracer>(nullptr);
- }
- return client_manager_->getTracer();
-}
-#endif
-
void PushConsumerImpl::iterateProcessQueue(const std::function<void(ProcessQueueSharedPtr)>& callback) {
absl::MutexLock lock(&process_queue_table_mtx_);
for (const auto& item : process_queue_table_) {
@@ -527,11 +532,11 @@
absl::Mutex mtx;
absl::CondVar cv;
int acquired = 0;
- auto callback = [&](const TopicRouteDataPtr& route) {
+ auto callback = [&](const std::error_code& ec, const TopicRouteDataPtr& route) {
absl::MutexLock lk(&mtx);
countdown--;
cv.SignalAll();
- if (route) {
+ if (!ec) {
acquired++;
}
};
@@ -552,12 +557,12 @@
if (message_listener_) {
switch (message_listener_->listenerType()) {
- case MessageListenerType::FIFO:
- request.set_fifo_flag(true);
- break;
- case MessageListenerType::STANDARD:
- request.set_fifo_flag(false);
- break;
+ case MessageListenerType::FIFO:
+ request.set_fifo_flag(true);
+ break;
+ case MessageListenerType::STANDARD:
+ request.set_fifo_flag(false);
+ break;
}
}
@@ -566,14 +571,14 @@
consumer_data->mutable_group()->set_resource_namespace(resource_namespace_);
switch (message_model_) {
- case MessageModel::BROADCASTING:
- consumer_data->set_consume_model(rmq::ConsumeModel::BROADCASTING);
- break;
- case MessageModel::CLUSTERING:
- consumer_data->set_consume_model(rmq::ConsumeModel::CLUSTERING);
- break;
- default:
- break;
+ case MessageModel::BROADCASTING:
+ consumer_data->set_consume_model(rmq::ConsumeModel::BROADCASTING);
+ break;
+ case MessageModel::CLUSTERING:
+ consumer_data->set_consume_model(rmq::ConsumeModel::CLUSTERING);
+ break;
+ default:
+ break;
}
auto subscriptions = consumer_data->mutable_subscriptions();
@@ -586,12 +591,12 @@
subscription->mutable_topic()->set_name(entry.first);
subscription->mutable_expression()->set_expression(entry.second.content_);
switch (entry.second.type_) {
- case ExpressionType::TAG:
- subscription->mutable_expression()->set_type(rmq::FilterType::TAG);
- break;
- case ExpressionType::SQL92:
- subscription->mutable_expression()->set_type(rmq::FilterType::SQL);
- break;
+ case ExpressionType::TAG:
+ subscription->mutable_expression()->set_type(rmq::FilterType::TAG);
+ break;
+ case ExpressionType::SQL92:
+ subscription->mutable_expression()->set_type(rmq::FilterType::SQL);
+ break;
}
subscriptions->AddAllocated(subscription);
}
diff --git a/src/main/cpp/rocketmq/SendCallbacks.cpp b/src/main/cpp/rocketmq/SendCallbacks.cpp
index 8b067ca..ae680e3 100644
--- a/src/main/cpp/rocketmq/SendCallbacks.cpp
+++ b/src/main/cpp/rocketmq/SendCallbacks.cpp
@@ -10,11 +10,11 @@
ROCKETMQ_NAMESPACE_BEGIN
-void OnewaySendCallback::onException(const MQException& e) {
- SPDLOG_WARN("Failed to one-way send message. Message: {}", e.what());
+void OnewaySendCallback::onFailure(const std::error_code& ec) noexcept {
+ SPDLOG_WARN("Failed to one-way send message. Message: {}", ec.message());
}
-void OnewaySendCallback::onSuccess(SendResult& send_result) {
+void OnewaySendCallback::onSuccess(SendResult& send_result) noexcept {
SPDLOG_DEBUG("Send message in one-way OK. MessageId: {}", send_result.getMsgId());
}
@@ -30,23 +30,21 @@
}
}
-void AwaitSendCallback::onSuccess(SendResult& send_result) {
+void AwaitSendCallback::onSuccess(SendResult& send_result) noexcept {
send_result_ = send_result;
- success_ = true;
completed_ = true;
absl::MutexLock lk(&mtx_);
cv_.SignalAll();
}
-void AwaitSendCallback::onException(const MQException& e) {
- success_ = false;
- error_message_ = e.what();
+void AwaitSendCallback::onFailure(const std::error_code& ec) noexcept {
completed_ = true;
+ ec_ = ec;
absl::MutexLock lk(&mtx_);
cv_.SignalAll();
}
-void RetrySendCallback::onSuccess(SendResult& send_result) {
+void RetrySendCallback::onSuccess(SendResult& send_result) noexcept {
{
// Mark end of send-message span.
span_.SetStatus(opencensus::trace::StatusCode::OK);
@@ -58,7 +56,7 @@
delete this;
}
-void RetrySendCallback::onException(const MQException& e) {
+void RetrySendCallback::onFailure(const std::error_code& ec) noexcept {
{
// Mark end of the send-message span.
span_.SetStatus(opencensus::trace::StatusCode::INTERNAL);
@@ -67,7 +65,7 @@
if (++attempt_times_ >= max_attempt_times_) {
SPDLOG_WARN("Retried {} times, which exceeds the limit: {}", attempt_times_, max_attempt_times_);
- callback_->onException(e);
+ callback_->onFailure(ec);
delete this;
return;
}
@@ -75,14 +73,14 @@
std::shared_ptr<ProducerImpl> producer = producer_.lock();
if (!producer) {
SPDLOG_WARN("Producer has been destructed");
- callback_->onException(e);
+ callback_->onFailure(ec);
delete this;
return;
}
if (candidates_.empty()) {
SPDLOG_WARN("No alternative hosts to perform additional retries");
- callback_->onException(e);
+ callback_->onFailure(ec);
delete this;
return;
}
diff --git a/src/main/cpp/rocketmq/include/AsyncReceiveMessageCallback.h b/src/main/cpp/rocketmq/include/AsyncReceiveMessageCallback.h
index 0c415c3..597819a 100644
--- a/src/main/cpp/rocketmq/include/AsyncReceiveMessageCallback.h
+++ b/src/main/cpp/rocketmq/include/AsyncReceiveMessageCallback.h
@@ -2,6 +2,7 @@
#include "ProcessQueue.h"
#include "ReceiveMessageCallback.h"
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
@@ -14,7 +15,7 @@
void onSuccess(ReceiveMessageResult& result) override;
- void onException(MQException& e) override;
+ void onFailure(const std::error_code& ec) override;
void receiveMessageLater();
diff --git a/src/main/cpp/rocketmq/include/AwaitPullCallback.h b/src/main/cpp/rocketmq/include/AwaitPullCallback.h
index 95d9572..971bb6e 100644
--- a/src/main/cpp/rocketmq/include/AwaitPullCallback.h
+++ b/src/main/cpp/rocketmq/include/AwaitPullCallback.h
@@ -1,3 +1,7 @@
+#pragma once
+
+#include <system_error>
+
#include "absl/synchronization/mutex.h"
#include "rocketmq/AsyncCallback.h"
@@ -8,22 +12,24 @@
public:
explicit AwaitPullCallback(PullResult& pull_result) : pull_result_(pull_result) {}
- void onSuccess(const PullResult& pull_result) override;
+ void onSuccess(const PullResult& pull_result) noexcept override;
- void onException(const MQException& e) override;
+ void onFailure(const std::error_code& ec) noexcept override;
bool await();
- bool hasFailure() const { return has_failure_; }
+ bool hasFailure() const { return ec_.operator bool(); }
bool isCompleted() const { return completed_; }
+ const std::error_code& errorCode() const noexcept { return ec_; }
+
private:
PullResult& pull_result_;
absl::Mutex mtx_;
absl::CondVar cv_;
bool completed_{false};
- bool has_failure_{false};
+ std::error_code ec_;
};
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/rocketmq/include/ClientImpl.h b/src/main/cpp/rocketmq/include/ClientImpl.h
index 40a9756..342bdcc 100644
--- a/src/main/cpp/rocketmq/include/ClientImpl.h
+++ b/src/main/cpp/rocketmq/include/ClientImpl.h
@@ -2,6 +2,7 @@
#include <chrono>
#include <cstdint>
+#include <system_error>
#include "absl/strings/string_view.h"
#include "apache/rocketmq/v1/definition.pb.h"
@@ -28,11 +29,12 @@
virtual void shutdown();
- void getRouteFor(const std::string& topic, const std::function<void(TopicRouteDataPtr)>& cb)
+ 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_);
/**
- * Gather collection of endpoints that are reachable from latest topic route table.
+ * Gather collection of endpoints that are reachable from latest topic route
+ * table.
*
* @param endpoints
*/
@@ -45,7 +47,9 @@
return State::STARTING == state || State::STARTED == state;
}
- void healthCheck() LOCKS_EXCLUDED(isolated_endpoints_mtx_) override;
+ void onRemoteEndpointRemoval(const std::vector<std::string>& hosts) override LOCKS_EXCLUDED(isolated_endpoints_mtx_);
+
+ void healthCheck() override LOCKS_EXCLUDED(isolated_endpoints_mtx_);
void schedule(const std::string& task_name, const std::function<void(void)>& task,
std::chrono::milliseconds delay) override;
@@ -62,7 +66,7 @@
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_
- absl::flat_hash_map<std::string, std::vector<std::function<void(const TopicRouteDataPtr&)>>>
+ absl::flat_hash_map<std::string, std::vector<std::function<void(const std::error_code&, const TopicRouteDataPtr&)>>>
inflight_route_requests_ GUARDED_BY(inflight_route_requests_mtx_);
absl::Mutex inflight_route_requests_mtx_ ACQUIRED_BEFORE(topic_route_table_mtx_); // Protects inflight_route_requests_
static const char* UPDATE_ROUTE_TASK_NAME;
@@ -91,7 +95,8 @@
virtual void resolveOrphanedTransactionalMessage(const std::string& transaction_id, const MQMessageExt& message) {}
/**
- * Concrete publisher/subscriber client is expected to fill other type-specific resources.
+ * Concrete publisher/subscriber client is expected to fill other
+ * type-specific resources.
*/
virtual ClientResourceBundle resourceBundle() {
ClientResourceBundle resource_bundle;
@@ -106,36 +111,38 @@
private:
/**
- * This is a low-level API that fetches route data from name server through gRPC unary request/response. Once
- * request/response is completed, either timeout or response arrival in time, callback would get invoked.
+ * This is a low-level API that fetches route data from name server through
+ * gRPC unary request/response. Once request/response is completed, either
+ * timeout or response arrival in time, callback would get invoked.
* @param topic
* @param cb
*/
- void fetchRouteFor(const std::string& topic, const std::function<void(const TopicRouteDataPtr&)>& cb);
+ void fetchRouteFor(const std::string& topic,
+ const std::function<void(const std::error_code&, const TopicRouteDataPtr&)>& cb);
/**
* Callback to execute once route data is fetched from name server.
* @param topic
* @param route
*/
- void onTopicRouteReady(const std::string& topic, const TopicRouteDataPtr& route)
+ void onTopicRouteReady(const std::string& topic, const std::error_code& ec, const TopicRouteDataPtr& route)
LOCKS_EXCLUDED(inflight_route_requests_mtx_);
/**
- * Update local cache for the topic. Note, route differences are logged in INFO level since route bears fundamental
- * importance.
+ * Update local cache for the topic. Note, route differences are logged in
+ * INFO level since route bears fundamental importance.
*
* @param topic
* @param route
*/
- void updateRouteCache(const std::string& topic, const TopicRouteDataPtr& route)
+ void updateRouteCache(const std::string& topic, const std::error_code& ec, const TopicRouteDataPtr& route)
LOCKS_EXCLUDED(topic_route_table_mtx_);
void multiplexing(const std::string& target, const MultiplexingRequest& request);
void onMultiplexingResponse(const InvocationContext<MultiplexingResponse>* ctx);
- void onHealthCheckResponse(const std::string& endpoint, const InvocationContext<HealthCheckResponse>* ctx)
+ void onHealthCheckResponse(const std::error_code& endpoint, const InvocationContext<HealthCheckResponse>* ctx)
LOCKS_EXCLUDED(isolated_endpoints_mtx_);
void fillGenericPollingRequest(MultiplexingRequest& request);
diff --git a/src/main/cpp/rocketmq/include/ConsumeMessageService.h b/src/main/cpp/rocketmq/include/ConsumeMessageService.h
index a205705..de4e62f 100644
--- a/src/main/cpp/rocketmq/include/ConsumeMessageService.h
+++ b/src/main/cpp/rocketmq/include/ConsumeMessageService.h
@@ -3,6 +3,7 @@
#include <memory>
#include <mutex>
#include <string>
+#include <system_error>
#include "ProcessQueue.h"
#include "RateLimiter.h"
@@ -123,7 +124,7 @@
void scheduleAckTask(const ProcessQueueWeakPtr& process_queue, const MQMessageExt& message);
- void onAck(const ProcessQueueWeakPtr& process_queue, const MQMessageExt& message, bool ok);
+ void onAck(const ProcessQueueWeakPtr& process_queue, const MQMessageExt& message, const std::error_code& ec);
void scheduleConsumeTask(const ProcessQueueWeakPtr& process_queue, const MQMessageExt& message);
diff --git a/src/main/cpp/rocketmq/include/ProducerImpl.h b/src/main/cpp/rocketmq/include/ProducerImpl.h
index 43459c2..263b11f 100644
--- a/src/main/cpp/rocketmq/include/ProducerImpl.h
+++ b/src/main/cpp/rocketmq/include/ProducerImpl.h
@@ -4,6 +4,7 @@
#include <memory>
#include <mutex>
#include <string>
+#include <system_error>
#include "absl/strings/string_view.h"
@@ -35,24 +36,15 @@
void shutdown() override;
- SendResult send(const MQMessage& message);
- SendResult send(const MQMessage& message, const std::string& message_group);
- SendResult send(const MQMessage& message, const MQMessageQueue& message_queue);
- SendResult send(const MQMessage& message, MessageQueueSelector* selector, void* arg);
-
- SendResult send(const MQMessage& message, MessageQueueSelector* selector, void* arg, int max_attempts);
+ SendResult send(const MQMessage& message, std::error_code& ec) noexcept;
void send(const MQMessage& message, SendCallback* callback);
- void send(const MQMessage& message, const MQMessageQueue& message_queue, SendCallback* callback);
- void send(const MQMessage& message, MessageQueueSelector* selector, void* arg, SendCallback* callback);
- void sendOneway(const MQMessage& message);
- void sendOneway(const MQMessage& message, const MQMessageQueue& message_queue);
- void sendOneway(const MQMessage& message, MessageQueueSelector* selector, void* arg);
+ void sendOneway(const MQMessage& message, std::error_code& ec);
void setLocalTransactionStateChecker(LocalTransactionStateCheckerPtr checker);
- std::unique_ptr<TransactionImpl> prepare(MQMessage& message);
+ std::unique_ptr<TransactionImpl> prepare(MQMessage& message, std::error_code& ec);
bool commit(const std::string& message_id, const std::string& transaction_id, const std::string& trace_context,
const std::string& target);
@@ -81,7 +73,7 @@
void setFailedTimes(int times) { failed_times_ = times; }
- std::vector<MQMessageQueue> getTopicMessageQueueInfo(const std::string& topic);
+ std::vector<MQMessageQueue> listMessageQueue(const std::string& topic, std::error_code& ec);
uint32_t compressBodyThreshold() const { return compress_body_threshold_; }
@@ -112,14 +104,12 @@
LocalTransactionStateCheckerPtr transaction_state_checker_;
- void asyncPublishInfo(const std::string& topic, const std::function<void(const TopicPublishInfoPtr&)>& cb)
+ void asyncPublishInfo(const std::string& topic,
+ const std::function<void(const std::error_code&, const TopicPublishInfoPtr&)>& cb)
LOCKS_EXCLUDED(topic_publish_info_mtx_);
TopicPublishInfoPtr getPublishInfo(const std::string& topic);
- bool executeMessageQueueSelector(const MQMessage& message, MessageQueueSelector* selector, void* arg,
- std::vector<MQMessageQueue>& result);
-
void takeMessageQueuesRoundRobin(const TopicPublishInfoPtr& publish_info, std::vector<MQMessageQueue>& message_queues,
int number);
@@ -128,7 +118,7 @@
bool isRunning() const;
- void ensureRunning() const;
+ void ensureRunning(std::error_code& ec) const noexcept;
bool validate(const MQMessage& message);
@@ -139,7 +129,7 @@
void isolatedEndpoints(absl::flat_hash_set<std::string>& endpoints) LOCKS_EXCLUDED(isolated_endpoints_mtx_);
- MQMessageQueue withServiceAddress(const MQMessageQueue& message_queue);
+ MQMessageQueue withServiceAddress(const MQMessageQueue& message_queue, std::error_code& ec);
};
ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/main/cpp/rocketmq/include/PushConsumer.h b/src/main/cpp/rocketmq/include/PushConsumer.h
index c71bd1e..4842727 100644
--- a/src/main/cpp/rocketmq/include/PushConsumer.h
+++ b/src/main/cpp/rocketmq/include/PushConsumer.h
@@ -2,6 +2,7 @@
#include <functional>
#include <memory>
+#include <system_error>
#include "ConsumeMessageService.h"
#include "Consumer.h"
@@ -20,7 +21,7 @@
virtual MessageModel messageModel() const = 0;
- virtual void ack(const MQMessageExt& msg, const std::function<void(bool)>& callback) = 0;
+ virtual void ack(const MQMessageExt& msg, const std::function<void(const std::error_code&)>& callback) = 0;
virtual void forwardToDeadLetterQueue(const MQMessageExt& message, const std::function<void(bool)>& cb) = 0;
@@ -32,7 +33,7 @@
virtual void updateOffset(const MQMessageQueue& message_queue, int64_t offset) = 0;
- virtual void nack(const MQMessageExt& message, const std::function<void(bool)>& callback) = 0;
+ virtual void nack(const MQMessageExt& message, const std::function<void(const std::error_code&)>& callback) = 0;
virtual std::shared_ptr<ConsumeMessageService> getConsumeMessageService() = 0;
diff --git a/src/main/cpp/rocketmq/include/PushConsumerImpl.h b/src/main/cpp/rocketmq/include/PushConsumerImpl.h
index c03653b..7499593 100644
--- a/src/main/cpp/rocketmq/include/PushConsumerImpl.h
+++ b/src/main/cpp/rocketmq/include/PushConsumerImpl.h
@@ -3,6 +3,7 @@
#include <memory>
#include <mutex>
#include <string>
+#include <system_error>
#include "absl/strings/string_view.h"
@@ -72,7 +73,8 @@
* @param topic Topic to query
* @return shared pointer to topic assignment info
*/
- void queryAssignment(const std::string& topic, const std::function<void(const TopicAssignmentPtr&)>& cb);
+ void queryAssignment(const std::string& topic,
+ const std::function<void(const std::error_code&, const TopicAssignmentPtr&)>& cb);
void syncProcessQueue(const std::string& topic, const TopicAssignmentPtr& topic_assignment,
const FilterExpression& filter_expression) LOCKS_EXCLUDED(process_queue_table_mtx_);
@@ -98,7 +100,7 @@
std::shared_ptr<ConsumeMessageService> getConsumeMessageService() override;
- void ack(const MQMessageExt& msg, const std::function<void(bool)>& callback) override;
+ void ack(const MQMessageExt& msg, const std::function<void(const std::error_code&)>& callback) override;
/**
* Negative acknowledge the given message; Refer to
@@ -109,7 +111,7 @@
*
* @param message Message to negate on the broker side.
*/
- void nack(const MQMessageExt& message, const std::function<void(bool)>& callback) override;
+ void nack(const MQMessageExt& message, const std::function<void(const std::error_code&)>& callback) override;
void forwardToDeadLetterQueue(const MQMessageExt& message, const std::function<void(bool)>& cb) override;
diff --git a/src/main/cpp/rocketmq/include/SendCallbacks.h b/src/main/cpp/rocketmq/include/SendCallbacks.h
index bdcc928..3063f93 100644
--- a/src/main/cpp/rocketmq/include/SendCallbacks.h
+++ b/src/main/cpp/rocketmq/include/SendCallbacks.h
@@ -1,6 +1,7 @@
#pragma once
#include <memory>
+#include <system_error>
#include "absl/container/flat_hash_map.h"
#include "absl/synchronization/mutex.h"
@@ -9,6 +10,7 @@
#include "TransactionImpl.h"
#include "rocketmq/AsyncCallback.h"
+#include "rocketmq/ErrorCode.h"
#include "rocketmq/MQMessage.h"
#include "rocketmq/MQMessageQueue.h"
@@ -18,69 +20,65 @@
class OnewaySendCallback : public SendCallback {
public:
- void onSuccess(SendResult &send_result) override;
+ void onSuccess(SendResult& send_result) noexcept override;
- void onException(const MQException &e) override;
+ void onFailure(const std::error_code& ec) noexcept override;
};
-OnewaySendCallback *onewaySendCallback();
+OnewaySendCallback* onewaySendCallback();
class AwaitSendCallback : public SendCallback {
public:
- void onSuccess(SendResult &send_result) override;
+ void onSuccess(SendResult& send_result) noexcept override;
- void onException(const MQException &e) override;
+ void onFailure(const std::error_code& ec) noexcept override;
void await();
- explicit operator bool() const { return success_; }
+ explicit operator bool() const { return !ec_.operator bool(); }
- const SendResult &sendResult() const { return send_result_; }
+ const SendResult& sendResult() const { return send_result_; }
- const std::string &errorMessage() const { return error_message_; }
+ const std::error_code& errorCode() const { return ec_; }
private:
absl::Mutex mtx_;
absl::CondVar cv_;
bool completed_{false};
- bool success_{false};
SendResult send_result_;
- std::string error_message_;
+ std::error_code ec_;
};
class ProducerImpl;
class RetrySendCallback : public SendCallback {
public:
- RetrySendCallback(std::weak_ptr<ProducerImpl> producer, MQMessage message,
- int max_attempt_times, SendCallback *callback,
- std::vector<MQMessageQueue> candidates)
- : producer_(std::move(producer)), message_(std::move(message)),
- max_attempt_times_(max_attempt_times), callback_(callback),
- candidates_(std::move(candidates)),
- span_(opencensus::trace::Span::BlankSpan()) {}
+ RetrySendCallback(std::weak_ptr<ProducerImpl> producer, MQMessage message, int max_attempt_times,
+ SendCallback* callback, std::vector<MQMessageQueue> candidates)
+ : producer_(std::move(producer)), message_(std::move(message)), max_attempt_times_(max_attempt_times),
+ callback_(callback), candidates_(std::move(candidates)), span_(opencensus::trace::Span::BlankSpan()) {}
- void onSuccess(SendResult &send_result) override;
+ void onSuccess(SendResult& send_result) noexcept override;
- void onException(const MQException &e) override;
+ void onFailure(const std::error_code& ec) noexcept override;
- MQMessage &message() { return message_; }
+ MQMessage& message() { return message_; }
int attemptTime() const { return attempt_times_; }
- const MQMessageQueue &messageQueue() const {
+ const MQMessageQueue& messageQueue() const {
int index = attempt_times_ % candidates_.size();
return candidates_[index];
}
- opencensus::trace::Span &span() { return span_; }
+ opencensus::trace::Span& span() { return span_; }
private:
std::weak_ptr<ProducerImpl> producer_;
MQMessage message_;
int attempt_times_{0};
int max_attempt_times_;
- SendCallback *callback_{nullptr};
+ SendCallback* callback_{nullptr};
/**
* @brief Once the first publish attempt failed, the following routable
diff --git a/src/main/cpp/rocketmq/mocks/include/PushConsumerMock.h b/src/main/cpp/rocketmq/mocks/include/PushConsumerMock.h
index feb9367..059577d 100644
--- a/src/main/cpp/rocketmq/mocks/include/PushConsumerMock.h
+++ b/src/main/cpp/rocketmq/mocks/include/PushConsumerMock.h
@@ -2,6 +2,7 @@
#include "ConsumerMock.h"
#include "PushConsumer.h"
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
@@ -11,7 +12,7 @@
MOCK_METHOD(MessageModel, messageModel, (), (const override));
- MOCK_METHOD(void, ack, (const MQMessageExt&, const std::function<void(bool)>&), (override));
+ MOCK_METHOD(void, ack, (const MQMessageExt&, const std::function<void(const std::error_code&)>&), (override));
MOCK_METHOD(void, forwardToDeadLetterQueue, (const MQMessageExt&, const std::function<void(bool)>&), (override));
@@ -23,7 +24,7 @@
MOCK_METHOD(void, updateOffset, (const MQMessageQueue&, int64_t), (override));
- MOCK_METHOD(void, nack, (const MQMessageExt&, const std::function<void(bool)>&), (override));
+ MOCK_METHOD(void, nack, (const MQMessageExt&, const std::function<void(const std::error_code&)>&), (override));
MOCK_METHOD(std::shared_ptr<ConsumeMessageService>, getConsumeMessageService, (), (override));
diff --git a/src/test/cpp/ut/api/BUILD.bazel b/src/test/cpp/ut/api/BUILD.bazel
new file mode 100644
index 0000000..eda5f4b
--- /dev/null
+++ b/src/test/cpp/ut/api/BUILD.bazel
@@ -0,0 +1,12 @@
+load("@rules_cc//cc:defs.bzl", "cc_test")
+
+cc_test(
+ name = "error_code_test",
+ srcs = [
+ "ErrorCodeTest.cpp",
+ ],
+ deps = [
+ "//src/main/cpp/base:base_library",
+ "@com_google_googletest//:gtest_main",
+ ],
+)
\ No newline at end of file
diff --git a/src/test/cpp/ut/api/ErrorCodeTest.cpp b/src/test/cpp/ut/api/ErrorCodeTest.cpp
new file mode 100644
index 0000000..42b525f
--- /dev/null
+++ b/src/test/cpp/ut/api/ErrorCodeTest.cpp
@@ -0,0 +1,14 @@
+#include "rocketmq/ErrorCode.h"
+#include "gtest/gtest.h"
+#include <system_error>
+
+ROCKETMQ_NAMESPACE_BEGIN
+
+TEST(ErrorCodeTest, testErrorCode) {
+ std::error_code ec = ErrorCode::BadRequest;
+ if (ec) {
+ std::cout << ec.message() << std::endl;
+ }
+}
+
+ROCKETMQ_NAMESPACE_END
\ No newline at end of file
diff --git a/src/test/cpp/ut/client/ClientManagerTest.cpp b/src/test/cpp/ut/client/ClientManagerTest.cpp
index 8cfa79e..e14ed02 100644
--- a/src/test/cpp/ut/client/ClientManagerTest.cpp
+++ b/src/test/cpp/ut/client/ClientManagerTest.cpp
@@ -4,6 +4,7 @@
#include "apache/rocketmq/v1/definition.pb.h"
#include "gtest/gtest.h"
#include <memory>
+#include <system_error>
ROCKETMQ_NAMESPACE_BEGIN
@@ -20,7 +21,9 @@
metadata_.insert({"name", "Donald.J.Trump"});
}
- void TearDown() override { client_manager_->shutdown(); }
+ void TearDown() override {
+ client_manager_->shutdown();
+ }
protected:
std::string resource_namespace_{"mq://test"};
@@ -65,7 +68,7 @@
QueryRouteRequest request;
request.mutable_topic()->set_resource_namespace(resource_namespace_);
request.mutable_topic()->set_name(topic_);
- auto callback = [&](bool, const TopicRouteDataPtr&) {
+ auto callback = [&](const std::error_code& ec, const TopicRouteDataPtr&) {
absl::MutexLock lk(&mtx);
completed = true;
cv.SignalAll();
@@ -97,7 +100,7 @@
.WillRepeatedly(testing::Invoke(mock_query_assignment));
QueryAssignmentRequest request;
bool callback_invoked = false;
- auto callback = [&](bool ok, const QueryAssignmentResponse& response) { callback_invoked = true; };
+ auto callback = [&](const std::error_code& ec, const QueryAssignmentResponse& response) { callback_invoked = true; };
client_manager_->queryAssignment(target_host_, metadata_, request, absl::ToChronoMilliseconds(io_timeout_), callback);
@@ -164,7 +167,7 @@
.WillRepeatedly(testing::Invoke(mock_async_receive));
ReceiveMessageRequest request;
- EXPECT_CALL(*receive_message_callback_, onException).Times(testing::AtLeast(1));
+ EXPECT_CALL(*receive_message_callback_, onFailure).Times(testing::AtLeast(1));
client_manager_->receiveMessage(target_host_, metadata_, request, absl::ToChronoMilliseconds(io_timeout_),
receive_message_callback_);
@@ -193,7 +196,7 @@
EXPECT_CALL(*rpc_client_, asyncAck).Times(testing::AtLeast(1)).WillRepeatedly(testing::Invoke(mock_ack));
AckMessageRequest request;
bool callback_invoked = false;
- auto callback = [&](bool ok) { callback_invoked = true; };
+ auto callback = [&](const std::error_code& ec) { callback_invoked = true; };
client_manager_->ack(target_host_, metadata_, request, absl::ToChronoMilliseconds(io_timeout_), callback);
@@ -222,7 +225,7 @@
EXPECT_CALL(*rpc_client_, asyncNack).Times(testing::AtLeast(1)).WillRepeatedly(testing::Invoke(mock_nack));
NackMessageRequest request;
bool callback_invoked = false;
- auto callback = [&](bool ok) { callback_invoked = true; };
+ auto callback = [&](const std::error_code& ec) { callback_invoked = true; };
client_manager_->nack(target_host_, metadata_, request, absl::ToChronoMilliseconds(io_timeout_), callback);
@@ -268,7 +271,8 @@
EXPECT_TRUE(callback_invoked);
}
-TEST_F(ClientManagerTest, testMultiplexingCall) {}
+TEST_F(ClientManagerTest, testMultiplexingCall) {
+}
TEST_F(ClientManagerTest, testEndTransaction) {
bool completed = false;
@@ -288,7 +292,7 @@
.WillRepeatedly(testing::Invoke(mock_end_transaction));
EndTransactionRequest request;
bool callback_invoked = false;
- auto callback = [&](bool ok, const EndTransactionResponse& response) { callback_invoked = true; };
+ auto callback = [&](const std::error_code& ec, const EndTransactionResponse& response) { callback_invoked = true; };
client_manager_->endTransaction(target_host_, metadata_, request, absl::ToChronoMilliseconds(io_timeout_), callback);
{
@@ -342,8 +346,9 @@
.WillRepeatedly(testing::Invoke(mock_health_check));
HealthCheckRequest request;
bool callback_invoked = false;
- auto callback = [&](const std::string& target_host,
- const InvocationContext<HealthCheckResponse>* invocation_context) { callback_invoked = true; };
+ auto callback = [&](const std::error_code& ec, const InvocationContext<HealthCheckResponse>* invocation_context) {
+ callback_invoked = true;
+ };
client_manager_->healthCheck(target_host_, metadata_, request, absl::ToChronoMilliseconds(io_timeout_), callback);
{
diff --git a/src/test/cpp/ut/rocketmq/AwaitPullCallbackTest.cpp b/src/test/cpp/ut/rocketmq/AwaitPullCallbackTest.cpp
index 3739c33..b286271 100644
--- a/src/test/cpp/ut/rocketmq/AwaitPullCallbackTest.cpp
+++ b/src/test/cpp/ut/rocketmq/AwaitPullCallbackTest.cpp
@@ -1,10 +1,13 @@
+#include <system_error>
+#include <thread>
+
#include "AwaitPullCallback.h"
#include "MessageAccessor.h"
+#include "rocketmq/ErrorCode.h"
#include "rocketmq/MQClientException.h"
#include "rocketmq/MQMessageExt.h"
#include "rocketmq/PullResult.h"
#include "gtest/gtest.h"
-#include <thread>
ROCKETMQ_NAMESPACE_BEGIN
@@ -51,8 +54,8 @@
AwaitPullCallback callback(pull_result);
std::string topic{"Test"};
auto task = [&]() {
- MQClientException e("test exception", -1, __FILE__, __LINE__);
- callback.onException(e);
+ std::error_code ec = ErrorCode::NotImplemented;
+ callback.onFailure(ec);
};
std::thread t(task);
diff --git a/src/test/cpp/ut/rocketmq/DefaultMQProducerTest.cpp b/src/test/cpp/ut/rocketmq/DefaultMQProducerTest.cpp
index 3d8b3d3..e7b8799 100644
--- a/src/test/cpp/ut/rocketmq/DefaultMQProducerTest.cpp
+++ b/src/test/cpp/ut/rocketmq/DefaultMQProducerTest.cpp
@@ -1,12 +1,14 @@
#include "rocketmq/DefaultMQProducer.h"
+#include <memory>
+#include <mutex>
+#include <system_error>
+#include <utility>
+
#include "MQClientTest.h"
#include "ProducerImpl.h"
#include "rocketmq/CredentialsProvider.h"
#include "rocketmq/MQSelector.h"
-#include <memory>
-#include <mutex>
-#include <utility>
ROCKETMQ_NAMESPACE_BEGIN
@@ -111,13 +113,15 @@
public:
UnitTestSendCallback(absl::Mutex& mtx, absl::CondVar& cv, std::string& msg_id, bool& completed)
: mtx_(mtx), cv_(cv), msg_id_(msg_id), completed_(completed) {}
- void onSuccess(SendResult& send_result) override {
+
+ void onSuccess(SendResult& send_result) noexcept override {
absl::MutexLock lk(&mtx_);
msg_id_ = send_result.getMsgId();
completed_ = true;
cv_.SignalAll();
}
- void onException(const MQException& e) override {
+
+ void onFailure(const std::error_code& e) noexcept override {
absl::MutexLock lk(&mtx_);
completed_ = true;
cv_.SignalAll();
@@ -136,7 +140,7 @@
producer->withNameServerResolver(name_server_resolver_);
producer->setCredentialsProvider(credentials_provider_);
producer->start();
-
+
MQMessage message;
message.setTopic(topic_);
message.setBody(body_);
@@ -166,7 +170,9 @@
message.setTopic(topic_);
message.setBody(body_);
- SendResult send_result = producer->send(message);
+ std::error_code ec;
+ SendResult send_result = producer->send(message, ec);
+ EXPECT_FALSE(ec);
ASSERT_EQ(send_result.getMsgId(), message_id_);
producer->shutdown();
}
diff --git a/src/test/cpp/ut/rocketmq/ProducerImplTest.cpp b/src/test/cpp/ut/rocketmq/ProducerImplTest.cpp
index a9e253b..b8375b3 100644
--- a/src/test/cpp/ut/rocketmq/ProducerImplTest.cpp
+++ b/src/test/cpp/ut/rocketmq/ProducerImplTest.cpp
@@ -1,4 +1,5 @@
#include <memory>
+#include <system_error>
#include "ClientManagerFactory.h"
#include "ClientManagerMock.h"
@@ -80,11 +81,13 @@
SchedulerImpl scheduler;
scheduler.start();
ON_CALL(*client_manager_, getScheduler).WillByDefault(testing::ReturnRef(scheduler));
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -103,7 +106,9 @@
producer_->start();
MQMessage message(topic_, tag_, message_body_);
- producer_->send(message);
+ std::error_code ec;
+ producer_->send(message, ec);
+ EXPECT_FALSE(ec);
EXPECT_TRUE(cb_invoked);
producer_->shutdown();
scheduler.shutdown();
@@ -113,11 +118,13 @@
SchedulerImpl scheduler;
scheduler.start();
ON_CALL(*client_manager_, getScheduler).WillByDefault(testing::ReturnRef(scheduler));
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -136,28 +143,26 @@
producer_->start();
MQMessage message(topic_, tag_, message_body_);
- producer_->send(message, message_group_);
+ message.bindMessageGroup(message_group_);
+ std::error_code ec;
+ producer_->send(message, ec);
+ EXPECT_FALSE(ec);
EXPECT_TRUE(cb_invoked);
producer_->shutdown();
scheduler.shutdown();
}
-class TestMessageQueueSelector : public MessageQueueSelector {
-public:
- MQMessageQueue select(const std::vector<MQMessageQueue>& mqs, const MQMessage& msg, void* arg) override {
- return *mqs.begin();
- }
-};
-
TEST_F(ProducerImplTest, testSend_WithMessageQueueSelector) {
SchedulerImpl scheduler;
scheduler.start();
ON_CALL(*client_manager_, getScheduler).WillByDefault(testing::ReturnRef(scheduler));
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -175,10 +180,17 @@
EXPECT_CALL(*client_manager_, send).Times(testing::AtLeast(1)).WillRepeatedly(testing::Invoke(mock_send));
producer_->start();
- auto selector = absl::make_unique<TestMessageQueueSelector>();
-
MQMessage message(topic_, tag_, message_body_);
- producer_->send(message, selector.get(), nullptr);
+
+ std::error_code ec;
+ auto&& list = producer_->listMessageQueue(topic_, ec);
+
+ EXPECT_FALSE(list.empty());
+
+ message.bindMessageQueue(list[0]);
+
+ producer_->send(message, ec);
+
EXPECT_TRUE(cb_invoked);
producer_->shutdown();
scheduler.shutdown();
@@ -187,12 +199,13 @@
class TestSendCallback : public SendCallback {
public:
TestSendCallback(bool& completed, absl::Mutex& mtx, absl::CondVar& cv) : completed_(completed), mtx_(mtx), cv_(cv) {}
- void onSuccess(SendResult& send_result) override {
+ void onSuccess(SendResult& send_result) noexcept override {
absl::MutexLock lk(&mtx_);
completed_ = true;
cv_.SignalAll();
}
- void onException(const MQException& e) override {
+
+ void onFailure(const std::error_code& ec) noexcept override {
absl::MutexLock lk(&mtx_);
completed_ = true;
cv_.SignalAll();
@@ -208,11 +221,13 @@
SchedulerImpl scheduler;
scheduler.start();
ON_CALL(*client_manager_, getScheduler).WillByDefault(testing::ReturnRef(scheduler));
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
diff --git a/src/test/cpp/ut/rocketmq/PullConsumerImplTest.cpp b/src/test/cpp/ut/rocketmq/PullConsumerImplTest.cpp
index 90f32d8..26d3de8 100644
--- a/src/test/cpp/ut/rocketmq/PullConsumerImplTest.cpp
+++ b/src/test/cpp/ut/rocketmq/PullConsumerImplTest.cpp
@@ -1,6 +1,7 @@
#include <chrono>
#include <memory>
#include <string>
+#include <system_error>
#include "ClientManagerFactory.h"
#include "ClientManagerMock.h"
@@ -78,11 +79,13 @@
TEST_F(PullConsumerImplTest, testQueuesFor) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -100,12 +103,12 @@
class TestPullCallback : public PullCallback {
public:
TestPullCallback(bool& success, bool& failure) : success_(success), failure_(failure) {}
- void onSuccess(const PullResult& pull_result) override {
+ void onSuccess(const PullResult& pull_result) noexcept override {
success_ = true;
failure_ = false;
}
- void onException(const MQException& e) override {
+ void onFailure(const std::error_code& e) noexcept override {
failure_ = true;
success_ = false;
}
@@ -117,11 +120,13 @@
TEST_F(PullConsumerImplTest, testPull) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -179,11 +184,13 @@
TEST_F(PullConsumerImplTest, testPull_gRPC_error) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -224,11 +231,13 @@
TEST_F(PullConsumerImplTest, testPull_biz_error) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -269,11 +278,13 @@
TEST_F(PullConsumerImplTest, testQueryOffset) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -289,7 +300,10 @@
auto mock_query_offset = [&](const std::string& target_host, const Metadata& metadata,
const QueryOffsetRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryOffsetResponse&)>& cb) { cb(true, response); };
+ const std::function<void(const std::error_code&, const QueryOffsetResponse&)>& cb) {
+ std::error_code ec;
+ cb(ec, response);
+ };
EXPECT_CALL(*client_manager_, queryOffset)
.Times(testing::AtLeast(1))
@@ -307,11 +321,13 @@
TEST_F(PullConsumerImplTest, testQueryOffset_End) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -327,7 +343,10 @@
auto mock_query_offset = [&](const std::string& target_host, const Metadata& metadata,
const QueryOffsetRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryOffsetResponse&)>& cb) { cb(true, response); };
+ const std::function<void(const std::error_code&, const QueryOffsetResponse&)>& cb) {
+ std::error_code ec;
+ cb(ec, response);
+ };
EXPECT_CALL(*client_manager_, queryOffset)
.Times(testing::AtLeast(1))
@@ -345,11 +364,13 @@
TEST_F(PullConsumerImplTest, testQueryOffset_Timepoint) {
pull_consumer_->start();
- auto mock_resolve_route = [this](const std::string& target_host, const Metadata& metadata,
- const QueryRouteRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const TopicRouteDataPtr& ptr)>& cb) {
- cb(true, topic_route_data_);
- };
+ auto mock_resolve_route =
+ [this](const std::string& target_host, const Metadata& metadata, const QueryRouteRequest& request,
+ std::chrono::milliseconds timeout,
+ const std::function<void(const std::error_code& ec, const TopicRouteDataPtr& ptr)>& cb) {
+ std::error_code ec;
+ cb(ec, topic_route_data_);
+ };
EXPECT_CALL(*client_manager_, resolveRoute)
.Times(testing::AtLeast(1))
@@ -365,7 +386,10 @@
auto mock_query_offset = [&](const std::string& target_host, const Metadata& metadata,
const QueryOffsetRequest& request, std::chrono::milliseconds timeout,
- const std::function<void(bool, const QueryOffsetResponse&)>& cb) { cb(true, response); };
+ const std::function<void(const std::error_code&, const QueryOffsetResponse&)>& cb) {
+ std::error_code ec;
+ cb(ec, response);
+ };
EXPECT_CALL(*client_manager_, queryOffset)
.Times(testing::AtLeast(1))
diff --git a/src/test/cpp/ut/rocketmq/PushConsumerImplTest.cpp b/src/test/cpp/ut/rocketmq/PushConsumerImplTest.cpp
index 0789af4..4bacd98 100644
--- a/src/test/cpp/ut/rocketmq/PushConsumerImplTest.cpp
+++ b/src/test/cpp/ut/rocketmq/PushConsumerImplTest.cpp
@@ -1,6 +1,7 @@
#include <memory>
#include "gtest/gtest.h"
+#include <system_error>
#include "ClientManagerFactory.h"
#include "ClientManagerMock.h"
@@ -11,7 +12,6 @@
#include "grpc/grpc.h"
#include "rocketmq/MQMessageExt.h"
#include "rocketmq/MessageListener.h"
-#include "rocketmq/RocketMQ.h"
ROCKETMQ_NAMESPACE_BEGIN
@@ -24,7 +24,8 @@
class PushConsumerImplTest : public testing::Test {
public:
- PushConsumerImplTest() : message_listener_(absl::make_unique<TestStandardMessageListener>()) {}
+ PushConsumerImplTest() : message_listener_(absl::make_unique<TestStandardMessageListener>()) {
+ }
void SetUp() override {
grpc_init();
@@ -39,7 +40,9 @@
push_consumer_->registerMessageListener(message_listener_.get());
}
- void TearDown() override { grpc_shutdown(); }
+ void TearDown() override {
+ grpc_shutdown();
+ }
protected:
std::string name_server_list_{"10.0.0.1:9876"};
@@ -63,7 +66,10 @@
ON_CALL(*client_manager_, getScheduler).WillByDefault(testing::ReturnRef(scheduler));
auto ack_cb = [](const std::string& target_host, const Metadata& metadata, const AckMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& cb) { cb(true); };
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& cb) {
+ std::error_code ec;
+ cb(ec);
+ };
EXPECT_CALL(*client_manager_, ack).Times(testing::AtLeast(1)).WillRepeatedly(testing::Invoke(ack_cb));
@@ -72,7 +78,7 @@
bool completed = false;
absl::Mutex mtx;
absl::CondVar cv;
- auto callback = [&](bool ok) {
+ auto callback = [&](const std::error_code& ec) {
absl::MutexLock lk(&mtx);
completed = true;
cv.SignalAll();
@@ -105,7 +111,10 @@
ON_CALL(*client_manager_, getScheduler).WillByDefault(testing::ReturnRef(scheduler));
auto nack_cb = [](const std::string& target_host, const Metadata& metadata, const NackMessageRequest& request,
- std::chrono::milliseconds timeout, const std::function<void(bool)>& cb) { cb(true); };
+ std::chrono::milliseconds timeout, const std::function<void(const std::error_code&)>& cb) {
+ std::error_code ec;
+ cb(ec);
+ };
EXPECT_CALL(*client_manager_, nack).Times(testing::AtLeast(1)).WillRepeatedly(testing::Invoke(nack_cb));
@@ -114,7 +123,7 @@
bool completed = false;
absl::Mutex mtx;
absl::CondVar cv;
- auto callback = [&](bool ok) {
+ auto callback = [&](const std::error_code& ec) {
absl::MutexLock lk(&mtx);
completed = true;
cv.SignalAll();
