[fix] Avoid resource leakage of AckGroupingTracker

Author: erobot

lib/AckGroupingTrackerEnabled.cc

@@ -51,6 +51,7 @@ AckGroupingTrackerEnabled::AckGroupingTrackerEnabled(ClientImplPtr clientPtr,
                                                      const HandlerBasePtr& handlerPtr, uint64_t consumerId,
                                                      long ackGroupingTimeMs, long ackGroupingMaxSize)
     : AckGroupingTracker(),
+      state_(NotStarted),
       handlerWeakPtr_(handlerPtr),
       consumerId_(consumerId),
       nextCumulativeAckMsgId_(MessageId::earliest()),
@@ -67,7 +68,10 @@ AckGroupingTrackerEnabled::AckGroupingTrackerEnabled(ClientImplPtr clientPtr,
                                                         << ackGroupingMaxSize);
 }
 
-void AckGroupingTrackerEnabled::start() { this->scheduleTimer(); }
+void AckGroupingTrackerEnabled::start() {
+    state_ = Ready;
+    this->scheduleTimer();
+}
 
 bool AckGroupingTrackerEnabled::isDuplicate(const MessageId& msgId) {
     {
@@ -110,6 +114,7 @@ void AckGroupingTrackerEnabled::addAcknowledgeCumulative(const MessageId& msgId)
 }
 
 void AckGroupingTrackerEnabled::close() {
+    state_ = Closed;
     this->flush();
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     if (this->timer_) {
@@ -164,6 +169,10 @@ void AckGroupingTrackerEnabled::flushAndClean() {
 }
 
 void AckGroupingTrackerEnabled::scheduleTimer() {
+    if (state_ != Ready) {
+        return;
+    }
+
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     this->timer_ = this->executor_->createDeadlineTimer();
     this->timer_->expires_from_now(boost::posix_time::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));

lib/AckGroupingTrackerEnabled.h

@@ -71,6 +71,15 @@ class AckGroupingTrackerEnabled : public AckGroupingTracker {
     //! Method for scheduling grouping timer.
     void scheduleTimer();
 
+    //! State
+    enum State
+    {
+        NotStarted,
+        Ready,
+        Closed,
+    };
+    std::atomic<State> state_;
+
     //! The connection handler.
     HandlerBaseWeakPtr handlerWeakPtr_;
 

lib/ConsumerImpl.cc

@@ -1212,6 +1212,9 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
 const std::string& ConsumerImpl::getName() const { return consumerStr_; }
 
 void ConsumerImpl::shutdown() {
+    if (ackGroupingTrackerPtr_) {
+        ackGroupingTrackerPtr_->close();
+    }
     incomingMessages_.clear();
     possibleSendToDeadLetterTopicMessages_.clear();
     resetCnx();