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
