| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| #include "ConsumerImplBase.h" |
| |
| #include <algorithm> |
| |
| #include "ConsumerImpl.h" |
| #include "ExecutorService.h" |
| #include "LogUtils.h" |
| #include "TimeUtils.h" |
| |
| DECLARE_LOG_OBJECT() |
| |
| namespace pulsar { |
| |
| ConsumerImplBase::ConsumerImplBase(ClientImplPtr client, const std::string& topic, Backoff backoff, |
| const ConsumerConfiguration& conf, ExecutorServicePtr listenerExecutor) |
| : HandlerBase(client, topic, backoff), |
| listenerExecutor_(listenerExecutor), |
| batchReceivePolicy_(conf.getBatchReceivePolicy()) { |
| auto userBatchReceivePolicy = conf.getBatchReceivePolicy(); |
| if (userBatchReceivePolicy.getMaxNumMessages() > conf.getReceiverQueueSize()) { |
| batchReceivePolicy_ = |
| BatchReceivePolicy(conf.getReceiverQueueSize(), userBatchReceivePolicy.getMaxNumBytes(), |
| userBatchReceivePolicy.getTimeoutMs()); |
| LOG_WARN("BatchReceivePolicy maxNumMessages: {" << userBatchReceivePolicy.getMaxNumMessages() |
| << "} is greater than maxReceiverQueueSize: {" |
| << conf.getReceiverQueueSize() |
| << "}, reset to " |
| "maxReceiverQueueSize. "); |
| } |
| batchReceiveTimer_ = listenerExecutor_->createDeadlineTimer(); |
| } |
| |
| void ConsumerImplBase::triggerBatchReceiveTimerTask(long timeoutMs) { |
| if (timeoutMs > 0) { |
| batchReceiveTimer_->expires_from_now(boost::posix_time::milliseconds(timeoutMs)); |
| std::weak_ptr<ConsumerImplBase> weakSelf{shared_from_this()}; |
| batchReceiveTimer_->async_wait([weakSelf](const boost::system::error_code& ec) { |
| auto self = weakSelf.lock(); |
| if (self && !ec) { |
| self->doBatchReceiveTimeTask(); |
| } |
| }); |
| } |
| } |
| |
| void ConsumerImplBase::doBatchReceiveTimeTask() { |
| if (state_ != Ready) { |
| return; |
| } |
| |
| bool hasPendingReceives = false; |
| long timeToWaitMs; |
| |
| Lock lock(batchPendingReceiveMutex_); |
| while (!batchPendingReceives_.empty()) { |
| OpBatchReceive& batchReceive = batchPendingReceives_.front(); |
| long diff = |
| batchReceivePolicy_.getTimeoutMs() - (TimeUtils::currentTimeMillis() - batchReceive.createAt_); |
| if (diff <= 0) { |
| Lock batchOptionLock(batchReceiveOptionMutex_); |
| notifyBatchPendingReceivedCallback(batchReceive.batchReceiveCallback_); |
| batchOptionLock.unlock(); |
| batchPendingReceives_.pop(); |
| } else { |
| hasPendingReceives = true; |
| timeToWaitMs = diff; |
| break; |
| } |
| } |
| lock.unlock(); |
| |
| if (hasPendingReceives) { |
| triggerBatchReceiveTimerTask(timeToWaitMs); |
| } |
| } |
| |
| void ConsumerImplBase::failPendingBatchReceiveCallback() { |
| Lock lock(batchPendingReceiveMutex_); |
| while (!batchPendingReceives_.empty()) { |
| OpBatchReceive opBatchReceive = batchPendingReceives_.front(); |
| batchPendingReceives_.pop(); |
| listenerExecutor_->postWork( |
| [opBatchReceive]() { opBatchReceive.batchReceiveCallback_(ResultAlreadyClosed, {}); }); |
| } |
| } |
| |
| void ConsumerImplBase::notifyBatchPendingReceivedCallback() { |
| Lock lock(batchPendingReceiveMutex_); |
| if (!batchPendingReceives_.empty()) { |
| OpBatchReceive& batchReceive = batchPendingReceives_.front(); |
| batchPendingReceives_.pop(); |
| lock.unlock(); |
| notifyBatchPendingReceivedCallback(batchReceive.batchReceiveCallback_); |
| } |
| } |
| |
| void ConsumerImplBase::batchReceiveAsync(BatchReceiveCallback callback) { |
| // fail the callback if consumer is closing or closed |
| if (state_ != Ready) { |
| callback(ResultAlreadyClosed, Messages()); |
| return; |
| } |
| |
| Lock batchOptionLock(batchReceiveOptionMutex_); |
| if (hasEnoughMessagesForBatchReceive()) { |
| notifyBatchPendingReceivedCallback(callback); |
| batchOptionLock.unlock(); |
| } else { |
| OpBatchReceive opBatchReceive(callback); |
| Lock lock(batchPendingReceiveMutex_); |
| batchPendingReceives_.emplace(opBatchReceive); |
| lock.unlock(); |
| triggerBatchReceiveTimerTask(batchReceivePolicy_.getTimeoutMs()); |
| } |
| } |
| |
| OpBatchReceive::OpBatchReceive(const BatchReceiveCallback& batchReceiveCallback) |
| : batchReceiveCallback_(batchReceiveCallback), createAt_(TimeUtils::currentTimeMillis()) {} |
| |
| } /* namespace pulsar */ |