/*

 * librdkafka - Apache Kafka C library

 *

 * Copyright (c) 2012-2013, 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_INT_H_

#define _RDKAFKA_INT_H_

#ifndef _MSC_VER

#define _GNU_SOURCE  /* for strndup() */

#include <syslog.h>

#else

typedef int mode_t;

#endif

#include <fcntl.h>

#include "rdsysqueue.h"

#include "rdkafka.h"

#include "rd.h"

#include "rdlog.h"

#include "rdtime.h"

#include "rdaddr.h"

#include "rdinterval.h"

#include "rdavg.h"

#include "rdlist.h"

#if WITH_SSL

#include <openssl/ssl.h>

#endif

typedef struct rd_kafka_itopic_s rd_kafka_itopic_t;

typedef struct rd_ikafka_s rd_ikafka_t;

#define rd_kafka_assert(rk, cond) do {                                  \

                if (unlikely(!(cond)))                                  \

                        rd_kafka_crash(__FILE__,__LINE__, __FUNCTION__, \

                                       (rk), "assert: " # cond);        \

        } while (0)

void

RD_NORETURN

rd_kafka_crash (const char *file, int line, const char *function,

                rd_kafka_t *rk, const char *reason);

/* Forward declarations */

struct rd_kafka_s;

struct rd_kafka_itopic_s;

struct rd_kafka_msg_s;

struct rd_kafka_broker_s;

struct rd_kafka_toppar_s;

typedef RD_SHARED_PTR_TYPE(, struct rd_kafka_toppar_s) shptr_rd_kafka_toppar_t;

typedef RD_SHARED_PTR_TYPE(, struct rd_kafka_itopic_s) shptr_rd_kafka_itopic_t;

#include "rdkafka_op.h"

#include "rdkafka_queue.h"

#include "rdkafka_msg.h"

#include "rdkafka_proto.h"

#include "rdkafka_buf.h"

#include "rdkafka_pattern.h"

#include "rdkafka_conf.h"

#include "rdkafka_transport.h"

#include "rdkafka_timer.h"

#include "rdkafka_assignor.h"

#include "rdkafka_metadata.h"

/**

 * Protocol level sanity

 */

#define RD_KAFKAP_BROKERS_MAX     1000

#define RD_KAFKAP_TOPICS_MAX      1000000

#define RD_KAFKAP_PARTITIONS_MAX  10000

#define RD_KAFKA_OFFSET_IS_LOGICAL(OFF)  ((OFF) < 0)

/**

 * Kafka handle, internal representation of the application's rd_kafka_t.

 */

typedef RD_SHARED_PTR_TYPE(shptr_rd_ikafka_s, rd_ikafka_t) shptr_rd_ikafka_t;

struct rd_kafka_s {

	rd_kafka_q_t *rk_rep;   /* kafka -> application reply queue */

	rd_kafka_q_t *rk_ops;   /* any -> rdkafka main thread ops */

	TAILQ_HEAD(, rd_kafka_broker_s) rk_brokers;

        rd_list_t                  rk_broker_by_id; /* Fast id lookups. */

	rd_atomic32_t              rk_broker_cnt;

	rd_atomic32_t              rk_broker_down_cnt;

        mtx_t                      rk_internal_rkb_lock;

	rd_kafka_broker_t         *rk_internal_rkb;

	/* Broadcasting of broker state changes to wake up

	 * functions waiting for a state change. */

	cnd_t                      rk_broker_state_change_cnd;

	mtx_t                      rk_broker_state_change_lock;

	int                        rk_broker_state_change_version;

	TAILQ_HEAD(, rd_kafka_itopic_s)  rk_topics;

	int              rk_topic_cnt;

        struct rd_kafka_cgrp_s *rk_cgrp;

        rd_kafka_conf_t  rk_conf;

        rd_kafka_q_t    *rk_logq;          /* Log queue if `log.queue` set */

        char             rk_name[128];

	rd_kafkap_str_t *rk_client_id;

        rd_kafkap_str_t *rk_group_id;    /* Consumer group id */

	int              rk_flags;

