include/pulsar/ConsumerConfiguration.h

/**
 * 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.
 */
#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;

/// Callback definition for non-data operation
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;

/// Callback definition for MessageListener
typedef std::function<void(Consumer& consumer, const Message& msg)> MessageListener;

typedef std::shared_ptr<ConsumerEventListener> ConsumerEventListenerPtr;

struct ConsumerConfigurationImpl;

/**
 * Class specifying the configuration of a consumer.
 */
class PULSAR_PUBLIC ConsumerConfiguration {
   public:
    ConsumerConfiguration();
    ~ConsumerConfiguration();
    ConsumerConfiguration(const ConsumerConfiguration&);
    ConsumerConfiguration& operator=(const ConsumerConfiguration&);

    /**
     * Create a new instance of ConsumerConfiguration with the same
     * initial settings as the current one.
     */
    ConsumerConfiguration clone() const;

    /**
     * Declare the schema of the data that this consumer will be accepting.
     *
     * The schema will be checked against the schema of the topic, and the
     * consumer creation will fail if it's not compatible.
     *
     * @param schemaInfo the schema definition object
     */
    ConsumerConfiguration& setSchema(const SchemaInfo& schemaInfo);

    /**
     * @return the schema information declared for this consumer
     */
    const SchemaInfo& getSchema() const;

    /**
     * Specify the consumer type. The consumer type enables
     * specifying the type of subscription. In Exclusive subscription,
     * only a single consumer is allowed to attach to the subscription. Other consumers
     * will get an error message. In Shared subscription, multiple consumers will be
     * able to use the same subscription name and the messages will be dispatched in a
     * round robin fashion. In Failover subscription, a primary-failover subscription model
     * allows for multiple consumers to attach to a single subscription, though only one
     * of them will be “master” at a given time. Only the primary consumer will receive
     * messages. When the primary consumer gets disconnected, one among the failover
     * consumers will be promoted to primary and will start getting messages.
     */
    ConsumerConfiguration& setConsumerType(ConsumerType consumerType);

    /**
     * @return the consumer type
     */
    ConsumerType getConsumerType() const;

    /**
     * Set KeyShared subscription policy for consumer.
     *
     * By default, KeyShared subscription use auto split hash range to maintain consumers. If you want to
     * set a different KeyShared policy, you can set by following example:
     *
     * @param keySharedPolicy The {@link KeySharedPolicy} want to specify
     */
    ConsumerConfiguration& setKeySharedPolicy(KeySharedPolicy keySharedPolicy);

    /**
     * @return the KeyShared subscription policy
     */
    KeySharedPolicy getKeySharedPolicy() const;

