Fetch topic route with name server list attached (#15)

Author: lizhanhui

example/rocketmq/ExampleProducer.cpp

@@ -33,15 +33,19 @@ int main(int argc, char* argv[]) {
   logger.init();
 
   DefaultMQProducer producer("TestGroup");
-  producer.setNamesrvAddr("11.165.223.199:9876");
+
+  const char* topic = "cpp_sdk_standard";
+  const char* name_server = "47.98.116.189:80";
+
+  producer.setNamesrvAddr(name_server);
   producer.compressBodyThreshold(256);
-  const char* arn = "MQ_INST_1973281269661160_BXmPlOA6";
-  producer.setRegion("cn-hangzhou");
+  const char* arn = "MQ_INST_1080056302921134_BXuIbML7";
+  producer.setRegion("cn-hangzhou-pre");
   producer.arn(arn);
   producer.setCredentialsProvider(std::make_shared<ConfigFileCredentialsProvider>());
 
   MQMessage message;
-  message.setTopic("yc001");
+  message.setTopic(topic);
   message.setTags("TagA");
   message.setKey("Yuck! Why-plural?");
 

src/main/cpp/base/MQMessage.cpp

@@ -51,7 +51,7 @@ const std::string& MQMessage::getMsgId() const { return impl_->system_attribute_
 std::string MQMessage::getBornHost() const { return impl_->system_attribute_.born_host; }
 
 std::chrono::system_clock::time_point MQMessage::deliveryTimestamp() const {
-  absl::ToChronoTime(impl_->system_attribute_.delivery_timestamp);
+  return absl::ToChronoTime(impl_->system_attribute_.delivery_timestamp);
 }
 
 void MQMessage::setProperty(const std::string& name, const std::string& value) {

src/main/cpp/base/MixAll.cpp

@@ -30,6 +30,7 @@ const uint32_t MixAll::DEFAULT_CONSUME_MESSAGE_BATCH_SIZE = 1;
 const int32_t MixAll::DEFAULT_MAX_DELIVERY_ATTEMPTS = 16;
 
 const RE2 MixAll::TOPIC_REGEX("[a-zA-Z0-9\\-_]{3,64}");
+const RE2 MixAll::IP_REGEX("\\d+\\.\\d+\\.\\d+\\.\\d+");
 
 const std::chrono::duration<long long> MixAll::DEFAULT_INVISIBLE_TIME_ = std::chrono::seconds(30);
 
@@ -221,4 +222,8 @@ bool MixAll::homeDirectory(std::string& home_dir) {
 #endif
 }
 
+bool MixAll::isIPv4(absl::string_view host) {
+  return RE2::FullMatch(re2::StringPiece(host.data(), host.length()), IP_REGEX);
+}
+
 ROCKETMQ_NAMESPACE_END

src/main/cpp/base/include/MixAll.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "absl/strings/string_view.h"
 #include "re2/re2.h"
 #include "rocketmq/MQMessage.h"
 #include <chrono>
@@ -22,6 +23,7 @@ class MixAll {
   static const int32_t DEFAULT_MAX_DELIVERY_ATTEMPTS;
 
   static const RE2 TOPIC_REGEX;
+  static const RE2 IP_REGEX;
 
   /**
    * The amount of time required before a popped message is eligible to be consumed again. By default, 30s.
@@ -100,6 +102,8 @@ class MixAll {
 
   static bool homeDirectory(std::string& home);
 
+  static bool isIPv4(absl::string_view host);
+
 private:
   static bool hexCharValue(char c, uint8_t& value);
 };

src/main/cpp/log/LoggerImpl.cpp

@@ -68,7 +68,8 @@ void LoggerImpl::init0() {
   setLevel(console_sink_, console_level_);
   console_sink_->set_pattern(pattern_);
 
-  file_sink_ = std::make_shared<spdlog::sinks::rotating_file_sink_mt>(log_home_, file_size_, file_count_, true);
+  file_sink_ = std::make_shared<spdlog::sinks::rotating_file_sink_mt>(log_home_, file_size_, file_count_,
+                                                                      /*rotate_on_open=*/false);
   file_sink_->set_pattern(pattern_);
   setLevel(file_sink_, level_);
 

src/main/cpp/rocketmq/ClientImpl.cpp

@@ -6,11 +6,16 @@
 #include "MessageAccessor.h"
 #include "Signature.h"
 #include "absl/strings/str_join.h"
+#include "absl/strings/str_split.h"
 #include "apache/rocketmq/v1/definition.pb.h"
 #include "rocketmq/MQMessageExt.h"
+#include <algorithm>
 #include <chrono>
+#include <cstdint>
+#include <iterator>
 #include <memory>
 #include <string>
+#include <utility>
 
 ROCKETMQ_NAMESPACE_BEGIN
 
@@ -53,7 +58,7 @@ void ClientImpl::start() {
     };
     name_server_update_handle_ =
         client_manager_->getScheduler().schedule(name_server_update_functor, UPDATE_NAME_SERVER_LIST_TASK_NAME,
-                                                  std::chrono::milliseconds(0), std::chrono::seconds(30));
+                                                 std::chrono::milliseconds(0), std::chrono::seconds(30));
   }
 
   auto route_update_functor = [ptr]() {
@@ -64,7 +69,7 @@ void ClientImpl::start() {
   };
 
   route_update_handle_ = client_manager_->getScheduler().schedule(route_update_functor, UPDATE_ROUTE_TASK_NAME,
-                                                                   std::chrono::seconds(10), std::chrono::seconds(30));
+                                                                  std::chrono::seconds(10), std::chrono::seconds(30));
   state_.store(State::STARTED);
 }
 
@@ -236,6 +241,37 @@ void ClientImpl::fetchRouteFor(const std::string& topic, const std::function<voi
   QueryRouteRequest request;
   request.mutable_topic()->set_arn(arn_);
   request.mutable_topic()->set_name(topic);
+  auto endpoints = request.mutable_endpoints()->mutable_addresses();
+  std::vector<std::pair<std::string, std::uint16_t>> pairs;
+  {
+    absl::MutexLock lk(&name_server_list_mtx_);
+    for (const auto& name_server_item : name_server_list_) {
+      std::string::size_type pos = name_server_item.rfind(":");
+      if (std::string::npos == pos) {
+        continue;
+      }
+      std::string host(name_server_item.substr(0, pos));
+      std::string port(name_server_item.substr(pos + 1));
+      pairs.emplace_back(std::make_pair(host, std::stoi(port)));
+    }
+  }
+
+  if (!pairs.empty()) {
+    for (const auto& host_port : pairs) {
+      auto address = new rmq::Address();
+      address->set_port(host_port.second);
+      address->set_host(host_port.first);
+      endpoints->AddAllocated(address);
+    }
+
+    if (MixAll::isIPv4(pairs.begin()->first)) {
+      request.mutable_endpoints()->set_scheme(rmq::AddressScheme::IPv4);
+    } else if (absl::StrContains(pairs.begin()->first, ':')) {
+      request.mutable_endpoints()->set_scheme(rmq::AddressScheme::IPv6);
+    } else {
+      request.mutable_endpoints()->set_scheme(rmq::AddressScheme::DOMAIN_NAME);
+    }
+  }
 
   absl::flat_hash_map<std::string, std::string> metadata;
   Signature::sign(this, metadata);
@@ -346,7 +382,7 @@ void ClientImpl::multiplexing(const std::string& target, const MultiplexingReque
   absl::flat_hash_map<std::string, std::string> metadata;
   Signature::sign(this, metadata);
   client_manager_->multiplexingCall(target, metadata, request, absl::ToChronoMilliseconds(long_polling_timeout_),
-                                     std::bind(&ClientImpl::onMultiplexingResponse, this, std::placeholders::_1));
+                                    std::bind(&ClientImpl::onMultiplexingResponse, this, std::placeholders::_1));
 }
 
 void ClientImpl::onMultiplexingResponse(const InvocationContext<MultiplexingResponse>* ctx) {
@@ -359,7 +395,7 @@ void ClientImpl::onMultiplexingResponse(const InvocationContext<MultiplexingResp
     };
     static std::string task_name = "Initiate multiplex request later";
     client_manager_->getScheduler().schedule(multiplexingLater, task_name, std::chrono::seconds(3),
-                                              std::chrono::seconds(0));
+                                             std::chrono::seconds(0));
     return;
   }
 