	rd_atomic32_t    rk_terminate;

	rwlock_t         rk_lock;

	rd_kafka_type_t  rk_type;

	struct timeval   rk_tv_state_change;

	rd_atomic32_t    rk_last_throttle;  /* Last throttle_time_ms value

					     * from broker. */

        /* Locks: rd_kafka_*lock() */

        rd_ts_t          rk_ts_metadata;    /* Timestamp of most recent

                                             * metadata. */

	struct rd_kafka_metadata *rk_full_metadata; /* Last full metadata. */

	rd_ts_t          rk_ts_full_metadata;       /* Timesstamp of .. */

        struct rd_kafka_metadata_cache rk_metadata_cache; /* Metadata cache */

        char            *rk_clusterid;      /* ClusterId from metadata */

        /* Simple consumer count:

         *  >0: Running in legacy / Simple Consumer mode,

         *   0: No consumers running

         *  <0: Running in High level consumer mode */

        rd_atomic32_t    rk_simple_cnt;

        /**

         * Exactly Once Semantics

         */

        struct {

                rd_kafkap_str_t *TransactionalId;

                int64_t          PID;

                int16_t          ProducerEpoch;

        } rk_eos;

	const rd_kafkap_bytes_t *rk_null_bytes;

	struct {

		mtx_t lock;       /* Protects acces to this struct */

		cnd_t cnd;        /* For waking up blocking injectors */

		unsigned int cnt; /* Current message count */

		size_t size;      /* Current message size sum */

	        unsigned int max_cnt; /* Max limit */

		size_t max_size; /* Max limit */

	} rk_curr_msgs;

        rd_kafka_timers_t rk_timers;

	thrd_t rk_thread;

        int rk_initialized;

};

#define rd_kafka_wrlock(rk)    rwlock_wrlock(&(rk)->rk_lock)

#define rd_kafka_rdlock(rk)    rwlock_rdlock(&(rk)->rk_lock)

#define rd_kafka_rdunlock(rk)    rwlock_rdunlock(&(rk)->rk_lock)

#define rd_kafka_wrunlock(rk)    rwlock_wrunlock(&(rk)->rk_lock)

/**

 * @brief Add \p cnt messages and of total size \p size bytes to the

 *        internal bookkeeping of current message counts.

 *        If the total message count or size after add would exceed the

 *        configured limits \c queue.buffering.max.messages and

 *        \c queue.buffering.max.kbytes then depending on the value of

 *        \p block the function either blocks until enough space is available

 *        if \p block is 1, else immediately returns

 *        RD_KAFKA_RESP_ERR__QUEUE_FULL.

 *

 * @param rdmtx If non-null and \p block is set and blocking is to ensue,

 *              then unlock this mutex for the duration of the blocking

 *              and then reacquire with a read-lock.

 */

static RD_INLINE RD_UNUSED rd_kafka_resp_err_t

rd_kafka_curr_msgs_add (rd_kafka_t *rk, unsigned int cnt, size_t size,

			int block, rwlock_t *rdlock) {

	if (rk->rk_type != RD_KAFKA_PRODUCER)

		return RD_KAFKA_RESP_ERR_NO_ERROR;

	mtx_lock(&rk->rk_curr_msgs.lock);

	while (unlikely(rk->rk_curr_msgs.cnt + cnt >

			rk->rk_curr_msgs.max_cnt ||

			(unsigned long long)(rk->rk_curr_msgs.size + size) >

			(unsigned long long)rk->rk_curr_msgs.max_size)) {

		if (!block) {

			mtx_unlock(&rk->rk_curr_msgs.lock);

			return RD_KAFKA_RESP_ERR__QUEUE_FULL;

		}

                if (rdlock)

                        rwlock_rdunlock(rdlock);

		cnd_wait(&rk->rk_curr_msgs.cnd, &rk->rk_curr_msgs.lock);

                if (rdlock)

                        rwlock_rdlock(rdlock);

	}

	rk->rk_curr_msgs.cnt  += cnt;

	rk->rk_curr_msgs.size += size;

	mtx_unlock(&rk->rk_curr_msgs.lock);

	return RD_KAFKA_RESP_ERR_NO_ERROR;

}

