/*
 * librdkafka - The Apache Kafka C/C++ library
 *
 * Copyright (c) 2017 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.
 */


/**
 * @brief Interceptor plugin test library
 *
 * Interceptors can be implemented in the app itself and use
 * the direct API to set the interceptors methods, or be implemented
 * as an external plugin library that uses the direct APIs.
 *
 * This file implements the latter, an interceptor plugin library.
 */

#define _CRT_SECURE_NO_WARNINGS /* Silence MSVC nonsense */

#include "../test.h"

#include <stdio.h>
#include <string.h>
#include <assert.h>

/* typical include path outside tests is <librdkafka/rdkafka.h> */
#include "rdkafka.h"

#include "interceptor_test.h"

#ifdef _MSC_VER
#define DLL_EXPORT __declspec(dllexport)
#else
#define DLL_EXPORT
#endif

/**
 * @brief Interceptor instance.
 *
 * An interceptor instance is created for each intercepted configuration
 * object (triggered through conf_init() which is the plugin loader,
 * or by conf_dup() which is a copying of a conf previously seen by conf_init())
 */
struct ici {
        rd_kafka_conf_t *conf;  /**< Interceptor config */
        char *config1;          /**< Interceptor-specific config */
        char *config2;

        int on_new_cnt;
        int on_conf_destroy_cnt;
};

static char *my_interceptor_plug_opaque = "my_interceptor_plug_opaque";