@@ -373,7 +409,7 @@ void ClientImpl::onMultiplexingResponse(const InvocationContext<MultiplexingResp
     absl::flat_hash_map<std::string, std::string> metadata;
     Signature::sign(this, metadata);
     client_manager_->multiplexingCall(ctx->remote_address, metadata, request, absl::ToChronoMilliseconds(io_timeout_),
-                                       std::bind(&ClientImpl::onMultiplexingResponse, this, std::placeholders::_1));
+                                      std::bind(&ClientImpl::onMultiplexingResponse, this, std::placeholders::_1));
     break;
   }
 

src/test/cpp/ut/base/MixAllTest.cpp

@@ -8,7 +8,7 @@ class MixAllTest : public testing::Test {
 public:
   static std::string toUpperCase(const std::string& s) {
     std::string result;
-    for (const char & c : s) {
+    for (const char& c : s) {
       if ('a' <= c && 'z' >= c) {
         result.push_back(static_cast<char>('A' + (c - 'a')));
       } else {
@@ -94,4 +94,11 @@ TEST_F(MixAllTest, testCrc32) {
   std::cout << digest << std::endl;
 }
 
+TEST_F(MixAllTest, testIsIpv4) {
+  const char* ip = "8.8.8.8";
+  const char* host = "www.taobao.com";
+  EXPECT_TRUE(MixAll::isIPv4(ip));
+  EXPECT_FALSE(MixAll::isIPv4(host));
+}
+
 ROCKETMQ_NAMESPACE_END