ConsumerConfiguration.h

 
#ifndef PULSAR_CONSUMERCONFIGURATION_H_
#define PULSAR_CONSUMERCONFIGURATION_H_
 
#include <pulsar/ConsumerCryptoFailureAction.h>
#include <pulsar/ConsumerEventListener.h>
#include <pulsar/ConsumerInterceptor.h>
#include <pulsar/ConsumerType.h>
#include <pulsar/CryptoKeyReader.h>
#include <pulsar/InitialPosition.h>
#include <pulsar/KeySharedPolicy.h>
#include <pulsar/Message.h>
#include <pulsar/RegexSubscriptionMode.h>
#include <pulsar/Result.h>
#include <pulsar/Schema.h>
#include <pulsar/TypedMessage.h>
#include <pulsar/defines.h>
 
#include <functional>
#include <memory>
 
#include "BatchReceivePolicy.h"
#include "DeadLetterPolicy.h"
 
namespace pulsar {
 
class Consumer;
class PulsarWrapper;
class PulsarFriend;
 
typedef std::vector<Message> Messages;
typedef std::function<void(Result result)> ResultCallback;
typedef std::function<void(Result, const Message& msg)> ReceiveCallback;
typedef std::function<void(Result, const Messages& msgs)> BatchReceiveCallback;
typedef std::function<void(Result result, MessageId messageId)> GetLastMessageIdCallback;
 
typedef std::function<void(Consumer& consumer, const Message& msg)> MessageListener;
 
typedef std::shared_ptr<ConsumerEventListener> ConsumerEventListenerPtr;
 
struct ConsumerConfigurationImpl;
 
class PULSAR_PUBLIC ConsumerConfiguration {
   public:
    ConsumerConfiguration();
    ~ConsumerConfiguration();
    ConsumerConfiguration(const ConsumerConfiguration&);
    ConsumerConfiguration& operator=(const ConsumerConfiguration&);
 
    ConsumerConfiguration clone() const;
 
    ConsumerConfiguration& setSchema(const SchemaInfo& schemaInfo);
 
    const SchemaInfo& getSchema() const;
 
    ConsumerConfiguration& setConsumerType(ConsumerType consumerType);
 
    ConsumerType getConsumerType() const;
 
    ConsumerConfiguration& setKeySharedPolicy(const KeySharedPolicy& keySharedPolicy);
 
    KeySharedPolicy getKeySharedPolicy() const;
 
    ConsumerConfiguration& setMessageListener(MessageListener messageListener);
 
    template <typename T>
    ConsumerConfiguration& setTypedMessageListener(
        std::function<void(Consumer&, const TypedMessage<T>&)> listener,
        typename TypedMessage<T>::Decoder decoder) {
        return setMessageListener([listener, decoder](Consumer& consumer, const Message& msg) {
            listener(consumer, TypedMessage<T>{msg, decoder});
        });
    }
 
    MessageListener getMessageListener() const;
 
    bool hasMessageListener() const;
 
    ConsumerConfiguration& setConsumerEventListener(ConsumerEventListenerPtr eventListener);
 
    ConsumerEventListenerPtr getConsumerEventListener() const;
 
    bool hasConsumerEventListener() const;
 
    void setReceiverQueueSize(int size);
 
    int getReceiverQueueSize() const;
 
    void setMaxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions);
 
    int getMaxTotalReceiverQueueSizeAcrossPartitions() const;
 
    void setConsumerName(const std::string& consumerName);
 
    const std::string& getConsumerName() const;
 
    void setUnAckedMessagesTimeoutMs(const uint64_t milliSeconds);
 
    long getUnAckedMessagesTimeoutMs() const;
 
    void setTickDurationInMs(const uint64_t milliSeconds);
 
    long getTickDurationInMs() const;
 
    void setNegativeAckRedeliveryDelayMs(long redeliveryDelayMillis);
 
    long getNegativeAckRedeliveryDelayMs() const;
 
    void setNegativeAckPrecisionBitCnt(int negativeAckPrecisionBitCnt);
 
