consumer_configuration.h

 
#pragma once
 
#include <pulsar/defines.h>
 
#include "consumer.h"
#include "producer_configuration.h"
 
#ifdef __cplusplus
extern "C" {
#endif
 
typedef struct _pulsar_consumer_configuration pulsar_consumer_configuration_t;
 
typedef enum
{
    pulsar_ConsumerExclusive,
 
    pulsar_ConsumerShared,
 
    pulsar_ConsumerFailover,
 
    pulsar_ConsumerKeyShared
} pulsar_consumer_type;
 
typedef enum
{
    initial_position_latest,
    initial_position_earliest
} initial_position;
 
typedef enum
{
    // This is the default option to fail consume until crypto succeeds
    pulsar_ConsumerFail,
    // Message is silently acknowledged and not delivered to the application
    pulsar_ConsumerDiscard,
    // Deliver the encrypted message to the application. It's the application's
    // responsibility to decrypt the message. If message is also compressed,
    // decompression will fail. If message contain batch messages, client will
    // not be able to retrieve individual messages in the batch
    pulsar_ConsumerConsume
} pulsar_consumer_crypto_failure_action;
 
typedef enum
{
    // Only subscribe to persistent topics.
    pulsar_consumer_regex_sub_mode_PersistentOnly = 0,
    // Only subscribe to non-persistent topics.
    pulsar_consumer_regex_sub_mode_NonPersistentOnly = 1,
    // Subscribe to both persistent and non-persistent topics.
    pulsar_consumer_regex_sub_mode_AllTopics = 2
} pulsar_consumer_regex_subscription_mode;
 
// Though any field could be non-positive, if all of them are non-positive, this policy will be treated as
// invalid
typedef struct {
    // Max num messages, a non-positive value means no limit.
    int maxNumMessages;
    // Max num bytes, a non-positive value means no limit.
    long maxNumBytes;
    // The receive timeout, a non-positive value means no limit.
    long timeoutMs;
} pulsar_consumer_batch_receive_policy_t;
 
typedef struct {
    // Name of the dead topic where the failing messages are sent.
    // If it's null, use sourceTopicName + "-" + subscriptionName + "-DLQ" as the value
    const char *dead_letter_topic;
    // Maximum number of times that a message is redelivered before being sent to the dead letter queue.
    // If it's not greater than 0, treat it as INT_MAX, it means DLQ disable.
    int max_redeliver_count;
    // Name of the initial subscription name of the dead letter topic.
    // If it's null, the initial subscription for the dead letter topic is not created.
    // If this field is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer
    // fails to be created.
    const char *initial_subscription_name;
} pulsar_consumer_config_dead_letter_policy_t;
 
typedef void (*pulsar_message_listener)(pulsar_consumer_t *consumer, pulsar_message_t *msg, void *ctx);
 
PULSAR_PUBLIC pulsar_consumer_configuration_t *pulsar_consumer_configuration_create();
 
PULSAR_PUBLIC void pulsar_consumer_configuration_free(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_consumer_type(
    pulsar_consumer_configuration_t *consumer_configuration, pulsar_consumer_type consumerType);
 
PULSAR_PUBLIC pulsar_consumer_type
pulsar_consumer_configuration_get_consumer_type(pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_schema_info(
    pulsar_consumer_configuration_t *consumer_configuration, pulsar_schema_type schemaType, const char *name,
    const char *schema, pulsar_string_map_t *properties);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_message_listener(
    pulsar_consumer_configuration_t *consumer_configuration, pulsar_message_listener messageListener,
    void *ctx);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_has_message_listener(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_receiver_queue_size(
    pulsar_consumer_configuration_t *consumer_configuration, int size);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_get_receiver_queue_size(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_set_max_total_receiver_queue_size_across_partitions(
    pulsar_consumer_configuration_t *consumer_configuration, int maxTotalReceiverQueueSizeAcrossPartitions);
 
PULSAR_PUBLIC int pulsar_consumer_get_max_total_receiver_queue_size_across_partitions(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_set_consumer_name(pulsar_consumer_configuration_t *consumer_configuration,
                                                     const char *consumerName);
 
PULSAR_PUBLIC const char *pulsar_consumer_get_consumer_name(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_set_unacked_messages_timeout_ms(
    pulsar_consumer_configuration_t *consumer_configuration, const uint64_t milliSeconds);
 
PULSAR_PUBLIC long pulsar_consumer_get_unacked_messages_timeout_ms(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_configure_set_negative_ack_redelivery_delay_ms(
    pulsar_consumer_configuration_t *consumer_configuration, long redeliveryDelayMillis);
 
PULSAR_PUBLIC long pulsar_configure_get_negative_ack_redelivery_delay_ms(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_configure_set_ack_grouping_time_ms(
    pulsar_consumer_configuration_t *consumer_configuration, long ackGroupingMillis);
 
PULSAR_PUBLIC long pulsar_configure_get_ack_grouping_time_ms(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_configure_set_ack_grouping_max_size(
    pulsar_consumer_configuration_t *consumer_configuration, long maxGroupingSize);
 
PULSAR_PUBLIC long pulsar_configure_get_ack_grouping_max_size(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC int pulsar_consumer_is_encryption_enabled(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_default_crypto_key_reader(
    pulsar_consumer_configuration_t *consumer_configuration, const char *public_key_path,
    const char *private_key_path);
 
PULSAR_PUBLIC pulsar_consumer_crypto_failure_action pulsar_consumer_configuration_get_crypto_failure_action(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_crypto_failure_action(
    pulsar_consumer_configuration_t *consumer_configuration,
    pulsar_consumer_crypto_failure_action cryptoFailureAction);
 
PULSAR_PUBLIC int pulsar_consumer_is_read_compacted(pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_set_read_compacted(pulsar_consumer_configuration_t *consumer_configuration,
                                                      int compacted);
 
PULSAR_PUBLIC int pulsar_consumer_get_subscription_initial_position(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_set_subscription_initial_position(
    pulsar_consumer_configuration_t *consumer_configuration, initial_position subscriptionInitialPosition);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_property(pulsar_consumer_configuration_t *conf,
                                                              const char *name, const char *value);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_priority_level(
    pulsar_consumer_configuration_t *consumer_configuration, int priority_level);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_get_priority_level(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_max_pending_chunked_message(
    pulsar_consumer_configuration_t *consumer_configuration, int max_pending_chunked_message);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_get_max_pending_chunked_message(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_auto_ack_oldest_chunked_message_on_queue_full(
    pulsar_consumer_configuration_t *consumer_configuration,
    int auto_ack_oldest_chunked_message_on_queue_full);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_is_auto_ack_oldest_chunked_message_on_queue_full(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_start_message_id_inclusive(
    pulsar_consumer_configuration_t *consumer_configuration, int start_message_id_inclusive);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_is_start_message_id_inclusive(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_batch_index_ack_enabled(
    pulsar_consumer_configuration_t *consumer_configuration, int enabled);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_is_batch_index_ack_enabled(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_regex_subscription_mode(
    pulsar_consumer_configuration_t *consumer_configuration,
    pulsar_consumer_regex_subscription_mode regex_sub_mode);
 
PULSAR_PUBLIC pulsar_consumer_regex_subscription_mode
pulsar_consumer_configuration_get_regex_subscription_mode(
    pulsar_consumer_configuration_t *consumer_configuration);
 
PULSAR_PUBLIC int pulsar_consumer_configuration_set_batch_receive_policy(
    pulsar_consumer_configuration_t *consumer_configuration,
    const pulsar_consumer_batch_receive_policy_t *batch_receive_policy);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_get_batch_receive_policy(
    pulsar_consumer_configuration_t *consumer_configuration,
    pulsar_consumer_batch_receive_policy_t *batch_receive_policy);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_set_dlq_policy(
    pulsar_consumer_configuration_t *consumer_configuration,
    const pulsar_consumer_config_dead_letter_policy_t *dlq_policy);
 
PULSAR_PUBLIC void pulsar_consumer_configuration_get_dlq_policy(
    pulsar_consumer_configuration_t *consumer_configuration,
    pulsar_consumer_config_dead_letter_policy_t *dlq_policy);
 
// const CryptoKeyReaderPtr getCryptoKeyReader()
//
// const;
// ConsumerConfiguration&
// setCryptoKeyReader(CryptoKeyReaderPtr
// cryptoKeyReader);
//
// ConsumerCryptoFailureAction getCryptoFailureAction()
//
// const;
// ConsumerConfiguration&
// setCryptoFailureAction(ConsumerCryptoFailureAction
// action);
 
#ifdef __cplusplus
}
#endif