/**

 * @brief Subtract \p cnt messages of total size \p size from the

 *        current bookkeeping and broadcast a wakeup on the condvar

 *        for any waiting & blocking threads.

 */

static RD_INLINE RD_UNUSED void

rd_kafka_curr_msgs_sub (rd_kafka_t *rk, unsigned int cnt, size_t size) {

        int broadcast = 0;

	if (rk->rk_type != RD_KAFKA_PRODUCER)

		return;

	mtx_lock(&rk->rk_curr_msgs.lock);

	rd_kafka_assert(NULL,

			rk->rk_curr_msgs.cnt >= cnt &&

			rk->rk_curr_msgs.size >= size);

        /* If the subtraction would pass one of the thresholds

         * broadcast a wake-up to any waiting listeners. */

        if ((rk->rk_curr_msgs.cnt >= rk->rk_curr_msgs.max_cnt &&

             rk->rk_curr_msgs.cnt - cnt < rk->rk_curr_msgs.max_cnt) ||

            (rk->rk_curr_msgs.size >= rk->rk_curr_msgs.max_size &&

             rk->rk_curr_msgs.size - size < rk->rk_curr_msgs.max_size))

                broadcast = 1;

	rk->rk_curr_msgs.cnt  -= cnt;

	rk->rk_curr_msgs.size -= size;

        if (unlikely(broadcast))

                cnd_broadcast(&rk->rk_curr_msgs.cnd);

	mtx_unlock(&rk->rk_curr_msgs.lock);

}

static RD_INLINE RD_UNUSED void

rd_kafka_curr_msgs_get (rd_kafka_t *rk, unsigned int *cntp, size_t *sizep) {

	if (rk->rk_type != RD_KAFKA_PRODUCER) {

		*cntp = 0;

		*sizep = 0;

		return;

	}

	mtx_lock(&rk->rk_curr_msgs.lock);

	*cntp = rk->rk_curr_msgs.cnt;

	*sizep = rk->rk_curr_msgs.size;

	mtx_unlock(&rk->rk_curr_msgs.lock);

}

static RD_INLINE RD_UNUSED int

rd_kafka_curr_msgs_cnt (rd_kafka_t *rk) {

	int cnt;

	if (rk->rk_type != RD_KAFKA_PRODUCER)

		return 0;

	mtx_lock(&rk->rk_curr_msgs.lock);

	cnt = rk->rk_curr_msgs.cnt;

	mtx_unlock(&rk->rk_curr_msgs.lock);

	return cnt;

}

void rd_kafka_destroy_final (rd_kafka_t *rk);

/**

 * Returns true if 'rk' handle is terminating.

 */

#define rd_kafka_terminating(rk) (rd_atomic32_get(&(rk)->rk_terminate))

#define rd_kafka_is_simple_consumer(rk) \

        (rd_atomic32_get(&(rk)->rk_simple_cnt) > 0)

int rd_kafka_simple_consumer_add (rd_kafka_t *rk);

#include "rdkafka_topic.h"

#include "rdkafka_partition.h"

/**

 * Debug contexts

 */

#define RD_KAFKA_DBG_GENERIC        0x1

#define RD_KAFKA_DBG_BROKER         0x2

#define RD_KAFKA_DBG_TOPIC          0x4

#define RD_KAFKA_DBG_METADATA       0x8

#define RD_KAFKA_DBG_FEATURE        0x10

#define RD_KAFKA_DBG_QUEUE          0x20

#define RD_KAFKA_DBG_MSG            0x40

#define RD_KAFKA_DBG_PROTOCOL       0x80

#define RD_KAFKA_DBG_CGRP           0x100

#define RD_KAFKA_DBG_SECURITY       0x200

#define RD_KAFKA_DBG_FETCH          0x400

#define RD_KAFKA_DBG_INTERCEPTOR    0x800

#define RD_KAFKA_DBG_PLUGIN         0x1000

#define RD_KAFKA_DBG_CONSUMER       0x2000