    int getNegativeAckPrecisionBitCnt() const;
 
    void setAckGroupingTimeMs(long ackGroupingMillis);
 
    long getAckGroupingTimeMs() const;
 
    void setAckGroupingMaxSize(long maxGroupingSize);
 
    long getAckGroupingMaxSize() const;
 
    void setBrokerConsumerStatsCacheTimeInMs(const long cacheTimeInMs);
 
    long getBrokerConsumerStatsCacheTimeInMs() const;
 
    bool isEncryptionEnabled() const;
 
    const CryptoKeyReaderPtr getCryptoKeyReader() const;
 
    ConsumerConfiguration& setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader);
 
    ConsumerCryptoFailureAction getCryptoFailureAction() const;
 
    ConsumerConfiguration& setCryptoFailureAction(ConsumerCryptoFailureAction action);
 
    bool isReadCompacted() const;
 
    void setReadCompacted(bool compacted);
 
    void setPatternAutoDiscoveryPeriod(int periodInSeconds);
 
    int getPatternAutoDiscoveryPeriod() const;
 
    ConsumerConfiguration& setRegexSubscriptionMode(RegexSubscriptionMode regexSubscriptionMode);
 
    RegexSubscriptionMode getRegexSubscriptionMode() const;
 
    void setSubscriptionInitialPosition(InitialPosition subscriptionInitialPosition);
 
    InitialPosition getSubscriptionInitialPosition() const;
 
    void setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy);
 
    const BatchReceivePolicy& getBatchReceivePolicy() const;
 
    void setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy);
 
    const DeadLetterPolicy& getDeadLetterPolicy() const;
 
    void setReplicateSubscriptionStateEnabled(bool enabled);
 
    bool isReplicateSubscriptionStateEnabled() const;
 
    bool hasProperty(const std::string& name) const;
 
    const std::string& getProperty(const std::string& name) const;
 
    std::map<std::string, std::string>& getProperties() const;
 
    ConsumerConfiguration& setProperty(const std::string& name, const std::string& value);
 
    ConsumerConfiguration& setProperties(const std::map<std::string, std::string>& properties);
 
    std::map<std::string, std::string>& getSubscriptionProperties() const;
 
    ConsumerConfiguration& setSubscriptionProperties(
        const std::map<std::string, std::string>& subscriptionProperties);
 
    ConsumerConfiguration& setPriorityLevel(int priorityLevel);
 
    int getPriorityLevel() const;
 
    ConsumerConfiguration& setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage);
 
    size_t getMaxPendingChunkedMessage() const;
 
    ConsumerConfiguration& setAutoAckOldestChunkedMessageOnQueueFull(
        bool autoAckOldestChunkedMessageOnQueueFull);
 
    bool isAutoAckOldestChunkedMessageOnQueueFull() const;
 
    ConsumerConfiguration& setExpireTimeOfIncompleteChunkedMessageMs(
        long expireTimeOfIncompleteChunkedMessageMs);
 
    long getExpireTimeOfIncompleteChunkedMessageMs() const;
 
    ConsumerConfiguration& setStartMessageIdInclusive(bool startMessageIdInclusive);
 
    bool isStartMessageIdInclusive() const;
 
    ConsumerConfiguration& setBatchIndexAckEnabled(bool enabled);
 
    bool isBatchIndexAckEnabled() const;
 
    ConsumerConfiguration& intercept(const std::vector<ConsumerInterceptorPtr>& interceptors);
 
    const std::vector<ConsumerInterceptorPtr>& getInterceptors() const;
 
    ConsumerConfiguration& setAckReceiptEnabled(bool ackReceiptEnabled);
 
    bool isAckReceiptEnabled() const;
 
    ConsumerConfiguration& setStartPaused(bool startPaused);
 
    bool isStartPaused() const;
 
    friend class PulsarWrapper;
    friend class PulsarFriend;
 
   private:
    std::shared_ptr<ConsumerConfigurationImpl> impl_;
};
}  // namespace pulsar
#endif /* PULSAR_CONSUMERCONFIGURATION_H_ */