lib/ConsumerConfiguration.cc - pulsar-client-cpp - Git at Google

 /**
  * 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 <pulsar/ConsumerConfiguration.h>

 #include <stdexcept>

 #include "ConsumerConfigurationImpl.h"

 namespace pulsar {

 const static std::string emptyString;

 ConsumerConfiguration::ConsumerConfiguration() : impl_(std::make_shared<ConsumerConfigurationImpl>()) {}

 ConsumerConfiguration::~ConsumerConfiguration() {}

 ConsumerConfiguration::ConsumerConfiguration(const ConsumerConfiguration& x) : impl_(x.impl_) {}

 ConsumerConfiguration& ConsumerConfiguration::operator=(const ConsumerConfiguration& x) {
     impl_ = x.impl_;
     return *this;
 }

 ConsumerConfiguration ConsumerConfiguration::clone() const {
     ConsumerConfiguration newConf;
     newConf.impl_.reset(new ConsumerConfigurationImpl(*this->impl_));
     return newConf;
 }

 ConsumerConfiguration& ConsumerConfiguration::setSchema(const SchemaInfo& schemaInfo) {
     impl_->schemaInfo = schemaInfo;
     return *this;
 }

 const SchemaInfo& ConsumerConfiguration::getSchema() const { return impl_->schemaInfo; }

 long ConsumerConfiguration::getBrokerConsumerStatsCacheTimeInMs() const {
     return impl_->brokerConsumerStatsCacheTimeInMs;
 }

 void ConsumerConfiguration::setBrokerConsumerStatsCacheTimeInMs(const long cacheTimeInMs) {
     impl_->brokerConsumerStatsCacheTimeInMs = cacheTimeInMs;
 }

 ConsumerConfiguration& ConsumerConfiguration::setConsumerType(ConsumerType consumerType) {
     impl_->consumerType = consumerType;
     return *this;
 }

 ConsumerType ConsumerConfiguration::getConsumerType() const { return impl_->consumerType; }

 ConsumerConfiguration& ConsumerConfiguration::setMessageListener(MessageListener messageListener) {
     impl_->messageListener = messageListener;
     impl_->hasMessageListener = true;
     return *this;
 }

 MessageListener ConsumerConfiguration::getMessageListener() const { return impl_->messageListener; }

 bool ConsumerConfiguration::hasMessageListener() const { return impl_->hasMessageListener; }

 ConsumerConfiguration& ConsumerConfiguration::setConsumerEventListener(
     ConsumerEventListenerPtr eventListener) {
     impl_->eventListener = eventListener;
     impl_->hasConsumerEventListener = true;
     return *this;
 }

 ConsumerEventListenerPtr ConsumerConfiguration::getConsumerEventListener() const {
     return impl_->eventListener;
 }

 bool ConsumerConfiguration::hasConsumerEventListener() const { return impl_->hasConsumerEventListener; }

 void ConsumerConfiguration::setReceiverQueueSize(int size) { impl_->receiverQueueSize = size; }

 int ConsumerConfiguration::getReceiverQueueSize() const { return impl_->receiverQueueSize; }

 void ConsumerConfiguration::setMaxTotalReceiverQueueSizeAcrossPartitions(int size) {
     impl_->maxTotalReceiverQueueSizeAcrossPartitions = size;
 }

 int ConsumerConfiguration::getMaxTotalReceiverQueueSizeAcrossPartitions() const {
     return impl_->maxTotalReceiverQueueSizeAcrossPartitions;
 }

 const std::string& ConsumerConfiguration::getConsumerName() const { return impl_->consumerName; }

 void ConsumerConfiguration::setConsumerName(const std::string& consumerName) {
     impl_->consumerName = consumerName;
 }

 long ConsumerConfiguration::getUnAckedMessagesTimeoutMs() const { return impl_->unAckedMessagesTimeoutMs; }

 void ConsumerConfiguration::setUnAckedMessagesTimeoutMs(const uint64_t milliSeconds) {
     if (milliSeconds < 10000 && milliSeconds != 0) {
         throw std::invalid_argument(
             "Consumer Config Exception: Unacknowledged message timeout should be greater than 10 seconds.");
     }
     impl_->unAckedMessagesTimeoutMs = milliSeconds;
 }

 long ConsumerConfiguration::getTickDurationInMs() const { return impl_->tickDurationInMs; }

 void ConsumerConfiguration::setTickDurationInMs(const uint64_t milliSeconds) {
     impl_->tickDurationInMs = milliSeconds;
 }

 void ConsumerConfiguration::setNegativeAckRedeliveryDelayMs(long redeliveryDelayMillis) {
     impl_->negativeAckRedeliveryDelayMs = redeliveryDelayMillis;
 }

 long ConsumerConfiguration::getNegativeAckRedeliveryDelayMs() const {
     return impl_->negativeAckRedeliveryDelayMs;
 }

 void ConsumerConfiguration::setAckGroupingTimeMs(long ackGroupingMillis) {
     impl_->ackGroupingTimeMs = ackGroupingMillis;
 }

 long ConsumerConfiguration::getAckGroupingTimeMs() const { return impl_->ackGroupingTimeMs; }

 void ConsumerConfiguration::setAckGroupingMaxSize(long maxGroupingSize) {
     impl_->ackGroupingMaxSize = maxGroupingSize;
 }

 long ConsumerConfiguration::getAckGroupingMaxSize() const { return impl_->ackGroupingMaxSize; }

 bool ConsumerConfiguration::isEncryptionEnabled() const { return (impl_->cryptoKeyReader != NULL); }

 const CryptoKeyReaderPtr ConsumerConfiguration::getCryptoKeyReader() const { return impl_->cryptoKeyReader; }

 ConsumerConfiguration& ConsumerConfiguration::setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader) {
     impl_->cryptoKeyReader = cryptoKeyReader;
     return *this;
 }

 ConsumerCryptoFailureAction ConsumerConfiguration::getCryptoFailureAction() const {
     return impl_->cryptoFailureAction;
 }

 ConsumerConfiguration& ConsumerConfiguration::setCryptoFailureAction(ConsumerCryptoFailureAction action) {
     impl_->cryptoFailureAction = action;
     return *this;
 }

 bool ConsumerConfiguration::isReadCompacted() const { return impl_->readCompacted; }

 void ConsumerConfiguration::setReadCompacted(bool compacted) { impl_->readCompacted = compacted; }

 void ConsumerConfiguration::setSubscriptionInitialPosition(InitialPosition subscriptionInitialPosition) {
     impl_->subscriptionInitialPosition = subscriptionInitialPosition;
 }

 InitialPosition ConsumerConfiguration::getSubscriptionInitialPosition() const {
     return impl_->subscriptionInitialPosition;
 }

 void ConsumerConfiguration::setPatternAutoDiscoveryPeriod(int periodInSeconds) {
     impl_->patternAutoDiscoveryPeriod = periodInSeconds;
 }

 int ConsumerConfiguration::getPatternAutoDiscoveryPeriod() const { return impl_->patternAutoDiscoveryPeriod; }

 void ConsumerConfiguration::setReplicateSubscriptionStateEnabled(bool enabled) {
     impl_->replicateSubscriptionStateEnabled = enabled;
 }

 bool ConsumerConfiguration::isReplicateSubscriptionStateEnabled() const {
     return impl_->replicateSubscriptionStateEnabled;
 }

 bool ConsumerConfiguration::hasProperty(const std::string& name) const {
     const std::map<std::string, std::string>& m = impl_->properties;
     return m.find(name) != m.end();
 }

 const std::string& ConsumerConfiguration::getProperty(const std::string& name) const {
     if (hasProperty(name)) {
         const std::map<std::string, std::string>& m = impl_->properties;
         return m.at(name);
     } else {
         return emptyString;
     }
 }

 std::map<std::string, std::string>& ConsumerConfiguration::getProperties() const { return impl_->properties; }

 ConsumerConfiguration& ConsumerConfiguration::setProperty(const std::string& name, const std::string& value) {
     impl_->properties.insert(std::make_pair(name, value));
     return *this;
 }

 ConsumerConfiguration& ConsumerConfiguration::setProperties(
     const std::map<std::string, std::string>& properties) {
     for (std::map<std::string, std::string>::const_iterator it = properties.begin(); it != properties.end();
          it++) {
         setProperty(it->first, it->second);
     }
     return *this;
 }

 std::map<std::string, std::string>& ConsumerConfiguration::getSubscriptionProperties() const {
     return impl_->subscriptionProperties;
 }

 ConsumerConfiguration& ConsumerConfiguration::setSubscriptionProperties(
     const std::map<std::string, std::string>& subscriptionProperties) {
     for (const auto& subscriptionProperty : subscriptionProperties) {
         impl_->subscriptionProperties.emplace(subscriptionProperty.first, subscriptionProperty.second);
     }
     return *this;
 }

 ConsumerConfiguration& ConsumerConfiguration::setPriorityLevel(int priorityLevel) {
     if (priorityLevel < 0) {
         throw std::invalid_argument("Consumer Config Exception: PriorityLevel should be nonnegative number.");
     }
     impl_->priorityLevel = priorityLevel;
     return *this;
 }

 int ConsumerConfiguration::getPriorityLevel() const { return impl_->priorityLevel; }

 ConsumerConfiguration& ConsumerConfiguration::setKeySharedPolicy(KeySharedPolicy keySharedPolicy) {
     impl_->keySharedPolicy = keySharedPolicy.clone();
     return *this;
 }

 KeySharedPolicy ConsumerConfiguration::getKeySharedPolicy() const { return impl_->keySharedPolicy; }

 ConsumerConfiguration& ConsumerConfiguration::setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage) {
     impl_->maxPendingChunkedMessage = maxPendingChunkedMessage;
     return *this;
 }

 size_t ConsumerConfiguration::getMaxPendingChunkedMessage() const { return impl_->maxPendingChunkedMessage; }

 ConsumerConfiguration& ConsumerConfiguration::setAutoAckOldestChunkedMessageOnQueueFull(
     bool autoAckOldestChunkedMessageOnQueueFull) {
     impl_->autoAckOldestChunkedMessageOnQueueFull = autoAckOldestChunkedMessageOnQueueFull;
     return *this;
 }

 bool ConsumerConfiguration::isAutoAckOldestChunkedMessageOnQueueFull() const {
     return impl_->autoAckOldestChunkedMessageOnQueueFull;
 }

 ConsumerConfiguration& ConsumerConfiguration::setExpireTimeOfIncompleteChunkedMessageMs(
     long expireTimeOfIncompleteChunkedMessageMs) {
     impl_->expireTimeOfIncompleteChunkedMessageMs = expireTimeOfIncompleteChunkedMessageMs;
     return *this;
 }

 long ConsumerConfiguration::getExpireTimeOfIncompleteChunkedMessageMs() const {
     return impl_->expireTimeOfIncompleteChunkedMessageMs;
 }

 ConsumerConfiguration& ConsumerConfiguration::setStartMessageIdInclusive(bool startMessageIdInclusive) {
     impl_->startMessageIdInclusive = startMessageIdInclusive;
     return *this;
 }

 bool ConsumerConfiguration::isStartMessageIdInclusive() const { return impl_->startMessageIdInclusive; }

 void ConsumerConfiguration::setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy) {
     impl_->batchReceivePolicy = batchReceivePolicy;
 }

 const BatchReceivePolicy& ConsumerConfiguration::getBatchReceivePolicy() const {
     return impl_->batchReceivePolicy;
 }

 ConsumerConfiguration& ConsumerConfiguration::setBatchIndexAckEnabled(bool enabled) {
     impl_->batchIndexAckEnabled = enabled;
     return *this;
 }

 bool ConsumerConfiguration::isBatchIndexAckEnabled() const { return impl_->batchIndexAckEnabled; }

 void ConsumerConfiguration::setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy) {
     impl_->deadLetterPolicy = deadLetterPolicy;
 }

 const DeadLetterPolicy& ConsumerConfiguration::getDeadLetterPolicy() const { return impl_->deadLetterPolicy; }

 ConsumerConfiguration& ConsumerConfiguration::intercept(
     const std::vector<ConsumerInterceptorPtr>& interceptors) {
     impl_->interceptors.insert(impl_->interceptors.end(), interceptors.begin(), interceptors.end());
     return *this;
 }

 const std::vector<ConsumerInterceptorPtr>& ConsumerConfiguration::getInterceptors() const {
     return impl_->interceptors;
 }

 ConsumerConfiguration& ConsumerConfiguration::setAckReceiptEnabled(bool ackReceiptEnabled) {
     impl_->ackReceiptEnabled = ackReceiptEnabled;
     return *this;
 }

 bool ConsumerConfiguration::isAckReceiptEnabled() const { return impl_->ackReceiptEnabled; }

 ConsumerConfiguration& ConsumerConfiguration::setRegexSubscriptionMode(
     RegexSubscriptionMode regexSubscriptionMode) {
     impl_->regexSubscriptionMode = regexSubscriptionMode;
     return *this;
 }

 RegexSubscriptionMode ConsumerConfiguration::getRegexSubscriptionMode() const {
     return impl_->regexSubscriptionMode;
 }

 }  // namespace pulsar
	/**
	* 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 <pulsar/ConsumerConfiguration.h>

	#include <stdexcept>

	#include "ConsumerConfigurationImpl.h"

	namespace pulsar {

	const static std::string emptyString;

	ConsumerConfiguration::ConsumerConfiguration() : impl_(std::make_shared<ConsumerConfigurationImpl>()) {}

	ConsumerConfiguration::~ConsumerConfiguration() {}

	ConsumerConfiguration::ConsumerConfiguration(const ConsumerConfiguration& x) : impl_(x.impl_) {}

	ConsumerConfiguration& ConsumerConfiguration::operator=(const ConsumerConfiguration& x) {
	impl_ = x.impl_;
	return *this;
	}

	ConsumerConfiguration ConsumerConfiguration::clone() const {
	ConsumerConfiguration newConf;
	newConf.impl_.reset(new ConsumerConfigurationImpl(*this->impl_));
	return newConf;
	}

	ConsumerConfiguration& ConsumerConfiguration::setSchema(const SchemaInfo& schemaInfo) {
	impl_->schemaInfo = schemaInfo;
	return *this;
	}

	const SchemaInfo& ConsumerConfiguration::getSchema() const { return impl_->schemaInfo; }

	long ConsumerConfiguration::getBrokerConsumerStatsCacheTimeInMs() const {
	return impl_->brokerConsumerStatsCacheTimeInMs;
	}

	void ConsumerConfiguration::setBrokerConsumerStatsCacheTimeInMs(const long cacheTimeInMs) {
	impl_->brokerConsumerStatsCacheTimeInMs = cacheTimeInMs;
	}

	ConsumerConfiguration& ConsumerConfiguration::setConsumerType(ConsumerType consumerType) {
	impl_->consumerType = consumerType;
	return *this;
	}

	ConsumerType ConsumerConfiguration::getConsumerType() const { return impl_->consumerType; }

	ConsumerConfiguration& ConsumerConfiguration::setMessageListener(MessageListener messageListener) {
	impl_->messageListener = messageListener;
	impl_->hasMessageListener = true;
	return *this;
	}

	MessageListener ConsumerConfiguration::getMessageListener() const { return impl_->messageListener; }

	bool ConsumerConfiguration::hasMessageListener() const { return impl_->hasMessageListener; }

	ConsumerConfiguration& ConsumerConfiguration::setConsumerEventListener(
	ConsumerEventListenerPtr eventListener) {
	impl_->eventListener = eventListener;
	impl_->hasConsumerEventListener = true;
	return *this;
	}

	ConsumerEventListenerPtr ConsumerConfiguration::getConsumerEventListener() const {
	return impl_->eventListener;
	}

	bool ConsumerConfiguration::hasConsumerEventListener() const { return impl_->hasConsumerEventListener; }

	void ConsumerConfiguration::setReceiverQueueSize(int size) { impl_->receiverQueueSize = size; }

	int ConsumerConfiguration::getReceiverQueueSize() const { return impl_->receiverQueueSize; }

	void ConsumerConfiguration::setMaxTotalReceiverQueueSizeAcrossPartitions(int size) {
	impl_->maxTotalReceiverQueueSizeAcrossPartitions = size;
	}

	int ConsumerConfiguration::getMaxTotalReceiverQueueSizeAcrossPartitions() const {
	return impl_->maxTotalReceiverQueueSizeAcrossPartitions;
	}

	const std::string& ConsumerConfiguration::getConsumerName() const { return impl_->consumerName; }

	void ConsumerConfiguration::setConsumerName(const std::string& consumerName) {
	impl_->consumerName = consumerName;
	}

	long ConsumerConfiguration::getUnAckedMessagesTimeoutMs() const { return impl_->unAckedMessagesTimeoutMs; }

	void ConsumerConfiguration::setUnAckedMessagesTimeoutMs(const uint64_t milliSeconds) {
	if (milliSeconds < 10000 && milliSeconds != 0) {
	throw std::invalid_argument(
	"Consumer Config Exception: Unacknowledged message timeout should be greater than 10 seconds.");
	}
	impl_->unAckedMessagesTimeoutMs = milliSeconds;
	}

	long ConsumerConfiguration::getTickDurationInMs() const { return impl_->tickDurationInMs; }

	void ConsumerConfiguration::setTickDurationInMs(const uint64_t milliSeconds) {
	impl_->tickDurationInMs = milliSeconds;
	}

	void ConsumerConfiguration::setNegativeAckRedeliveryDelayMs(long redeliveryDelayMillis) {
	impl_->negativeAckRedeliveryDelayMs = redeliveryDelayMillis;
	}

	long ConsumerConfiguration::getNegativeAckRedeliveryDelayMs() const {
	return impl_->negativeAckRedeliveryDelayMs;
	}

	void ConsumerConfiguration::setAckGroupingTimeMs(long ackGroupingMillis) {
	impl_->ackGroupingTimeMs = ackGroupingMillis;
	}

	long ConsumerConfiguration::getAckGroupingTimeMs() const { return impl_->ackGroupingTimeMs; }

	void ConsumerConfiguration::setAckGroupingMaxSize(long maxGroupingSize) {
	impl_->ackGroupingMaxSize = maxGroupingSize;
	}

	long ConsumerConfiguration::getAckGroupingMaxSize() const { return impl_->ackGroupingMaxSize; }

	bool ConsumerConfiguration::isEncryptionEnabled() const { return (impl_->cryptoKeyReader != NULL); }

	const CryptoKeyReaderPtr ConsumerConfiguration::getCryptoKeyReader() const { return impl_->cryptoKeyReader; }

	ConsumerConfiguration& ConsumerConfiguration::setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader) {
	impl_->cryptoKeyReader = cryptoKeyReader;
	return *this;
	}

	ConsumerCryptoFailureAction ConsumerConfiguration::getCryptoFailureAction() const {
	return impl_->cryptoFailureAction;
	}

	ConsumerConfiguration& ConsumerConfiguration::setCryptoFailureAction(ConsumerCryptoFailureAction action) {
	impl_->cryptoFailureAction = action;
	return *this;
	}

	bool ConsumerConfiguration::isReadCompacted() const { return impl_->readCompacted; }

	void ConsumerConfiguration::setReadCompacted(bool compacted) { impl_->readCompacted = compacted; }

	void ConsumerConfiguration::setSubscriptionInitialPosition(InitialPosition subscriptionInitialPosition) {
	impl_->subscriptionInitialPosition = subscriptionInitialPosition;
	}

	InitialPosition ConsumerConfiguration::getSubscriptionInitialPosition() const {
	return impl_->subscriptionInitialPosition;
	}

	void ConsumerConfiguration::setPatternAutoDiscoveryPeriod(int periodInSeconds) {
	impl_->patternAutoDiscoveryPeriod = periodInSeconds;
	}

	int ConsumerConfiguration::getPatternAutoDiscoveryPeriod() const { return impl_->patternAutoDiscoveryPeriod; }

	void ConsumerConfiguration::setReplicateSubscriptionStateEnabled(bool enabled) {
	impl_->replicateSubscriptionStateEnabled = enabled;
	}

	bool ConsumerConfiguration::isReplicateSubscriptionStateEnabled() const {
	return impl_->replicateSubscriptionStateEnabled;
	}

	bool ConsumerConfiguration::hasProperty(const std::string& name) const {
	const std::map<std::string, std::string>& m = impl_->properties;
	return m.find(name) != m.end();
	}

	const std::string& ConsumerConfiguration::getProperty(const std::string& name) const {
	if (hasProperty(name)) {
	const std::map<std::string, std::string>& m = impl_->properties;
	return m.at(name);
	} else {
	return emptyString;
	}
	}

	std::map<std::string, std::string>& ConsumerConfiguration::getProperties() const { return impl_->properties; }

	ConsumerConfiguration& ConsumerConfiguration::setProperty(const std::string& name, const std::string& value) {
	impl_->properties.insert(std::make_pair(name, value));
	return *this;
	}

	ConsumerConfiguration& ConsumerConfiguration::setProperties(
	const std::map<std::string, std::string>& properties) {
	for (std::map<std::string, std::string>::const_iterator it = properties.begin(); it != properties.end();
	it++) {
	setProperty(it->first, it->second);
	}
	return *this;
	}

	std::map<std::string, std::string>& ConsumerConfiguration::getSubscriptionProperties() const {
	return impl_->subscriptionProperties;
	}

	ConsumerConfiguration& ConsumerConfiguration::setSubscriptionProperties(
	const std::map<std::string, std::string>& subscriptionProperties) {
	for (const auto& subscriptionProperty : subscriptionProperties) {
	impl_->subscriptionProperties.emplace(subscriptionProperty.first, subscriptionProperty.second);
	}
	return *this;
	}

	ConsumerConfiguration& ConsumerConfiguration::setPriorityLevel(int priorityLevel) {
	if (priorityLevel < 0) {
	throw std::invalid_argument("Consumer Config Exception: PriorityLevel should be nonnegative number.");
	}
	impl_->priorityLevel = priorityLevel;
	return *this;
	}

	int ConsumerConfiguration::getPriorityLevel() const { return impl_->priorityLevel; }

	ConsumerConfiguration& ConsumerConfiguration::setKeySharedPolicy(KeySharedPolicy keySharedPolicy) {
	impl_->keySharedPolicy = keySharedPolicy.clone();
	return *this;
	}

	KeySharedPolicy ConsumerConfiguration::getKeySharedPolicy() const { return impl_->keySharedPolicy; }

	ConsumerConfiguration& ConsumerConfiguration::setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage) {
	impl_->maxPendingChunkedMessage = maxPendingChunkedMessage;
	return *this;
	}

	size_t ConsumerConfiguration::getMaxPendingChunkedMessage() const { return impl_->maxPendingChunkedMessage; }

	ConsumerConfiguration& ConsumerConfiguration::setAutoAckOldestChunkedMessageOnQueueFull(
	bool autoAckOldestChunkedMessageOnQueueFull) {
	impl_->autoAckOldestChunkedMessageOnQueueFull = autoAckOldestChunkedMessageOnQueueFull;
	return *this;
	}

	bool ConsumerConfiguration::isAutoAckOldestChunkedMessageOnQueueFull() const {
	return impl_->autoAckOldestChunkedMessageOnQueueFull;
	}

	ConsumerConfiguration& ConsumerConfiguration::setExpireTimeOfIncompleteChunkedMessageMs(
	long expireTimeOfIncompleteChunkedMessageMs) {
	impl_->expireTimeOfIncompleteChunkedMessageMs = expireTimeOfIncompleteChunkedMessageMs;
	return *this;
	}

	long ConsumerConfiguration::getExpireTimeOfIncompleteChunkedMessageMs() const {
	return impl_->expireTimeOfIncompleteChunkedMessageMs;
	}

	ConsumerConfiguration& ConsumerConfiguration::setStartMessageIdInclusive(bool startMessageIdInclusive) {
	impl_->startMessageIdInclusive = startMessageIdInclusive;
	return *this;
	}

	bool ConsumerConfiguration::isStartMessageIdInclusive() const { return impl_->startMessageIdInclusive; }

	void ConsumerConfiguration::setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy) {
	impl_->batchReceivePolicy = batchReceivePolicy;
	}

	const BatchReceivePolicy& ConsumerConfiguration::getBatchReceivePolicy() const {
	return impl_->batchReceivePolicy;
	}

	ConsumerConfiguration& ConsumerConfiguration::setBatchIndexAckEnabled(bool enabled) {
	impl_->batchIndexAckEnabled = enabled;
	return *this;
	}

	bool ConsumerConfiguration::isBatchIndexAckEnabled() const { return impl_->batchIndexAckEnabled; }

	void ConsumerConfiguration::setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy) {
	impl_->deadLetterPolicy = deadLetterPolicy;
	}

	const DeadLetterPolicy& ConsumerConfiguration::getDeadLetterPolicy() const { return impl_->deadLetterPolicy; }

	ConsumerConfiguration& ConsumerConfiguration::intercept(
	const std::vector<ConsumerInterceptorPtr>& interceptors) {
	impl_->interceptors.insert(impl_->interceptors.end(), interceptors.begin(), interceptors.end());
	return *this;
	}

	const std::vector<ConsumerInterceptorPtr>& ConsumerConfiguration::getInterceptors() const {
	return impl_->interceptors;
	}

	ConsumerConfiguration& ConsumerConfiguration::setAckReceiptEnabled(bool ackReceiptEnabled) {
	impl_->ackReceiptEnabled = ackReceiptEnabled;
	return *this;
	}

	bool ConsumerConfiguration::isAckReceiptEnabled() const { return impl_->ackReceiptEnabled; }

	ConsumerConfiguration& ConsumerConfiguration::setRegexSubscriptionMode(
	RegexSubscriptionMode regexSubscriptionMode) {
	impl_->regexSubscriptionMode = regexSubscriptionMode;
	return *this;
	}

	RegexSubscriptionMode ConsumerConfiguration::getRegexSubscriptionMode() const {
	return impl_->regexSubscriptionMode;
	}

	} // namespace pulsar