    /**
     * A message listener enables your application to configure how to process
     * and acknowledge messages delivered. A listener will be called in order
     * for every message received.
     */
    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});
        });
    }

    /**
     * @return the message listener
     */
    MessageListener getMessageListener() const;

    /**
     * @return true if the message listener has been set
     */
    bool hasMessageListener() const;

    /**
     * A event listener enables your application to react the consumer state
     * change event (active or inactive).
     */
    ConsumerConfiguration& setConsumerEventListener(ConsumerEventListenerPtr eventListener);

    /**
     * @return the consumer event listener
     */
    ConsumerEventListenerPtr getConsumerEventListener() const;

    /**
     * @return true if the consumer event listener has been set
     */
    bool hasConsumerEventListener() const;

    /**
     * Sets the size of the consumer receive queue.
     *
     * The consumer receive queue controls how many messages can be accumulated by the consumer before the
     * application calls receive(). Using a higher value may potentially increase the consumer throughput
     * at the expense of bigger memory utilization.
     *
     * Setting the consumer queue size to 0 decreases the throughput of the consumer by disabling
     * pre-fetching of
     * messages. This approach improves the message distribution on shared subscription by pushing messages
     * only to
     * the consumers that are ready to process them. Neither receive with timeout nor partitioned topics can
     * be
     * used if the consumer queue size is 0. The receive() function call should not be interrupted when
     * the consumer queue size is 0.
     *
     * The default value is 1000 messages and it is appropriate for the most use cases.
     *
     * @param size the new receiver queue size value
     *
     */
    void setReceiverQueueSize(int size);

    /**
     * @return the receiver queue size
     */
    int getReceiverQueueSize() const;

    /**
     * Set the max total receiver queue size across partitons.
     *
     * This setting is used to reduce the receiver queue size for individual partitions
     * {@link #setReceiverQueueSize(int)} if the total exceeds this value (default: 50000).
     *
     * @param maxTotalReceiverQueueSizeAcrossPartitions
     */
    void setMaxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions);

    /**
     * @return the configured max total receiver queue size across partitions
     */
    int getMaxTotalReceiverQueueSizeAcrossPartitions() const;

    /**
     * Set the consumer name.
     *
     * @param consumerName
     */
    void setConsumerName(const std::string& consumerName);

    /**
     * @return the consumer name
     */
    const std::string& getConsumerName() const;

    /**
     * Set the timeout in milliseconds for unacknowledged messages, the timeout needs to be greater than
     * 10 seconds. An Exception is thrown if the given value is less than 10000 (10 seconds).
     * If a successful acknowledgement is not sent within the timeout all the unacknowledged messages are
     * redelivered.
     *
     * Default: 0, which means the the tracker for unacknowledged messages is disabled.
     *
     * @param timeout in milliseconds
     */
    void setUnAckedMessagesTimeoutMs(const uint64_t milliSeconds);

    /**
     * @return the configured timeout in milliseconds for unacked messages.
     */
    long getUnAckedMessagesTimeoutMs() const;

    /**
     * Set the tick duration time that defines the granularity of the ack-timeout redelivery (in
     * milliseconds).
     *
     * The default value is 1000, which means 1 second.
     *
     * Using a higher tick time reduces
     * the memory overhead to track messages when the ack-timeout is set to a bigger value.
     *
     * @param milliSeconds the tick duration time (in milliseconds)
     */
    void setTickDurationInMs(const uint64_t milliSeconds);

    /**
     * @return the tick duration time (in milliseconds)
     */
    long getTickDurationInMs() const;

    /**
     * Set the delay to wait before re-delivering messages that have failed to be process.
     *
     * When application uses {@link Consumer#negativeAcknowledge(Message)}, the failed message
     * will be redelivered after a fixed timeout. The default is 1 min.
     *
     * @param redeliveryDelay
     *            redelivery delay for failed messages
     * @param timeUnit
     *            unit in which the timeout is provided.
     * @return the consumer builder instance
     */
    void setNegativeAckRedeliveryDelayMs(long redeliveryDelayMillis);

    /**
     * Get the configured delay to wait before re-delivering messages that have failed to be process.
     *
     * @return redelivery delay for failed messages
     */
    long getNegativeAckRedeliveryDelayMs() const;

    /**
     * Set time window in milliseconds for grouping message ACK requests. An ACK request is not sent
     * to broker until the time window reaches its end, or the number of grouped messages reaches
     * limit. Default is 100 milliseconds. If it's set to a non-positive value, ACK requests will be
     * directly sent to broker without grouping.
     *
     * @param ackGroupMillis time of ACK grouping window in milliseconds.
     */
    void setAckGroupingTimeMs(long ackGroupingMillis);

    /**
     * Get grouping time window in milliseconds.
     *
     * @return grouping time window in milliseconds.
     */
    long getAckGroupingTimeMs() const;

    /**
     * Set max number of grouped messages within one grouping time window. If it's set to a
     * non-positive value, number of grouped messages is not limited. Default is 1000.
     *
     * @param maxGroupingSize max number of grouped messages with in one grouping time window.
     */
    void setAckGroupingMaxSize(long maxGroupingSize);

    /**
     * Get max number of grouped messages within one grouping time window.
     *
     * @return max number of grouped messages within one grouping time window.
     */
    long getAckGroupingMaxSize() const;

    /**
     * Set the time duration for which the broker side consumer stats will be cached in the client.
     *
     * Default: 30000, which means 30 seconds.
     *
     * @param cacheTimeInMs in milliseconds
     */
    void setBrokerConsumerStatsCacheTimeInMs(const long cacheTimeInMs);

    /**
     * @return the configured timeout in milliseconds caching BrokerConsumerStats.
     */
    long getBrokerConsumerStatsCacheTimeInMs() const;

    /**
     * @return true if encryption keys are added
     */
    bool isEncryptionEnabled() const;

    /**
     * @return the shared pointer to CryptoKeyReader.
     */
    const CryptoKeyReaderPtr getCryptoKeyReader() const;

    /**
     * Set the shared pointer to CryptoKeyReader.
     *
     * @param the shared pointer to CryptoKeyReader
     */
    ConsumerConfiguration& setCryptoKeyReader(CryptoKeyReaderPtr cryptoKeyReader);

    /**
     * @return the ConsumerCryptoFailureAction
     */
    ConsumerCryptoFailureAction getCryptoFailureAction() const;

    /**
     * Set the ConsumerCryptoFailureAction.
     */
    ConsumerConfiguration& setCryptoFailureAction(ConsumerCryptoFailureAction action);

    /**
     * @return true if readCompacted is enabled
     */
    bool isReadCompacted() const;

    /**
     * If enabled, the consumer reads messages from the compacted topics rather than reading the full message
     * backlog of the topic. This means that if the topic has been compacted, the consumer only sees the
     * latest value for each key in the topic, up until the point in the topic message backlog that has been
     * compacted. Beyond that point, message is sent as normal.
     *
     * `readCompacted` can only be enabled subscriptions to persistent topics, which have a single active
     * consumer (for example, failure or exclusive subscriptions). Attempting to enable it on subscriptions to
     * a non-persistent topics or on a shared subscription leads to the subscription call failure.
     *
     * @param readCompacted
     *            whether to read from the compacted topic
     */
    void setReadCompacted(bool compacted);

    /**
     * Set the time duration in minutes, for which the PatternMultiTopicsConsumer will do a pattern auto
     * discovery.
     * The default value is 60 seconds. less than 0 will disable auto discovery.
     *
     * @param periodInSeconds       period in seconds to do an auto discovery
     */
    void setPatternAutoDiscoveryPeriod(int periodInSeconds);

    /**
     * @return the time duration for the PatternMultiTopicsConsumer performs a pattern auto discovery
     */
    int getPatternAutoDiscoveryPeriod() const;

    /**
     * Determines which topics this consumer should be subscribed to - Persistent, Non-Persistent, or
     * AllTopics. Only used with pattern subscriptions.
     *
     * @param regexSubscriptionMode The default value is `PersistentOnly`.
     */
    ConsumerConfiguration& setRegexSubscriptionMode(RegexSubscriptionMode regexSubscriptionMode);

    /**
     * @return the regex subscription mode for the pattern consumer.
     */
    RegexSubscriptionMode getRegexSubscriptionMode() const;

    /**
     * The default value is `InitialPositionLatest`.
     *
     * @param subscriptionInitialPosition the initial position at which to set
     * the cursor when subscribing to the topic for the first time
     */
    void setSubscriptionInitialPosition(InitialPosition subscriptionInitialPosition);

    /**
     * @return the configured `InitialPosition` for the consumer
     */
    InitialPosition getSubscriptionInitialPosition() const;

    /**
     * Set batch receive policy.
     *
     * @param batchReceivePolicy the default is
     * {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100}
     */
    void setBatchReceivePolicy(const BatchReceivePolicy& batchReceivePolicy);

    /**
     * Get batch receive policy.
     *
     * @return batch receive policy
     */
    const BatchReceivePolicy& getBatchReceivePolicy() const;

    /**
     * Set dead letter policy for consumer
     *
     * By default, some messages are redelivered many times, even to the extent that they can never be
     * stopped. By using the dead letter mechanism, messages have the max redelivery count, when they
     * exceeding the maximum number of redeliveries. Messages are sent to dead letter topics and acknowledged
     * automatically.
     *
     * You can enable the dead letter mechanism by setting the dead letter policy.
     * Example:
     *
     * <pre>
     * * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder()
     *                       .maxRedeliverCount(10)
     *                       .build();
     * </pre>
     * Default dead letter topic name is {TopicName}-{Subscription}-DLQ.
     * To set a custom dead letter topic name
     * <pre>
     * DeadLetterPolicy dlqPolicy = DeadLetterPolicyBuilder()
     *                       .deadLetterTopic("dlq-topic")
     *                       .maxRedeliverCount(10)
     *                       .initialSubscriptionName("init-sub-name")
     *                       .build();
     * </pre>
     * @param deadLetterPolicy Default value is empty
     */
    void setDeadLetterPolicy(const DeadLetterPolicy& deadLetterPolicy);

    /**
     * Get dead letter policy.
     *
     * @return dead letter policy
     */
    const DeadLetterPolicy& getDeadLetterPolicy() const;

    /**
     * Set whether the subscription status should be replicated.
     * The default value is `false`.
     *
     * @param replicateSubscriptionState whether the subscription status should be replicated
     */
    void setReplicateSubscriptionStateEnabled(bool enabled);

    /**
     * @return whether the subscription status should be replicated
     */
    bool isReplicateSubscriptionStateEnabled() const;

    /**
     * Check whether the message has a specific property attached.
     *
     * @param name the name of the property to check
     * @return true if the message has the specified property
     * @return false if the property is not defined
     */
    bool hasProperty(const std::string& name) const;

    /**
     * Get the value of a specific property
     *
     * @param name the name of the property
     * @return the value of the property or null if the property was not defined
     */
    const std::string& getProperty(const std::string& name) const;

    /**
     * Get all the properties attached to this producer.
     */
    std::map<std::string, std::string>& getProperties() const;

    /**
     * Sets a new property on a message.
     * @param name   the name of the property
     * @param value  the associated value
     */
    ConsumerConfiguration& setProperty(const std::string& name, const std::string& value);

    /**
     * Add all the properties in the provided map
     */
    ConsumerConfiguration& setProperties(const std::map<std::string, std::string>& properties);

    /**
     * Get all the subscription properties attached to this subscription.
     */
    std::map<std::string, std::string>& getSubscriptionProperties() const;

    /**
     * Sets a new subscription properties for this subscription.
     * Notice: SubscriptionProperties are immutable, and consumers under the same subscription will fail to
     * create a subscription if they use different properties.
     *
     * @param subscriptionProperties all the subscription properties in the provided map
     */
    ConsumerConfiguration& setSubscriptionProperties(
        const std::map<std::string, std::string>& subscriptionProperties);

    /**
     * Set the Priority Level for consumer (0 is the default value and means the highest priority).
     *
     * @param priorityLevel the priority of this consumer
     * @return the ConsumerConfiguration instance
     */
    ConsumerConfiguration& setPriorityLevel(int priorityLevel);

    /**
     * @return the configured priority for the consumer
     */
    int getPriorityLevel() const;

    /**
     * Consumer buffers chunk messages into memory until it receives all the chunks of the original message.
     * While consuming chunk-messages, chunks from same message might not be contiguous in the stream and they
     * might be mixed with other messages' chunks. so, consumer has to maintain multiple buffers to manage
     * chunks coming from different messages. This mainly happens when multiple publishers are publishing
     * messages on the topic concurrently or publisher failed to publish all chunks of the messages.
     *
     * eg: M1-C1, M2-C1, M1-C2, M2-C2
     * Here, Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 belong to M2 message.
     *
     * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it
     * can be guarded by providing this maxPendingChunkedMessage threshold. Once, consumer reaches this
     * threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver
     * later by marking it unacked. See setAutoAckOldestChunkedMessageOnQueueFull.
     *
     * If it's zero, the pending chunked messages will not be limited.
     *
     * Default: 10
     *
     * @param maxPendingChunkedMessage the number of max pending chunked messages
     */
    ConsumerConfiguration& setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage);

    /**
     * The associated getter of setMaxPendingChunkedMessage
     */
    size_t getMaxPendingChunkedMessage() const;

    /**
     * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it
     * can be guarded by providing the maxPendingChunkedMessage threshold. See setMaxPendingChunkedMessage.
     * Once, consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acking
     * if autoAckOldestChunkedMessageOnQueueFull is true else it marks them for redelivery.
     *
     * Default: false
     *
     * @param autoAckOldestChunkedMessageOnQueueFull whether to ack the discarded chunked message
     */
    ConsumerConfiguration& setAutoAckOldestChunkedMessageOnQueueFull(
        bool autoAckOldestChunkedMessageOnQueueFull);

    /**
     * The associated getter of setAutoAckOldestChunkedMessageOnQueueFull
     */
    bool isAutoAckOldestChunkedMessageOnQueueFull() const;

    /**
     * If producer fails to publish all the chunks of a message then consumer can expire incomplete chunks if
     * consumer won't be able to receive all chunks in expire times. Use value 0 to disable this feature.
     *
     * Default: 60000, which means 1 minutes
     *
     * @param expireTimeOfIncompleteChunkedMessageMs expire time in milliseconds
     * @return Consumer Configuration
     */
    ConsumerConfiguration& setExpireTimeOfIncompleteChunkedMessageMs(
        long expireTimeOfIncompleteChunkedMessageMs);

    /**
     *
     * Get the expire time of incomplete chunked message in milliseconds
     *
     * @return the expire time of incomplete chunked message in milliseconds
     */
    long getExpireTimeOfIncompleteChunkedMessageMs() const;

    /**
     * Set the consumer to include the given position of any reset operation like Consumer::seek.
     *
     * Default: false
     *
     * @param startMessageIdInclusive whether to include the reset position
     */
    ConsumerConfiguration& setStartMessageIdInclusive(bool startMessageIdInclusive);

    /**
     * The associated getter of setStartMessageIdInclusive
     */
    bool isStartMessageIdInclusive() const;

    /**
     * Enable the batch index acknowledgment.
     *
     * It should be noted that this option can only work when the broker side also enables the batch index
     * acknowledgment. See the `acknowledgmentAtBatchIndexLevelEnabled` config in `broker.conf`.
     *
     * Default: false
     *
     * @param enabled whether to enable the batch index acknowledgment
     */
    ConsumerConfiguration& setBatchIndexAckEnabled(bool enabled);

    /**
     * The associated getter of setBatchingEnabled
     */
    bool isBatchIndexAckEnabled() const;

    /**
     * Intercept the consumer
     *
     * @param interceptors the list of interceptors to intercept the consumer
     * @return Consumer Configuration
     */
    ConsumerConfiguration& intercept(const std::vector<ConsumerInterceptorPtr>& interceptors);

    const std::vector<ConsumerInterceptorPtr>& getInterceptors() const;

    /**
     * Whether to receive the ACK receipt from broker.
     *
     * By default, when Consumer::acknowledge is called, it won't wait until the corresponding response from
     * broker. After it's enabled, the `acknowledge` method will return a Result that indicates if the
     * acknowledgment succeeded.
     *
     * Default: false
     */
    ConsumerConfiguration& setAckReceiptEnabled(bool ackReceiptEnabled);

    /**
     * The associated getter of setAckReceiptEnabled.
     */
    bool isAckReceiptEnabled() const;

    /**
     * Starts the consumer in a paused state.
     *
     * When enabled, the consumer does not immediately fetch messages when the consumer is created.
     * Instead, the consumer waits to fetch messages until Consumer::resumeMessageListener is called.
     *
     * Default: false
     */
    ConsumerConfiguration& setStartPaused(bool startPaused);

    /**
     * The associated getter of setStartPaused.
     */
    bool isStartPaused() const;

    friend class PulsarWrapper;
    friend class PulsarFriend;

   private:
    std::shared_ptr<ConsumerConfigurationImpl> impl_;
};
}  // namespace pulsar
#endif /* PULSAR_CONSUMERCONFIGURATION_H_ */