fix: improve code quality and add unit tests for various components

Author: noodanee

.gitignore

@@ -204,6 +204,7 @@ Debug/
 Release/
 toolchain/
 deps/rocketmq/bin
+deps/rocketmq/test/bin
 deps/rocketmq/tmp_*
 deps/rocketmq/libs/signature/lib
 -

deps/rocketmq/CMakeLists.txt

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-cmake_minimum_required(VERSION 3.15)
+cmake_minimum_required(VERSION 3.5)
 
 # CMake complains if we don't have this.
 if(COMMAND cmake_policy)

deps/rocketmq/build.sh

@@ -455,6 +455,7 @@ build_rocketmq_client() {
     -DCMAKE_LINKER="${LINK}"
     -DCMAKE_SYSTEM_PROCESSOR="${CMAKE_SYSTEM_PROCESSOR}"
     -DCMAKE_SYSTEM_NAME="${CMAKE_SYSTEM_NAME}"
+    -DCMAKE_POLICY_VERSION_MINIMUM=3.5
     -DLibevent_USE_STATIC_LIBS=ON
     -DJSONCPP_USE_STATIC_LIBS=ON
     -DZLIB_USE_STATIC_LIBS=ON
@@ -479,6 +480,11 @@ build_rocketmq_client() {
   log_info "Compiling RocketMQ C++ client (static + bundled library)"
   cmake --build "${BUILD_DIR}" --target rocketmq_static rocketmq_
 
+  if [ $TEST -eq 1 ]; then
+    log_info "Building unit test binaries"
+    cmake --build "${BUILD_DIR}"
+  fi
+
   # Uncomment if installation is needed
   # log_info "Installing RocketMQ C++ client"
   # cmake --install "${BUILD_DIR}"
@@ -524,6 +530,7 @@ build_googletest() {
     -DCMAKE_RANLIB="${RANLIB}" \
     -DCMAKE_LINKER="${LINK}" \
     -DCMAKE_CXX_FLAGS=-fPIC \
+    -DCMAKE_POLICY_VERSION_MINIMUM=3.5 \
     -DBUILD_STATIC_LIBS=ON \
     -DBUILD_SHARED_LIBS=OFF \
     -DCMAKE_INSTALL_PREFIX="${INSTALL_LIB_DIR}"

deps/rocketmq/include/DefaultLitePullConsumer.h

@@ -17,8 +17,11 @@
 #ifndef ROCKETMQ_DEFAULTLITEPULLCONSUMER_H_
 #define ROCKETMQ_DEFAULTLITEPULLCONSUMER_H_
 
+#include <memory>
+
 #include "DefaultLitePullConsumerConfigProxy.h"
 #include "LitePullConsumer.h"
+#include "PullResult.h"
 #include "RPCHook.h"
 
 namespace rocketmq {
@@ -67,6 +70,13 @@ class ROCKETMQCLIENT_API DefaultLitePullConsumer : public DefaultLitePullConsume
       const std::string& topic,
       TopicMessageQueueChangeListener* topicMessageQueueChangeListener) override;
 
+  std::unique_ptr<PullResult> pullOnce(const MQMessageQueue& mq,
+                                       const std::string& subExpression,
+                                       int64_t offset,
+                                       int maxNums,
+                                       bool block,
+                                       long timeoutMillis) override;
+
  public:
   void setRPCHook(RPCHookPtr rpcHook);
 

deps/rocketmq/include/LitePullConsumer.h

@@ -17,13 +17,17 @@
 #ifndef ROCKETMQ_LITEPULLCONSUMER_H_
 #define ROCKETMQ_LITEPULLCONSUMER_H_
 
+#include <memory>
+
 #include "TopicMessageQueueChangeListener.h"
 #include "MessageSelector.h"
 #include "MQMessageExt.h"
 #include "MQMessageQueue.h"
 
 namespace rocketmq {
 
+class PullResult;
+
 /**
  * LitePullConsumer - interface for pull consumer
  */
@@ -72,6 +76,13 @@ class ROCKETMQCLIENT_API LitePullConsumer {
   virtual void registerTopicMessageQueueChangeListener(
       const std::string& topic,
       TopicMessageQueueChangeListener* topicMessageQueueChangeListener) = 0;
+
+  virtual std::unique_ptr<PullResult> pullOnce(const MQMessageQueue& mq,
+                                               const std::string& subExpression,
+                                               int64_t offset,
+                                               int maxNums,
+                                               bool block,
+                                               long timeout_millis) = 0;
 };
 
 }  // namespace rocketmq

deps/rocketmq/src/MQClientAPIImpl.h

@@ -88,16 +88,19 @@ class MQClientAPIImpl {
 
   MQMessageExt viewMessage(const std::string& addr, int64_t phyoffset, int timeoutMillis);
 
-  int64_t searchOffset(const std::string& addr,
-                       const std::string& topic,
-                       int queueId,
-                       int64_t timestamp,
-                       int timeoutMillis);
+  virtual int64_t searchOffset(const std::string& addr,
+                               const std::string& topic,
+                               int queueId,
+                               int64_t timestamp,
+                               int timeoutMillis);
 
-  int64_t getMaxOffset(const std::string& addr, const std::string& topic, int queueId, int timeoutMillis);
-  int64_t getMinOffset(const std::string& addr, const std::string& topic, int queueId, int timeoutMillis);
+  virtual int64_t getMaxOffset(const std::string& addr, const std::string& topic, int queueId, int timeoutMillis);
+  virtual int64_t getMinOffset(const std::string& addr, const std::string& topic, int queueId, int timeoutMillis);
 
-  int64_t getEarliestMsgStoretime(const std::string& addr, const std::string& topic, int queueId, int timeoutMillis);
+  virtual int64_t getEarliestMsgStoretime(const std::string& addr,
+                                          const std::string& topic,
+                                          int queueId,
+                                          int timeoutMillis);
 
   void getConsumerIdListByGroup(const std::string& addr,
                                 const std::string& consumerGroup,
@@ -141,7 +144,7 @@ class MQClientAPIImpl {
                      int timeoutMillis,
                      bool oneway = false);
 
-  TopicRouteData* getTopicRouteInfoFromNameServer(const std::string& topic, int timeoutMillis);
+  virtual TopicRouteData* getTopicRouteInfoFromNameServer(const std::string& topic, int timeoutMillis);
 
   TopicList* getTopicListFromNameServer();
 

deps/rocketmq/src/common/SubscriptionGroupConfig.h

@@ -23,7 +23,7 @@ namespace rocketmq {
 
 class SubscriptionGroupConfig {
  public:
-  SubscriptionGroupConfig(const string& groupName) {
+  SubscriptionGroupConfig(const std::string& groupName) {
     this->groupName = groupName;
     consumeEnable = true;
     consumeFromMinEnable = true;

deps/rocketmq/src/common/VirtualEnvUtil.cpp

@@ -26,30 +26,41 @@ static const char* VIRTUAL_APPGROUP_PREFIX = "%%PROJECT_%s%%";
 namespace rocketmq {
 
 std::string VirtualEnvUtil::buildWithProjectGroup(const std::string& origin, const std::string& projectGroup) {
-  if (!UtilAll::isBlank(projectGroup)) {
-    char prefix[1024];
-    sprintf(prefix, VIRTUAL_APPGROUP_PREFIX, projectGroup.c_str());
-
-    if (origin.find_last_of(prefix) == std::string::npos) {
-      return origin + prefix;
-    } else {
-      return origin;
-    }
-  } else {
+  if (UtilAll::isBlank(projectGroup)) {
     return origin;
   }
+
+  char prefix[1024];
+  sprintf(prefix, VIRTUAL_APPGROUP_PREFIX, projectGroup.c_str());
+  std::string suffix(prefix);
+
+  if (origin.size() >= suffix.size() &&
+      origin.compare(origin.size() - suffix.size(), suffix.size(), suffix) == 0) {
+    return origin;
+  }
+
+  return origin + suffix;
 }
 
 std::string VirtualEnvUtil::clearProjectGroup(const std::string& origin, const std::string& projectGroup) {
+  if (UtilAll::isBlank(projectGroup)) {
+    return origin;
+  }
+
   char prefix[1024];
   sprintf(prefix, VIRTUAL_APPGROUP_PREFIX, projectGroup.c_str());
-  std::string::size_type pos = origin.find_last_of(prefix);
+  std::string suffix(prefix);
 
-  if (!UtilAll::isBlank(prefix) && pos != std::string::npos) {
-    return origin.substr(0, pos);
-  } else {
+  if (origin.size() < suffix.size()) {
     return origin;
   }
+
+  auto pos = origin.rfind(suffix);
+  if (pos != std::string::npos && pos + suffix.size() == origin.size()) {
+    return origin.substr(0, pos);
+  }
+
+  return origin;
 }
 
 }  // namespace rocketmq

deps/rocketmq/src/consumer/AssignedMessageQueue.hpp

@@ -65,6 +65,8 @@ class MessageQueueState {
 
 class AssignedMessageQueue {
  public:
+  AssignedMessageQueue() : rebalance_impl_(nullptr) {}
+
   std::vector<MQMessageQueue> messageQueues() {
     std::vector<MQMessageQueue> mqs;
     std::lock_guard<std::mutex> lock(assigned_message_queue_state_mutex_);

deps/rocketmq/src/consumer/DefaultLitePullConsumer.cpp

@@ -123,6 +123,16 @@ void DefaultLitePullConsumer::registerTopicMessageQueueChangeListener(
   pull_consumer_impl_->registerTopicMessageQueueChangeListener(topic, topicMessageQueueChangeListener);
 }
 
+std::unique_ptr<PullResult> DefaultLitePullConsumer::pullOnce(const MQMessageQueue& mq,
+                                                              const std::string& subExpression,
+                                                              int64_t offset,
+                                                              int maxNums,
+                                                              bool block,
+                                                              long timeoutMillis) {
+  long effectiveTimeout = timeoutMillis > 0 ? timeoutMillis : consumer_pull_timeout_millis();
+  return pull_consumer_impl_->pullOnce(mq, subExpression, offset, maxNums, block, effectiveTimeout);
+}
+
 void DefaultLitePullConsumer::setRPCHook(RPCHookPtr rpcHook) {
   dynamic_cast<DefaultLitePullConsumerImpl*>(pull_consumer_impl_.get())->setRPCHook(rpcHook);
 }