#define RD_KAFKA_DBG_ALL            0xffff

#define RD_KAFKA_DBG_NONE           0x0

void rd_kafka_log0(const rd_kafka_conf_t *conf,

                   const rd_kafka_t *rk, const char *extra, int level,

                   const char *fac, const char *fmt, ...) RD_FORMAT(printf,

                                                                    6, 7);

#define rd_kafka_log(rk,level,fac,...) \

        rd_kafka_log0(&rk->rk_conf, rk, NULL, level, fac, __VA_ARGS__)

#define rd_kafka_dbg(rk,ctx,fac,...) do {                               \

                if (unlikely((rk)->rk_conf.debug & (RD_KAFKA_DBG_ ## ctx))) \

                        rd_kafka_log0(&rk->rk_conf,rk,NULL,             \

                                      LOG_DEBUG,fac,__VA_ARGS__);       \

        } while (0)

/* dbg() not requiring an rk, just the conf object, for early logging */

#define rd_kafka_dbg0(conf,ctx,fac,...) do {                            \

                if (unlikely((conf)->debug & (RD_KAFKA_DBG_ ## ctx)))   \

                        rd_kafka_log0(conf,NULL,NULL,                   \

                                      LOG_DEBUG,fac,__VA_ARGS__);       \

        } while (0)

/* NOTE: The local copy of _logname is needed due rkb_logname_lock lock-ordering

 *       when logging another broker's name in the message. */

#define rd_rkb_log(rkb,level,fac,...) do {				\

		char _logname[RD_KAFKA_NODENAME_SIZE];			\

                mtx_lock(&(rkb)->rkb_logname_lock);                     \

		strncpy(_logname, rkb->rkb_logname, sizeof(_logname)-1); \

		_logname[RD_KAFKA_NODENAME_SIZE-1] = '\0';		\

                mtx_unlock(&(rkb)->rkb_logname_lock);                   \

		rd_kafka_log0(&(rkb)->rkb_rk->rk_conf, \

                              (rkb)->rkb_rk, _logname,                  \

                              level, fac, __VA_ARGS__);                 \

        } while (0)

#define rd_rkb_dbg(rkb,ctx,fac,...) do {				\

		if (unlikely((rkb)->rkb_rk->rk_conf.debug &		\

			     (RD_KAFKA_DBG_ ## ctx))) {			\

			rd_rkb_log(rkb, LOG_DEBUG, fac, __VA_ARGS__);	\

                }                                                       \

	} while (0)

extern rd_kafka_resp_err_t RD_TLS rd_kafka_last_error_code;

static RD_UNUSED RD_INLINE

rd_kafka_resp_err_t rd_kafka_set_last_error (rd_kafka_resp_err_t err,

					     int errnox) {

        if (errnox) {

#ifdef _MSC_VER

                /* This is the correct way to set errno on Windows,

                 * but it is still pointless due to different errnos in

                 * in different runtimes:

                 * https://social.msdn.microsoft.com/Forums/vstudio/en-US/b4500c0d-1b69-40c7-9ef5-08da1025b5bf/setting-errno-from-within-a-dll?forum=vclanguage/

                 * errno is thus highly deprecated, and buggy, on Windows

                 * when using librdkafka as a dynamically loaded DLL. */

                _set_errno(errnox);

#else

                errno = errnox;

#endif

        }

	rd_kafka_last_error_code = err;

	return err;

}

extern rd_atomic32_t rd_kafka_thread_cnt_curr;

void rd_kafka_set_thread_name (const char *fmt, ...);

void rd_kafka_set_thread_sysname (const char *fmt, ...);

int rd_kafka_path_is_dir (const char *path);

rd_kafka_op_res_t

rd_kafka_poll_cb (rd_kafka_t *rk, rd_kafka_q_t *rkq, rd_kafka_op_t *rko,

                  rd_kafka_q_cb_type_t cb_type, void *opaque);

rd_kafka_resp_err_t rd_kafka_subscribe_rkt (rd_kafka_itopic_t *rkt);

#endif /* _RDKAFKA_INT_H_ */