[GitHub] [pulsar-client-cpp] Demogorgon314 commented on a diff in pull request #71: [feat] Support expiration for chunked messages

[GitBox]

Demogorgon314 commented on code in PR #71:
URL: https://github.com/apache/pulsar-client-cpp/pull/71#discussion_r1012465057


##########
lib/ConsumerImpl.cc:
##########
@@ -319,6 +320,31 @@ void ConsumerImpl::unsubscribeAsync(ResultCallback 
originalCallback) {
     }
 }
 
+void ConsumerImpl::triggerCheckExpiredChunkedTimer() {
+    checkExpiredChunkedTimer_->expires_from_now(
+        
boost::posix_time::milliseconds(expireTimeOfIncompleteChunkedMessageMs_));
+    std::weak_ptr<ConsumerImplBase> weakSelf{shared_from_this()};
+    checkExpiredChunkedTimer_->async_wait([this, weakSelf](const 
boost::system::error_code& ec) -> void {
+        auto self = weakSelf.lock();
+        if (!self) {
+            return;
+        }
+        if (ec) {
+            LOG_DEBUG(getName() << " Check expired chunked messages was failed 
or cancelled, code[" << ec
+                                << "].");
+            return;
+        }
+        Lock lock(chunkProcessMutex_);
+        long currentTimeMs = TimeUtils::currentTimeMillis();
+        chunkedMessageCache_.removeOldestValuesIf(
+            [this, &currentTimeMs](const std::string& uuid, const 
ChunkedMessageCtx& ctx) -> bool {
+                return currentTimeMs > ctx.getReceivedTimeMs() + 
expireTimeOfIncompleteChunkedMessageMs_;

Review Comment:
   Do we need to do ACK for the removed messageIds?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org