/*
* 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.
*/
#include "rdkafka_int.h"
#include "rdkafka_plugin.h"
#include "rddl.h"
typedef struct rd_kafka_plugin_s {
char *rkplug_path; /* Library path */
rd_kafka_t *rkplug_rk; /* Backpointer to the rk handle */
void *rkplug_handle; /* dlopen (or similar) handle */
void *rkplug_opaque; /* Plugin's opaque */
} rd_kafka_plugin_t;
/**
* @brief Plugin path comparator
*/
static int rd_kafka_plugin_cmp (const void *_a, const void *_b) {
const rd_kafka_plugin_t *a = _a, *b = _b;
return strcmp(a->rkplug_path, b->rkplug_path);
}
/**
* @brief Add plugin (by library path) and calls its conf_init() constructor
*
* @returns an error code on error.
* @remark duplicate plugins are silently ignored.
*
* @remark Libraries are refcounted and thus not unloaded until all
* plugins referencing the library have been destroyed.
* (dlopen() and LoadLibrary() does this for us)
*/
static rd_kafka_resp_err_t
rd_kafka_plugin_new (rd_kafka_conf_t *conf, const char *path,
char *errstr, size_t errstr_size) {
rd_kafka_plugin_t *rkplug;
const rd_kafka_plugin_t skel = { .rkplug_path = (char *)path };
rd_kafka_plugin_f_conf_init_t *conf_init;
rd_kafka_resp_err_t err;
void *handle;
void *plug_opaque = NULL;
/* Avoid duplicates */
if (rd_list_find(&conf->plugins, &skel, rd_kafka_plugin_cmp)) {
rd_snprintf(errstr, errstr_size,
"Ignoring duplicate plugin %s", path);
return RD_KAFKA_RESP_ERR_NO_ERROR;
}
rd_kafka_dbg0(conf, PLUGIN, "PLUGLOAD",
"Loading plugin \"%s\"", path);
/* Attempt to load library */
if (!(handle = rd_dl_open(path, errstr, errstr_size))) {
rd_kafka_dbg0(conf, PLUGIN, "PLUGLOAD",
"Failed to load plugin \"%s\": %s",
path, errstr);
return RD_KAFKA_RESP_ERR__FS;
}
/* Find conf_init() function */
if (!(conf_init = rd_dl_sym(handle, "conf_init",
errstr, errstr_size))) {
rd_dl_close(handle);
return RD_KAFKA_RESP_ERR__INVALID_ARG;
}
/* Call conf_init() */
rd_kafka_dbg0(conf, PLUGIN, "PLUGINIT",
"Calling plugin \"%s\" conf_init()", path);
if ((err = conf_init(conf, &plug_opaque, errstr, errstr_size))) {
rd_dl_close(handle);
return err;
}
rkplug = rd_calloc(1, sizeof(*rkplug));
rkplug->rkplug_path = rd_strdup(path);
rkplug->rkplug_handle = handle;
rkplug->rkplug_opaque = plug_opaque;
rd_list_add(&conf->plugins, rkplug);
rd_kafka_dbg0(conf, PLUGIN, "PLUGLOAD",
"Plugin \"%s\" loaded", path);
return RD_KAFKA_RESP_ERR_NO_ERROR;
}
/**
* @brief Free the plugin, any conf_destroy() interceptors will have been
* called prior to this call.
* @remark plugin is not removed from any list (caller's responsibility)
* @remark this relies on the actual library loader to refcount libraries,
* especially in the config copy case.
* This is true for POSIX dlopen() and Win32 LoadLibrary().
* @locality application thread
*/
static void rd_kafka_plugin_destroy (rd_kafka_plugin_t *rkplug) {
rd_dl_close(rkplug->rkplug_handle);
rd_free(rkplug->rkplug_path);
rd_free(rkplug);
}
/**
* @brief Initialize all configured plugins.
*
* @remark Any previously loaded plugins will be unloaded.
*
* @returns the error code of the first failing plugin.
* @locality application thread calling rd_kafka_new().
*/
static rd_kafka_conf_res_t
rd_kafka_plugins_conf_set0 (rd_kafka_conf_t *conf, const char *paths,
char *errstr, size_t errstr_size) {
char *s;
rd_list_destroy(&conf->plugins);
rd_list_init(&conf->plugins, 0, (void *)&rd_kafka_plugin_destroy);
if (!paths || !*paths)
return RD_KAFKA_CONF_OK;
/* Split paths by ; */
rd_strdupa(&s, paths);
rd_kafka_dbg0(conf, PLUGIN, "PLUGLOAD",
"Loading plugins from conf object %p: \"%s\"",
conf, paths);
while (s && *s) {
char *path = s;
char *t;
rd_kafka_resp_err_t err;
if ((t = strchr(s, ';'))) {
*t = '\0';
s = t+1;
} else {
s = NULL;
}
if ((err = rd_kafka_plugin_new(conf, path,
errstr, errstr_size))) {
/* Failed to load plugin */
size_t elen = errstr_size > 0 ? strlen(errstr) : 0;
/* See if there is room for appending the
* plugin path to the error message. */
if (elen + strlen("(plugin )") + strlen(path) <
errstr_size)
rd_snprintf(errstr+elen, errstr_size-elen,
" (plugin %s)", path);
rd_list_destroy(&conf->plugins);
return RD_KAFKA_CONF_INVALID;
}
}
return RD_KAFKA_CONF_OK;
}
/**
* @brief Conf setter for "plugin.library.paths"
*/
rd_kafka_conf_res_t rd_kafka_plugins_conf_set (
int scope, void *pconf, const char *name, const char *value,
void *dstptr, rd_kafka_conf_set_mode_t set_mode,
char *errstr, size_t errstr_size) {
assert(scope == _RK_GLOBAL);
return rd_kafka_plugins_conf_set0((rd_kafka_conf_t *)pconf,
set_mode == _RK_CONF_PROP_SET_DEL ?
NULL : value, errstr, errstr_size);
}