/* Producer methods */
rd_kafka_resp_err_t on_send (rd_kafka_t *rk,
                             rd_kafka_message_t *rkmessage,
                             void *ic_opaque) {
        struct ici *ici = ic_opaque;
        printf("on_send: %p\n", ici);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


rd_kafka_resp_err_t on_acknowledgement (rd_kafka_t *rk,
                                        rd_kafka_message_t *rkmessage,
                                        void *ic_opaque) {
        struct ici *ici = ic_opaque;
        printf("on_acknowledgement: %p: err %d, partition %"PRId32"\n",
               ici, rkmessage->err, rkmessage->partition);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}

/* Consumer methods */
rd_kafka_resp_err_t on_consume (rd_kafka_t *rk,
                                rd_kafka_message_t *rkmessage,
                                void *ic_opaque) {
        struct ici *ici = ic_opaque;
        printf("on_consume: %p: partition %"PRId32" @ %"PRId64"\n",
               ici, rkmessage->partition, rkmessage->offset);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}

rd_kafka_resp_err_t on_commit (rd_kafka_t *rk,
                               const rd_kafka_topic_partition_list_t *offsets,
                               rd_kafka_resp_err_t err, void *ic_opaque) {
        struct ici *ici = ic_opaque;
        printf("on_commit: %p: err %d\n", ici, err);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


static void ici_destroy (struct ici *ici) {
        if (ici->conf)
                rd_kafka_conf_destroy(ici->conf);
        if (ici->config1)
                free(ici->config1);
        if (ici->config2)
                free(ici->config2);
        free(ici);
}

rd_kafka_resp_err_t on_destroy (rd_kafka_t *rk, void *ic_opaque) {
        struct ici *ici = ic_opaque;
        printf("on_destroy: %p\n", ici);
        /* the ici is freed from on_conf_destroy() */
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * @brief Called from rd_kafka_new(). We use it to set up interceptors.
 */
static rd_kafka_resp_err_t on_new (rd_kafka_t *rk, const rd_kafka_conf_t *conf,
                                   void *ic_opaque,
                                   char *errstr, size_t errstr_size) {
        struct ici *ici = ic_opaque;

        ictest.on_new.cnt++;
        ici->on_new_cnt++;

        TEST_SAY("on_new(rk %p, conf %p, ici->conf %p): %p: #%d\n",
                 rk, conf, ici->conf, ici, ictest.on_new.cnt);

        ICTEST_CNT_CHECK(on_new);
        TEST_ASSERT(ici->on_new_cnt == 1);

        TEST_ASSERT(!ictest.session_timeout_ms);
        TEST_ASSERT(!ictest.socket_timeout_ms);
        /* Extract some well known config properties from the interceptor's
         * configuration. */
        ictest.session_timeout_ms = rd_strdup(test_conf_get(ici->conf, "session.timeout.ms"));
        ictest.socket_timeout_ms  = rd_strdup(test_conf_get(ici->conf, "socket.timeout.ms"));
        ictest.config1 = rd_strdup(ici->config1);
        ictest.config2 = rd_strdup(ici->config2);

        rd_kafka_interceptor_add_on_send(rk, __FILE__, on_send, ici);
        rd_kafka_interceptor_add_on_acknowledgement(rk, __FILE__,
                                                    on_acknowledgement, ici);
        rd_kafka_interceptor_add_on_consume(rk, __FILE__, on_consume, ici);
        rd_kafka_interceptor_add_on_commit(rk, __FILE__, on_commit, ici);
        rd_kafka_interceptor_add_on_destroy(rk, __FILE__, on_destroy, ici);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * @brief Configuration set handler
 */
static rd_kafka_conf_res_t on_conf_set (rd_kafka_conf_t *conf,
                                        const char *name, const char *val,
                                        char *errstr, size_t errstr_size,
                                        void *ic_opaque) {
        struct ici *ici = ic_opaque;
        int level = 3;

        if (!strcmp(name, "session.timeout.ms") ||
            !strcmp(name, "socket.timeout.ms") ||
            !strncmp(name, "interceptor_test", strlen("interceptor_test")))
                level = 2;

        TEST_SAYL(level, "on_conf_set(conf %p, \"%s\", \"%s\"): %p\n",
                  conf, name, val, ici);

        if (!strcmp(name, "interceptor_test.good"))
                return RD_KAFKA_CONF_OK;
        else if (!strcmp(name, "interceptor_test.bad")) {
                strncpy(errstr, "on_conf_set failed deliberately",
                        errstr_size-1);
                errstr[errstr_size-1] = '\0';
                return RD_KAFKA_CONF_INVALID;
        } else if (!strcmp(name, "interceptor_test.config1")) {
                if (ici->config1) {
                        free(ici->config1);
                        ici->config1 = NULL;
                }
                if (val)
                        ici->config1 = rd_strdup(val);
                TEST_SAY("on_conf_set(conf %p, %s, %s): %p\n",
                         conf, name, val, ici);
                return RD_KAFKA_CONF_OK;
        } else if (!strcmp(name, "interceptor_test.config2")) {
                if (ici->config2) {
                        free(ici->config2);
                        ici->config2 = NULL;
                }
                if (val)
                        ici->config2 = rd_strdup(val);
                return RD_KAFKA_CONF_OK;
        } else {
                /* Apply intercepted client's config properties on
                 * interceptor config. */
                rd_kafka_conf_set(ici->conf, name, val,
                                  errstr, errstr_size);
                /* UNKNOWN makes the conf_set() call continue with
                 * other interceptors and finally the librdkafka properties. */
                return RD_KAFKA_CONF_UNKNOWN;
        }

        return RD_KAFKA_CONF_UNKNOWN;
}

static void conf_init0 (rd_kafka_conf_t *conf);


/**
 * @brief Set up new configuration on copy.
 */
static rd_kafka_resp_err_t on_conf_dup (rd_kafka_conf_t *new_conf,
                                        const rd_kafka_conf_t *old_conf,
                                        size_t filter_cnt, const char **filter,
                                        void *ic_opaque) {
        struct ici *ici = ic_opaque;
        TEST_SAY("on_conf_dup(new_conf %p, old_conf %p, filter_cnt %"PRIusz
                 ", ici %p)\n",
                 new_conf, old_conf, filter_cnt, ici);
        conf_init0(new_conf);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


static rd_kafka_resp_err_t on_conf_destroy (void *ic_opaque) {
        struct ici *ici = ic_opaque;
        ici->on_conf_destroy_cnt++;
        printf("conf_destroy called (opaque %p vs %p) ici %p\n",
               ic_opaque, my_interceptor_plug_opaque, ici);
        TEST_ASSERT(ici->on_conf_destroy_cnt == 1);
        ici_destroy(ici);
        return RD_KAFKA_RESP_ERR_NO_ERROR;
}



/**
 * @brief Configuration init is intercepted both from plugin.library.paths
 *        as well as rd_kafka_conf_dup().
 *        This internal method serves both cases.
 */
static void conf_init0 (rd_kafka_conf_t *conf) {
        struct ici *ici;
        const char *filter[] = { "plugin.library.paths",
                                 "interceptor_test." };
        size_t filter_cnt = sizeof(filter) / sizeof(*filter);

        /* Create new interceptor instance */
        ici = calloc(1, sizeof(*ici));

        ictest.conf_init.cnt++;
        ICTEST_CNT_CHECK(conf_init);

        /* Create own copy of configuration, after filtering out what
         * brought us here (plugins and our own interceptor config). */
        ici->conf = rd_kafka_conf_dup_filter(conf, filter_cnt, filter);
        TEST_SAY("conf_init0(conf %p) for ici %p with ici->conf %p\n",
                 conf, ici, ici->conf);


        /* Add interceptor methods */
        rd_kafka_conf_interceptor_add_on_new(conf, __FILE__, on_new, ici);

        rd_kafka_conf_interceptor_add_on_conf_set(conf, __FILE__, on_conf_set,
                                                  ici);
        rd_kafka_conf_interceptor_add_on_conf_dup(conf, __FILE__, on_conf_dup,
                                                  ici);
        rd_kafka_conf_interceptor_add_on_conf_destroy(conf, __FILE__,
                                                      on_conf_destroy, ici);
}

/**
 * @brief Plugin conf initializer called when plugin.library.paths is set.
 */
DLL_EXPORT
rd_kafka_resp_err_t conf_init (rd_kafka_conf_t *conf,
                               void **plug_opaquep,
                               char *errstr, size_t errstr_size) {
        *plug_opaquep = (void *)my_interceptor_plug_opaque;

        TEST_SAY("conf_init(conf %p) called (setting opaque to %p)\n",
                 conf, *plug_opaquep);

        conf_init0(conf);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


