shibd commented on code in PR #132:
URL: https://github.com/apache/pulsar-client-cpp/pull/132#discussion_r1041064643
##########
lib/ConsumerImpl.cc:
##########
@@ -997,72 +997,86 @@ inline CommandSubscribe_InitialPosition
ConsumerImpl::getInitialPosition() {
BOOST_THROW_EXCEPTION(std::logic_error("Invalid InitialPosition
enumeration value"));
}
-void ConsumerImpl::statsAckCallback(Result res, ResultCallback callback,
CommandAck_AckType ackType,
- uint32_t numAcks) {
- consumerStatsBasePtr_->messageAcknowledged(res, ackType, numAcks);
- if (callback) {
- callback(res);
- }
-}
-
void ConsumerImpl::acknowledgeAsync(const MessageId& msgId, ResultCallback
callback) {
- ResultCallback cb = std::bind(&ConsumerImpl::statsAckCallback,
get_shared_this_ptr(),
- std::placeholders::_1, callback,
CommandAck_AckType_Individual, 1);
- if (msgId.batchIndex() != -1 &&
- !batchAcknowledgementTracker_.isBatchReady(msgId,
CommandAck_AckType_Individual)) {
- cb(ResultOk);
- return;
+ auto pair = prepareIndividualAck(msgId);
+ const auto& msgIdToAck = pair.first;
+ const bool readyToAck = pair.second;
+ if (readyToAck) {
+ ackGroupingTrackerPtr_->addAcknowledge(msgIdToAck);
+ }
+ if (callback) {
+ callback(ResultOk);
}
- doAcknowledgeIndividual(msgId, cb);
}
void ConsumerImpl::acknowledgeAsync(const MessageIdList& messageIdList,
ResultCallback callback) {
- ResultCallback cb =
- std::bind(&ConsumerImpl::statsAckCallback, get_shared_this_ptr(),
std::placeholders::_1, callback,
- proto::CommandAck_AckType_Individual, messageIdList.size());
- // Currently not supported batch message id individual index ack.
- this->ackGroupingTrackerPtr_->addAcknowledgeList(messageIdList);
- this->unAckedMessageTrackerPtr_->remove(messageIdList);
- cb(ResultOk);
+ MessageIdList messageIdListToAck;
+ for (auto&& msgId : messageIdList) {
+ auto pair = prepareIndividualAck(msgId);
+ const auto& msgIdToAck = pair.first;
+ const bool readyToAck = pair.second;
+ if (readyToAck) {
+ messageIdListToAck.emplace_back(msgIdToAck);
+ }
+ }
+ this->ackGroupingTrackerPtr_->addAcknowledgeList(messageIdListToAck);
+ if (callback) {
+ callback(ResultOk);
+ }
}
void ConsumerImpl::acknowledgeCumulativeAsync(const MessageId& msgId,
ResultCallback callback) {
- ResultCallback cb = std::bind(&ConsumerImpl::statsAckCallback,
get_shared_this_ptr(),
- std::placeholders::_1, callback,
CommandAck_AckType_Cumulative, 1);
if (!isCumulativeAcknowledgementAllowed(config_.getConsumerType())) {
- cb(ResultCumulativeAcknowledgementNotAllowedError);
+ if (callback) {
+ callback(ResultCumulativeAcknowledgementNotAllowedError);
+ }
return;
}
- if (msgId.batchIndex() != -1 &&
- !batchAcknowledgementTracker_.isBatchReady(msgId,
CommandAck_AckType_Cumulative)) {
- MessageId messageId =
batchAcknowledgementTracker_.getGreatestCumulativeAckReady(msgId);
- if (messageId == MessageId()) {
- // Nothing to ACK, because the batch that msgId belongs to is NOT
completely consumed.
- cb(ResultOk);
- } else {
- doAcknowledgeCumulative(messageId, cb);
- }
- } else {
- doAcknowledgeCumulative(msgId, cb);
+ auto pair = prepareCumulativeAck(msgId);
+ const auto& msgIdToAck = pair.first;
+ const auto& readyToAck = pair.second;
+ if (readyToAck) {
+ consumerStatsBasePtr_->messageAcknowledged(ResultOk,
CommandAck_AckType_Cumulative, 1);
+ unAckedMessageTrackerPtr_->removeMessagesTill(msgIdToAck);
+ ackGroupingTrackerPtr_->addAcknowledgeCumulative(msgIdToAck);
+ }
+ if (callback) {
+ callback(ResultOk);
}
}
bool ConsumerImpl::isCumulativeAcknowledgementAllowed(ConsumerType
consumerType) {
return consumerType != ConsumerKeyShared && consumerType != ConsumerShared;
}
-void ConsumerImpl::doAcknowledgeIndividual(const MessageId& messageId,
ResultCallback callback) {
- this->unAckedMessageTrackerPtr_->remove(messageId);
- this->batchAcknowledgementTracker_.deleteAckedMessage(messageId,
proto::CommandAck::Individual);
- this->ackGroupingTrackerPtr_->addAcknowledge(messageId);
- callback(ResultOk);
+std::pair<MessageId, bool> ConsumerImpl::prepareIndividualAck(const MessageId&
messageId) {
+ auto messageIdImpl = Commands::getMessageIdImpl(messageId);
+ auto batchedMessageIdImpl =
std::dynamic_pointer_cast<BatchedMessageIdImpl>(messageIdImpl);
+
+ auto batchSize = messageId.batchSize();
+ if (!batchedMessageIdImpl ||
batchedMessageIdImpl->ackIndividual(messageId.batchIndex())) {
Review Comment:
When messageId is not `BatchedMessageIdImpl`, the `!batchedMessageIdImpl`
condition is true, right?
--
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]
