text: /*
   * librdkafka - Apache Kafka C library
   *
   * Copyright (c) 2014-2018 Magnus Edenhill
   * All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions are met:
   *
   * 1. Redistributions of source code must retain the above copyright notice,
   *    this list of conditions and the following disclaimer.
   * 2. Redistributions in binary form must reproduce the above copyright notice,
   *    this list of conditions and the following disclaimer in the documentation
   *    and/or other materials provided with the distribution.
   *
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
   * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
   * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
   * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
   * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
   * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
   * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
   * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
   * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   * POSSIBILITY OF SUCH DAMAGE.
   */
  
  #ifndef _RDKAFKA_CONF_H_
  #define _RDKAFKA_CONF_H_
  
  #include "rdlist.h"
  
  
  /**
   * Forward declarations
   */
  struct rd_kafka_transport_s;
  
  
  /**
   * MessageSet compression codecs
   */
  typedef enum {
  	RD_KAFKA_COMPRESSION_NONE,
  	RD_KAFKA_COMPRESSION_GZIP = RD_KAFKA_MSG_ATTR_GZIP,
  	RD_KAFKA_COMPRESSION_SNAPPY = RD_KAFKA_MSG_ATTR_SNAPPY,
  	RD_KAFKA_COMPRESSION_LZ4 = RD_KAFKA_MSG_ATTR_LZ4,
  	RD_KAFKA_COMPRESSION_ZSTD = RD_KAFKA_MSG_ATTR_ZSTD,
  	RD_KAFKA_COMPRESSION_INHERIT, /* Inherit setting from global conf */
          RD_KAFKA_COMPRESSION_NUM
  } rd_kafka_compression_t;
  
  static RD_INLINE RD_UNUSED const char *
  rd_kafka_compression2str (rd_kafka_compression_t compr) {
          static const char *names[RD_KAFKA_COMPRESSION_NUM] = {
                  [RD_KAFKA_COMPRESSION_NONE] = "none",
                  [RD_KAFKA_COMPRESSION_GZIP] = "gzip",
                  [RD_KAFKA_COMPRESSION_SNAPPY] = "snappy",
                  [RD_KAFKA_COMPRESSION_LZ4] = "lz4",
                  [RD_KAFKA_COMPRESSION_ZSTD] = "zstd",
                  [RD_KAFKA_COMPRESSION_INHERIT] = "inherit"
          };
          return names[compr];
  }
  
  /**
   * MessageSet compression levels
   */
  typedef enum {
  	RD_KAFKA_COMPLEVEL_DEFAULT = -1,
  	RD_KAFKA_COMPLEVEL_MIN = -1,
  	RD_KAFKA_COMPLEVEL_GZIP_MAX = 9,
  	RD_KAFKA_COMPLEVEL_LZ4_MAX = 12,
  	RD_KAFKA_COMPLEVEL_SNAPPY_MAX = 0,
  	RD_KAFKA_COMPLEVEL_ZSTD_MAX = 22,
  	RD_KAFKA_COMPLEVEL_MAX = 12
  } rd_kafka_complevel_t;
  
  typedef enum {
  	RD_KAFKA_PROTO_PLAINTEXT,
  	RD_KAFKA_PROTO_SSL,
  	RD_KAFKA_PROTO_SASL_PLAINTEXT,
  	RD_KAFKA_PROTO_SASL_SSL,
  	RD_KAFKA_PROTO_NUM,
  } rd_kafka_secproto_t;
  
  
  typedef enum {
  	RD_KAFKA_CONFIGURED,
  	RD_KAFKA_LEARNED,
  	RD_KAFKA_INTERNAL,
          RD_KAFKA_LOGICAL
  } rd_kafka_confsource_t;
  
  static RD_INLINE RD_UNUSED
  const char *rd_kafka_confsource2str (rd_kafka_confsource_t source) {
          static const char *names[] = {
                  "configured",
                  "learned",
                  "internal",
                  "logical"
          };
  
          return names[source];
  }
  
  
  typedef	enum {
  	_RK_GLOBAL = 0x1,
  	_RK_PRODUCER = 0x2,
  	_RK_CONSUMER = 0x4,
  	_RK_TOPIC = 0x8,
          _RK_CGRP = 0x10,
          _RK_DEPRECATED = 0x20,
          _RK_HIDDEN = 0x40,
          _RK_HIGH = 0x80, /* High Importance */
          _RK_MED = 0x100  /* Medium Importance */
  } rd_kafka_conf_scope_t;
  
  /**< While the client groups is a generic concept, it is currently
   *   only implemented for consumers in librdkafka. */
  #define _RK_CGRP _RK_CONSUMER
  
  typedef enum {
  	_RK_CONF_PROP_SET_REPLACE,  /* Replace current value (default) */
  	_RK_CONF_PROP_SET_ADD,      /* Add value (S2F) */
  	_RK_CONF_PROP_SET_DEL      /* Remove value (S2F) */
  } rd_kafka_conf_set_mode_t;
  
  
  
  typedef enum {
          RD_KAFKA_OFFSET_METHOD_NONE,
          RD_KAFKA_OFFSET_METHOD_FILE,
          RD_KAFKA_OFFSET_METHOD_BROKER
  } rd_kafka_offset_method_t;
  
  
  
  /* Increase in steps of 64 as needed. */
  #define RD_KAFKA_CONF_PROPS_IDX_MAX (64*24)
  
  /**
   * @struct rd_kafka_anyconf_t
   * @brief The anyconf header must be the first field in the
   *        rd_kafka_conf_t and rd_kafka_topic_conf_t structs.
   *        It provides a way to track which property has been modified.
   */
  struct rd_kafka_anyconf_hdr {
          uint64_t modified[RD_KAFKA_CONF_PROPS_IDX_MAX/64];
  };
  
  
  /**
   * Optional configuration struct passed to rd_kafka_new*().
   *
   * The struct is populated ted through string properties
   * by calling rd_kafka_conf_set().
   *
   */
  struct rd_kafka_conf_s {
          struct rd_kafka_anyconf_hdr hdr;  /**< Must be first field */
  
  	/*
  	 * Generic configuration
  	 */
  	int     enabled_events;
  	int     max_msg_size;
  	int     msg_copy_max_size;
          int     recv_max_msg_size;
  	int     max_inflight;
  	int     metadata_request_timeout_ms;
  	int     metadata_refresh_interval_ms;
  	int     metadata_refresh_fast_cnt;
  	int     metadata_refresh_fast_interval_ms;
          int     metadata_refresh_sparse;
          int     metadata_max_age_ms;
  	int     debug;
  	int     broker_addr_ttl;
          int     broker_addr_family;
  	int     socket_timeout_ms;
  	int     socket_blocking_max_ms;
  	int     socket_sndbuf_size;
  	int     socket_rcvbuf_size;
          int     socket_keepalive;
  	int     socket_nagle_disable;
          int     socket_max_fails;
  	char   *client_id_str;
  	char   *brokerlist;
  	int     stats_interval_ms;
  	int     term_sig;
          int     reconnect_backoff_ms;
          int     reconnect_backoff_max_ms;
          int     reconnect_jitter_ms;
          int     sparse_connections;
          int     sparse_connect_intvl;
  	int     api_version_request;
  	int     api_version_request_timeout_ms;
  	int     api_version_fallback_ms;
  	char   *broker_version_fallback;
  	rd_kafka_secproto_t security_protocol;
  
  #if WITH_SSL
  	struct {
  		SSL_CTX *ctx;
  		char *cipher_suites;
  #if OPENSSL_VERSION_NUMBER >= 0x1000200fL && !defined(LIBRESSL_VERSION_NUMBER)
  		char *curves_list;
  		char *sigalgs_list;
  #endif
  		char *key_location;
  		char *key_password;
  		char *cert_location;
  		char *ca_location;
  		char *crl_location;
  		char *keystore_location;
  		char *keystore_password;
  	} ssl;
  #endif
  
          struct {
                  const struct rd_kafka_sasl_provider *provider;
                  char *principal;
                  char *mechanisms;
                  char *service_name;
                  char *kinit_cmd;
                  char *keytab;
                  int   relogin_min_time;
                  char *username;
                  char *password;
  #if WITH_SASL_SCRAM
                  /* SCRAM EVP-wrapped hash function
                   * (return value from EVP_shaX()) */
                  const void/*EVP_MD*/ *scram_evp;
                  /* SCRAM direct hash function (e.g., SHA256()) */
                  unsigned char *(*scram_H) (const unsigned char *d, size_t n,
                                             unsigned char *md);
                  /* Hash size */
                  size_t         scram_H_size;
  #endif
          } sasl;
  
  #if WITH_PLUGINS
          char *plugin_paths;
          rd_list_t plugins;
  #endif
  
          /* Interceptors */
          struct {
                  /* rd_kafka_interceptor_method_t lists */
                  rd_list_t on_conf_set;        /* on_conf_set interceptors
                                                 * (not copied on conf_dup()) */
                  rd_list_t on_conf_dup;        /* .. (not copied) */
                  rd_list_t on_conf_destroy;    /* .. (not copied) */
                  rd_list_t on_new;             /* .. (copied) */
                  rd_list_t on_destroy;         /* .. (copied) */
                  rd_list_t on_send;            /* .. (copied) */
                  rd_list_t on_acknowledgement; /* .. (copied) */
                  rd_list_t on_consume;         /* .. (copied) */
                  rd_list_t on_commit;          /* .. (copied) */
                  rd_list_t on_request_sent;    /* .. (copied) */
  
                  /* rd_strtup_t list */
                  rd_list_t config;             /* Configuration name=val's
                                                 * handled by interceptors. */
          } interceptors;
  
          /* Client group configuration */
          int    coord_query_intvl_ms;
          int    max_poll_interval_ms;
  
  	int    builtin_features;
  	/*
  	 * Consumer configuration
  	 */
          int    check_crcs;
  	int    queued_min_msgs;
          int    queued_max_msg_kbytes;
          int64_t queued_max_msg_bytes;
  	int    fetch_wait_max_ms;
          int    fetch_msg_max_bytes;
          int    fetch_max_bytes;
  	int    fetch_min_bytes;
  	int    fetch_error_backoff_ms;
          char  *group_id_str;
  
          rd_kafka_pattern_list_t *topic_blacklist;
          struct rd_kafka_topic_conf_s *topic_conf; /* Default topic config
                                                     * for automatically
                                                     * subscribed topics. */
          int enable_auto_commit;
  	int enable_auto_offset_store;
          int auto_commit_interval_ms;
          int group_session_timeout_ms;
          int group_heartbeat_intvl_ms;
          rd_kafkap_str_t *group_protocol_type;
          char *partition_assignment_strategy;
          rd_list_t partition_assignors;
  	int enabled_assignor_cnt;
          struct rd_kafka_assignor_s *assignor;
  
          void (*rebalance_cb) (rd_kafka_t *rk,
                                rd_kafka_resp_err_t err,
  			      rd_kafka_topic_partition_list_t *partitions,
                                void *opaque);
  
          void (*offset_commit_cb) (rd_kafka_t *rk,
                                    rd_kafka_resp_err_t err,
                                    rd_kafka_topic_partition_list_t *offsets,
                                    void *opaque);
  
          rd_kafka_offset_method_t offset_store_method;
  	int enable_partition_eof;
  
  	/*
  	 * Producer configuration
  	 */
          struct {
                  int    idempotence;  /**< Enable Idempotent Producer */
                  rd_bool_t gapless;   /**< Raise fatal error if
                                        *   gapless guarantee can't be
                                        *   satisfied. */
          } eos;
  	int    queue_buffering_max_msgs;
  	int    queue_buffering_max_kbytes;
  	int    buffering_max_ms;
          int    queue_backpressure_thres;
  	int    max_retries;
  	int    retry_backoff_ms;
  	int    batch_num_messages;
  	rd_kafka_compression_t compression_codec;
  	int    dr_err_only;
  
  	/* Message delivery report callback.
  	 * Called once for each produced message, either on
  	 * successful and acknowledged delivery to the broker in which
  	 * case 'err' is 0, or if the message could not be delivered
  	 * in which case 'err' is non-zero (use rd_kafka_err2str()
  	 * to obtain a human-readable error reason).
  	 *
  	 * If the message was produced with neither RD_KAFKA_MSG_F_FREE
  	 * or RD_KAFKA_MSG_F_COPY set then 'payload' is the original
  	 * pointer provided to rd_kafka_produce().
  	 * rdkafka will not perform any further actions on 'payload'
  	 * at this point and the application may rd_free the payload data
  	 * at this point.
  	 *
  	 * 'opaque' is 'conf.opaque', while 'msg_opaque' is
  	 * the opaque pointer provided in the rd_kafka_produce() call.
  	 */
  	void (*dr_cb) (rd_kafka_t *rk,
  		       void *payload, size_t len,
  		       rd_kafka_resp_err_t err,
  		       void *opaque, void *msg_opaque);
  
          void (*dr_msg_cb) (rd_kafka_t *rk, const rd_kafka_message_t *rkmessage,
                             void *opaque);
  
          /* Consume callback */
          void (*consume_cb) (rd_kafka_message_t *rkmessage, void *opaque);
  
          /* Log callback */
          void (*log_cb) (const rd_kafka_t *rk, int level,
                          const char *fac, const char *buf);
          int    log_level;
          int    log_queue;
          int    log_thread_name;
          int    log_connection_close;
  
          /* Error callback */
  	void (*error_cb) (rd_kafka_t *rk, int err,
  			  const char *reason, void *opaque);
  
  	/* Throttle callback */
  	void (*throttle_cb) (rd_kafka_t *rk, const char *broker_name,
  			     int32_t broker_id, int throttle_time_ms,
  			     void *opaque);
  
  	/* Stats callback */
  	int (*stats_cb) (rd_kafka_t *rk,
  			 char *json,
  			 size_t json_len,
  			 void *opaque);
  
          /* Socket creation callback */
          int (*socket_cb) (int domain, int type, int protocol, void *opaque);
  
          /* Connect callback */
          int (*connect_cb) (int sockfd,
                             const struct sockaddr *addr,
                             int addrlen,
                             const char *id,
                             void *opaque);
  
          /* Close socket callback */
          int (*closesocket_cb) (int sockfd, void *opaque);
  
  		/* File open callback */
          int (*open_cb) (const char *pathname, int flags, mode_t mode,
                          void *opaque);
  
          /* Background queue event callback */
          void (*background_event_cb) (rd_kafka_t *rk, rd_kafka_event_t *rkev,
                                       void *opaque);
  
  
  	/* Opaque passed to callbacks. */
  	void  *opaque;
  
          /* For use with value-less properties. */
          int     dummy;
  
  
          /* Admin client defaults */
          struct {
                  int request_timeout_ms;  /* AdminOptions.request_timeout */
          } admin;
  
  
          /*
           * Unit test pluggable interfaces
           */
          struct {
                  /**< Inject errors in ProduceResponse handler */
                  rd_kafka_resp_err_t (*handle_ProduceResponse) (
                          rd_kafka_t *rk,
                          int32_t brokerid,
                          uint64_t msgid,
                          rd_kafka_resp_err_t err);
          } ut;
  };
  
  int rd_kafka_socket_cb_linux (int domain, int type, int protocol, void *opaque);
  int rd_kafka_socket_cb_generic (int domain, int type, int protocol,
                                  void *opaque);
  #ifndef _MSC_VER
  int rd_kafka_open_cb_linux (const char *pathname, int flags, mode_t mode,
                              void *opaque);
  #endif
  int rd_kafka_open_cb_generic (const char *pathname, int flags, mode_t mode,
                                void *opaque);
  
  
  
  struct rd_kafka_topic_conf_s {
          struct rd_kafka_anyconf_hdr hdr;  /**< Must be first field */
  
  	int     required_acks;
  	int32_t request_timeout_ms;
  	int     message_timeout_ms;
  
  	int32_t (*partitioner) (const rd_kafka_topic_t *rkt,
  				const void *keydata, size_t keylen,
  				int32_t partition_cnt,
  				void *rkt_opaque,
  				void *msg_opaque);
          char   *partitioner_str;
  
          int queuing_strategy; /* RD_KAFKA_QUEUE_FIFO|LIFO */
          int (*msg_order_cmp) (const void *a, const void *b);
  
  	rd_kafka_compression_t compression_codec;
  	rd_kafka_complevel_t compression_level;
          int     produce_offset_report;
  
          int     consume_callback_max_msgs;
  	int     auto_commit;
  	int     auto_commit_interval_ms;
  	int     auto_offset_reset;
  	char   *offset_store_path;
  	int     offset_store_sync_interval_ms;
  
          rd_kafka_offset_method_t offset_store_method;
  
  	/* Application provided opaque pointer (this is rkt_opaque) */
  	void   *opaque;
  };
  
  
  
  void rd_kafka_anyconf_destroy (int scope, void *conf);
  
  const char *rd_kafka_conf_finalize (rd_kafka_type_t cltype,
                                      rd_kafka_conf_t *conf);
  const char *rd_kafka_topic_conf_finalize (rd_kafka_type_t cltype,
                                            const rd_kafka_conf_t *conf,
                                            rd_kafka_topic_conf_t *tconf);
  
  
  int rd_kafka_conf_warn (rd_kafka_t *rk);
  
  
  #include "rdkafka_confval.h"
  
  int unittest_conf (void);
  
  #endif /* _RDKAFKA_CONF_H_ */
  