BewareMyPower commented on code in PR #426: URL: https://github.com/apache/pulsar-client-cpp/pull/426#discussion_r1618299101
########## lib/MultiTopicsConsumerImpl.h: ########## @@ -187,5 +194,42 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase { }; typedef std::shared_ptr<MultiTopicsConsumerImpl> MultiTopicsConsumerImplPtr; + +template <typename SeekArg> +#if __cplusplus >= 202002L + requires std::convertible_to<SeekArg, uint64_t> || + std::same_as<std::remove_cv_t<std::remove_reference_t<SeekArg>>, MessageId> +#endif + inline void MultiTopicsConsumerImpl::seekAllAsync(const SeekArg& seekArg, ResultCallback callback) { + if (state_ != Ready) { + callback(ResultAlreadyClosed); + return; + } + beforeSeek(); + auto weakSelf = weak_from_this(); + auto failed = std::make_shared<std::atomic_bool>(false); + consumers_.forEachValue( + [this, weakSelf, &seekArg, callback, failed](const ConsumerImplPtr& consumer, SharedFuture future) { + consumer->seekAsync(seekArg, [this, weakSelf, callback, failed, future](Result result) { + auto self = weakSelf.lock(); + if (!self || failed->load(std::memory_order_acquire)) { + callback(result); + return; + } + if (result != ResultOk) { + failed->store(true, std::memory_order_release); // skip the following callbacks + afterSeek(); + callback(result); + return; + } + if (future.tryComplete()) { + afterSeek(); + callback(ResultOk); + } + }); + }, + [callback] { callback(ResultOk); }); Review Comment: @shibd I make the `seek` call fail with `ResultOperationNotSupported` with a error log now. PTAL again. The discussion mail: https://lists.apache.org/thread/qrwvl1zshmdohphjtdyp9v98hdngxb30 -- 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