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

Thu, 03 Nov 2022 00:38:55 -0700 


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


##########
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:
   Good point! The user should be able to choose the behavior(ack or just 
discard it).



-- 
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: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to