You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bq...@apache.org on 2017/11/16 21:15:36 UTC
[06/42] nifi-minifi-cpp git commit: MINIFICPP-274: PutKafka Processor
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdkafka_topic.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_topic.h b/thirdparty/librdkafka-0.11.1/src/rdkafka_topic.h
new file mode 100644
index 0000000..49d1d29
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdkafka_topic.h
@@ -0,0 +1,185 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include "rdlist.h"
+
+extern const char *rd_kafka_topic_state_names[];
+
+
+/* rd_kafka_itopic_t: internal representation of a topic */
+struct rd_kafka_itopic_s {
+ TAILQ_ENTRY(rd_kafka_itopic_s) rkt_link;
+
+ rd_refcnt_t rkt_refcnt;
+
+ rwlock_t rkt_lock;
+ rd_kafkap_str_t *rkt_topic;
+
+ shptr_rd_kafka_toppar_t *rkt_ua; /* unassigned partition */
+ shptr_rd_kafka_toppar_t **rkt_p;
+ int32_t rkt_partition_cnt;
+
+ rd_list_t rkt_desp; /* Desired partitions
+ * that are not yet seen
+ * in the cluster. */
+
+ rd_ts_t rkt_ts_metadata; /* Timestamp of last metadata
+ * update for this topic. */
+
+ mtx_t rkt_app_lock; /* Protects rkt_app_* */
+ rd_kafka_topic_t *rkt_app_rkt; /* A shared topic pointer
+ * to be used for callbacks
+ * to the application. */
+
+ int rkt_app_refcnt; /* Number of active rkt's new()ed
+ * by application. */
+
+ enum {
+ RD_KAFKA_TOPIC_S_UNKNOWN, /* No cluster information yet */
+ RD_KAFKA_TOPIC_S_EXISTS, /* Topic exists in cluster */
+ RD_KAFKA_TOPIC_S_NOTEXISTS, /* Topic is not known in cluster */
+ } rkt_state;
+
+ int rkt_flags;
+#define RD_KAFKA_TOPIC_F_LEADER_UNAVAIL 0x1 /* Leader lost/unavailable
+ * for at least one partition. */
+
+ rd_kafka_t *rkt_rk;
+
+ shptr_rd_kafka_itopic_t *rkt_shptr_app; /* Application's topic_new() */
+
+ rd_kafka_topic_conf_t rkt_conf;
+};
+
+#define rd_kafka_topic_rdlock(rkt) rwlock_rdlock(&(rkt)->rkt_lock)
+#define rd_kafka_topic_wrlock(rkt) rwlock_wrlock(&(rkt)->rkt_lock)
+#define rd_kafka_topic_rdunlock(rkt) rwlock_rdunlock(&(rkt)->rkt_lock)
+#define rd_kafka_topic_wrunlock(rkt) rwlock_wrunlock(&(rkt)->rkt_lock)
+
+
+/* Converts a shptr..itopic_t to an internal itopic_t */
+#define rd_kafka_topic_s2i(s_rkt) rd_shared_ptr_obj(s_rkt)
+
+/* Converts an application topic_t (a shptr topic) to an internal itopic_t */
+#define rd_kafka_topic_a2i(app_rkt) \
+ rd_kafka_topic_s2i((shptr_rd_kafka_itopic_t *)app_rkt)
+
+/* Converts a shptr..itopic_t to an app topic_t (they are the same thing) */
+#define rd_kafka_topic_s2a(s_rkt) ((rd_kafka_topic_t *)(s_rkt))
+
+/* Converts an app topic_t to a shptr..itopic_t (they are the same thing) */
+#define rd_kafka_topic_a2s(app_rkt) ((shptr_rd_kafka_itopic_t *)(app_rkt))
+
+
+
+
+
+/**
+ * Returns a shared pointer for the topic.
+ */
+#define rd_kafka_topic_keep(rkt) \
+ rd_shared_ptr_get(rkt, &(rkt)->rkt_refcnt, shptr_rd_kafka_itopic_t)
+
+/* Same, but casts to an app topic_t */
+#define rd_kafka_topic_keep_a(rkt) \
+ ((rd_kafka_topic_t *)rd_shared_ptr_get(rkt, &(rkt)->rkt_refcnt, \
+ shptr_rd_kafka_itopic_t))
+
+void rd_kafka_topic_destroy_final (rd_kafka_itopic_t *rkt);
+
+
+/**
+ * Frees a shared pointer previously returned by ..topic_keep()
+ */
+static RD_INLINE RD_UNUSED void
+rd_kafka_topic_destroy0 (shptr_rd_kafka_itopic_t *s_rkt) {
+ rd_shared_ptr_put(s_rkt,
+ &rd_kafka_topic_s2i(s_rkt)->rkt_refcnt,
+ rd_kafka_topic_destroy_final(
+ rd_kafka_topic_s2i(s_rkt)));
+}
+
+
+shptr_rd_kafka_itopic_t *rd_kafka_topic_new0 (rd_kafka_t *rk, const char *topic,
+ rd_kafka_topic_conf_t *conf,
+ int *existing, int do_lock);
+
+shptr_rd_kafka_itopic_t *rd_kafka_topic_find_fl (const char *func, int line,
+ rd_kafka_t *rk,
+ const char *topic,
+ int do_lock);
+shptr_rd_kafka_itopic_t *rd_kafka_topic_find0_fl (const char *func, int line,
+ rd_kafka_t *rk,
+ const rd_kafkap_str_t *topic);
+#define rd_kafka_topic_find(rk,topic,do_lock) \
+ rd_kafka_topic_find_fl(__FUNCTION__,__LINE__,rk,topic,do_lock)
+#define rd_kafka_topic_find0(rk,topic) \
+ rd_kafka_topic_find0_fl(__FUNCTION__,__LINE__,rk,topic)
+int rd_kafka_topic_cmp_s_rkt (const void *_a, const void *_b);
+
+void rd_kafka_topic_partitions_remove (rd_kafka_itopic_t *rkt);
+
+void rd_kafka_topic_metadata_none (rd_kafka_itopic_t *rkt);
+
+int rd_kafka_topic_metadata_update2 (rd_kafka_broker_t *rkb,
+ const struct rd_kafka_metadata_topic *mdt);
+
+int rd_kafka_topic_scan_all (rd_kafka_t *rk, rd_ts_t now);
+
+
+typedef struct rd_kafka_topic_info_s {
+ const char *topic; /**< Allocated along with struct */
+ int partition_cnt;
+} rd_kafka_topic_info_t;
+
+
+int rd_kafka_topic_info_cmp (const void *_a, const void *_b);
+rd_kafka_topic_info_t *rd_kafka_topic_info_new (const char *topic,
+ int partition_cnt);
+void rd_kafka_topic_info_destroy (rd_kafka_topic_info_t *ti);
+
+int rd_kafka_topic_match (rd_kafka_t *rk, const char *pattern,
+ const char *topic);
+
+int rd_kafka_toppar_leader_update (rd_kafka_toppar_t *rktp,
+ int32_t leader_id, rd_kafka_broker_t *rkb);
+
+rd_kafka_resp_err_t
+rd_kafka_topics_leader_query_sync (rd_kafka_t *rk, int all_topics,
+ const rd_list_t *topics, int timeout_ms);
+void rd_kafka_topic_leader_query0 (rd_kafka_t *rk, rd_kafka_itopic_t *rkt,
+ int do_rk_lock);
+#define rd_kafka_topic_leader_query(rk,rkt) \
+ rd_kafka_topic_leader_query0(rk,rkt,1/*lock*/)
+
+#define rd_kafka_topic_fast_leader_query(rk) \
+ rd_kafka_metadata_fast_leader_query(rk)
+
+void rd_kafka_local_topics_to_list (rd_kafka_t *rk, rd_list_t *topics);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.c b/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.c
new file mode 100644
index 0000000..3a5f93c
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.c
@@ -0,0 +1,1523 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2015, 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.
+ */
+#ifdef _MSC_VER
+#pragma comment(lib, "ws2_32.lib")
+#endif
+
+#define __need_IOV_MAX
+
+#define _DARWIN_C_SOURCE /* MSG_DONTWAIT */
+
+#include "rdkafka_int.h"
+#include "rdaddr.h"
+#include "rdkafka_transport.h"
+#include "rdkafka_transport_int.h"
+#include "rdkafka_broker.h"
+
+#include <errno.h>
+
+#if WITH_VALGRIND
+/* OpenSSL relies on uninitialized memory, which Valgrind will whine about.
+ * We use in-code Valgrind macros to suppress those warnings. */
+#include <valgrind/memcheck.h>
+#else
+#define VALGRIND_MAKE_MEM_DEFINED(A,B)
+#endif
+
+
+#ifdef _MSC_VER
+#define socket_errno WSAGetLastError()
+#else
+#include <sys/socket.h>
+#define socket_errno errno
+#define SOCKET_ERROR -1
+#endif
+
+/* AIX doesn't have MSG_DONTWAIT */
+#ifndef MSG_DONTWAIT
+# define MSG_DONTWAIT MSG_NONBLOCK
+#endif
+
+
+#if WITH_SSL
+static mtx_t *rd_kafka_ssl_locks;
+static int rd_kafka_ssl_locks_cnt;
+#endif
+
+
+
+/**
+ * Low-level socket close
+ */
+static void rd_kafka_transport_close0 (rd_kafka_t *rk, int s) {
+ if (rk->rk_conf.closesocket_cb)
+ rk->rk_conf.closesocket_cb(s, rk->rk_conf.opaque);
+ else {
+#ifndef _MSC_VER
+ close(s);
+#else
+ closesocket(s);
+#endif
+ }
+
+}
+
+/**
+ * Close and destroy a transport handle
+ */
+void rd_kafka_transport_close (rd_kafka_transport_t *rktrans) {
+#if WITH_SSL
+ if (rktrans->rktrans_ssl) {
+ SSL_shutdown(rktrans->rktrans_ssl);
+ SSL_free(rktrans->rktrans_ssl);
+ }
+#endif
+
+ rd_kafka_sasl_close(rktrans);
+
+ if (rktrans->rktrans_recv_buf)
+ rd_kafka_buf_destroy(rktrans->rktrans_recv_buf);
+
+ if (rktrans->rktrans_s != -1)
+ rd_kafka_transport_close0(rktrans->rktrans_rkb->rkb_rk,
+ rktrans->rktrans_s);
+
+ rd_free(rktrans);
+}
+
+
+static const char *socket_strerror(int err) {
+#ifdef _MSC_VER
+ static RD_TLS char buf[256];
+ rd_strerror_w32(err, buf, sizeof(buf));
+ return buf;
+#else
+ return rd_strerror(err);
+#endif
+}
+
+
+
+
+#ifndef _MSC_VER
+/**
+ * @brief sendmsg() abstraction, converting a list of segments to iovecs.
+ * @remark should only be called if the number of segments is > 1.
+ */
+ssize_t rd_kafka_transport_socket_sendmsg (rd_kafka_transport_t *rktrans,
+ rd_slice_t *slice,
+ char *errstr, size_t errstr_size) {
+ struct iovec iov[IOV_MAX];
+ struct msghdr msg = { .msg_iov = iov };
+ size_t iovlen;
+ ssize_t r;
+
+ rd_slice_get_iov(slice, msg.msg_iov, &iovlen, IOV_MAX,
+ /* FIXME: Measure the effects of this */
+ rktrans->rktrans_sndbuf_size);
+ msg.msg_iovlen = (typeof(msg.msg_iovlen))iovlen;
+
+#ifdef sun
+ /* See recvmsg() comment. Setting it here to be safe. */
+ socket_errno = EAGAIN;
+#endif
+
+ r = sendmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT
+#ifdef MSG_NOSIGNAL
+ | MSG_NOSIGNAL
+#endif
+ );
+
+ if (r == -1) {
+ if (socket_errno == EAGAIN)
+ return 0;
+ rd_snprintf(errstr, errstr_size, "%s", rd_strerror(errno));
+ }
+
+ /* Update buffer read position */
+ rd_slice_read(slice, NULL, (size_t)r);
+
+ return r;
+}
+#endif
+
+
+/**
+ * @brief Plain send() abstraction
+ */
+static ssize_t
+rd_kafka_transport_socket_send0 (rd_kafka_transport_t *rktrans,
+ rd_slice_t *slice,
+ char *errstr, size_t errstr_size) {
+ ssize_t sum = 0;
+ const void *p;
+ size_t rlen;
+
+ while ((rlen = rd_slice_peeker(slice, &p))) {
+ ssize_t r;
+
+ r = send(rktrans->rktrans_s, p,
+#ifdef _MSC_VER
+ (int)rlen, (int)0
+#else
+ rlen, 0
+#endif
+ );
+
+#ifdef _MSC_VER
+ if (unlikely(r == SOCKET_ERROR)) {
+ if (sum > 0 || WSAGetLastError() == WSAEWOULDBLOCK)
+ return sum;
+ else {
+ rd_snprintf(errstr, errstr_size, "%s",
+ socket_strerror(WSAGetLastError()));
+ return -1;
+ }
+ }
+#else
+ if (unlikely(r <= 0)) {
+ if (r == 0 || errno == EAGAIN)
+ return 0;
+ rd_snprintf(errstr, errstr_size, "%s",
+ socket_strerror(socket_errno));
+ return -1;
+ }
+#endif
+
+ /* Update buffer read position */
+ rd_slice_read(slice, NULL, (size_t)r);
+
+ sum += r;
+
+ /* FIXME: remove this and try again immediately and let
+ * the next write() call fail instead? */
+ if ((size_t)r < rlen)
+ break;
+ }
+
+ return sum;
+}
+
+
+static ssize_t
+rd_kafka_transport_socket_send (rd_kafka_transport_t *rktrans,
+ rd_slice_t *slice,
+ char *errstr, size_t errstr_size) {
+#ifndef _MSC_VER
+ /* FIXME: Use sendmsg() with iovecs if there's more than one segment
+ * remaining, otherwise (or if platform does not have sendmsg)
+ * use plain send(). */
+ return rd_kafka_transport_socket_sendmsg(rktrans, slice,
+ errstr, errstr_size);
+#endif
+ return rd_kafka_transport_socket_send0(rktrans, slice,
+ errstr, errstr_size);
+}
+
+
+
+#ifndef _MSC_VER
+/**
+ * @brief recvmsg() abstraction, converting a list of segments to iovecs.
+ * @remark should only be called if the number of segments is > 1.
+ */
+static ssize_t
+rd_kafka_transport_socket_recvmsg (rd_kafka_transport_t *rktrans,
+ rd_buf_t *rbuf,
+ char *errstr, size_t errstr_size) {
+ ssize_t r;
+ struct iovec iov[IOV_MAX];
+ struct msghdr msg = { .msg_iov = iov };
+ size_t iovlen;
+
+ rd_buf_get_write_iov(rbuf, msg.msg_iov, &iovlen, IOV_MAX,
+ /* FIXME: Measure the effects of this */
+ rktrans->rktrans_rcvbuf_size);
+ msg.msg_iovlen = (typeof(msg.msg_iovlen))iovlen;
+
+#ifdef sun
+ /* SunOS doesn't seem to set errno when recvmsg() fails
+ * due to no data and MSG_DONTWAIT is set. */
+ socket_errno = EAGAIN;
+#endif
+ r = recvmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT);
+ if (unlikely(r <= 0)) {
+ if (r == -1 && socket_errno == EAGAIN)
+ return 0;
+ else if (r == 0) {
+ /* Receive 0 after POLLIN event means
+ * connection closed. */
+ rd_snprintf(errstr, errstr_size, "Disconnected");
+ return -1;
+ } else if (r == -1) {
+ rd_snprintf(errstr, errstr_size, "%s",
+ rd_strerror(errno));
+ return -1;
+ }
+ }
+
+ /* Update buffer write position */
+ rd_buf_write(rbuf, NULL, (size_t)r);
+
+ return r;
+}
+#endif
+
+
+/**
+ * @brief Plain recv()
+ */
+static ssize_t
+rd_kafka_transport_socket_recv0 (rd_kafka_transport_t *rktrans,
+ rd_buf_t *rbuf,
+ char *errstr, size_t errstr_size) {
+ ssize_t sum = 0;
+ void *p;
+ size_t len;
+
+ while ((len = rd_buf_get_writable(rbuf, &p))) {
+ ssize_t r;
+
+ r = recv(rktrans->rktrans_s, p,
+#ifdef _MSC_VER
+ (int)
+#endif
+ len,
+ 0);
+
+#ifdef _MSC_VER
+ if (unlikely(r == SOCKET_ERROR)) {
+ if (WSAGetLastError() == WSAEWOULDBLOCK)
+ return sum;
+ rd_snprintf(errstr, errstr_size, "%s",
+ socket_strerror(WSAGetLastError()));
+ return -1;
+ }
+#else
+ if (unlikely(r <= 0)) {
+ if (r == -1 && socket_errno == EAGAIN)
+ return 0;
+ else if (r == 0) {
+ /* Receive 0 after POLLIN event means
+ * connection closed. */
+ rd_snprintf(errstr, errstr_size,
+ "Disconnected");
+ return -1;
+ } else if (r == -1) {
+ rd_snprintf(errstr, errstr_size, "%s",
+ rd_strerror(errno));
+ return -1;
+ }
+ }
+#endif
+
+ /* Update buffer write position */
+ rd_buf_write(rbuf, NULL, (size_t)r);
+
+ sum += r;
+
+ /* FIXME: remove this and try again immediately and let
+ * the next recv() call fail instead? */
+ if ((size_t)r < len)
+ break;
+ }
+ return sum;
+}
+
+
+static ssize_t
+rd_kafka_transport_socket_recv (rd_kafka_transport_t *rktrans,
+ rd_buf_t *buf,
+ char *errstr, size_t errstr_size) {
+#ifndef _MSC_VER
+ /* FIXME: Use recvmsg() with iovecs if there's more than one segment
+ * remaining, otherwise (or if platform does not have sendmsg)
+ * use plain send(). */
+ return rd_kafka_transport_socket_recvmsg(rktrans, buf,
+ errstr, errstr_size);
+#endif
+ return rd_kafka_transport_socket_recv0(rktrans, buf,
+ errstr, errstr_size);
+}
+
+
+
+
+
+/**
+ * CONNECT state is failed (errstr!=NULL) or done (TCP is up, SSL is working..).
+ * From this state we either hand control back to the broker code,
+ * or if authentication is configured we ente the AUTH state.
+ */
+void rd_kafka_transport_connect_done (rd_kafka_transport_t *rktrans,
+ char *errstr) {
+ rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
+
+ rd_kafka_broker_connect_done(rkb, errstr);
+}
+
+
+
+#if WITH_SSL
+
+
+/**
+ * Serves the entire OpenSSL error queue and logs each error.
+ * The last error is not logged but returned in 'errstr'.
+ *
+ * If 'rkb' is non-NULL broker-specific logging will be used,
+ * else it will fall back on global 'rk' debugging.
+ */
+static char *rd_kafka_ssl_error (rd_kafka_t *rk, rd_kafka_broker_t *rkb,
+ char *errstr, size_t errstr_size) {
+ unsigned long l;
+ const char *file, *data;
+ int line, flags;
+ int cnt = 0;
+
+ while ((l = ERR_get_error_line_data(&file, &line, &data, &flags)) != 0) {
+ char buf[256];
+
+ if (cnt++ > 0) {
+ /* Log last message */
+ if (rkb)
+ rd_rkb_log(rkb, LOG_ERR, "SSL", "%s", errstr);
+ else
+ rd_kafka_log(rk, LOG_ERR, "SSL", "%s", errstr);
+ }
+
+ ERR_error_string_n(l, buf, sizeof(buf));
+
+ rd_snprintf(errstr, errstr_size, "%s:%d: %s: %s",
+ file, line, buf, (flags & ERR_TXT_STRING) ? data : "");
+
+ }
+
+ if (cnt == 0)
+ rd_snprintf(errstr, errstr_size, "No error");
+
+ return errstr;
+}
+
+
+static void rd_kafka_transport_ssl_lock_cb (int mode, int i,
+ const char *file, int line) {
+ if (mode & CRYPTO_LOCK)
+ mtx_lock(&rd_kafka_ssl_locks[i]);
+ else
+ mtx_unlock(&rd_kafka_ssl_locks[i]);
+}
+
+static unsigned long rd_kafka_transport_ssl_threadid_cb (void) {
+#ifdef _MSC_VER
+ /* Windows makes a distinction between thread handle
+ * and thread id, which means we can't use the
+ * thrd_current() API that returns the handle. */
+ return (unsigned long)GetCurrentThreadId();
+#else
+ return (unsigned long)(intptr_t)thrd_current();
+#endif
+}
+
+
+/**
+ * Global OpenSSL cleanup.
+ */
+void rd_kafka_transport_ssl_term (void) {
+ int i;
+
+ CRYPTO_set_id_callback(NULL);
+ CRYPTO_set_locking_callback(NULL);
+ CRYPTO_cleanup_all_ex_data();
+
+ for (i = 0 ; i < rd_kafka_ssl_locks_cnt ; i++)
+ mtx_destroy(&rd_kafka_ssl_locks[i]);
+
+ rd_free(rd_kafka_ssl_locks);
+
+}
+
+
+/**
+ * Global OpenSSL init.
+ */
+void rd_kafka_transport_ssl_init (void) {
+ int i;
+
+ rd_kafka_ssl_locks_cnt = CRYPTO_num_locks();
+ rd_kafka_ssl_locks = rd_malloc(rd_kafka_ssl_locks_cnt *
+ sizeof(*rd_kafka_ssl_locks));
+ for (i = 0 ; i < rd_kafka_ssl_locks_cnt ; i++)
+ mtx_init(&rd_kafka_ssl_locks[i], mtx_plain);
+
+ CRYPTO_set_id_callback(rd_kafka_transport_ssl_threadid_cb);
+ CRYPTO_set_locking_callback(rd_kafka_transport_ssl_lock_cb);
+
+ SSL_load_error_strings();
+ SSL_library_init();
+ OpenSSL_add_all_algorithms();
+}
+
+
+/**
+ * Set transport IO event polling based on SSL error.
+ *
+ * Returns -1 on permanent errors.
+ *
+ * Locality: broker thread
+ */
+static RD_INLINE int
+rd_kafka_transport_ssl_io_update (rd_kafka_transport_t *rktrans, int ret,
+ char *errstr, size_t errstr_size) {
+ int serr = SSL_get_error(rktrans->rktrans_ssl, ret);
+ int serr2;
+
+ switch (serr)
+ {
+ case SSL_ERROR_WANT_READ:
+ rd_kafka_transport_poll_set(rktrans, POLLIN);
+ break;
+
+ case SSL_ERROR_WANT_WRITE:
+ case SSL_ERROR_WANT_CONNECT:
+ rd_kafka_transport_poll_set(rktrans, POLLOUT);
+ break;
+
+ case SSL_ERROR_SYSCALL:
+ if (!(serr2 = SSL_get_error(rktrans->rktrans_ssl, ret))) {
+ if (ret == 0)
+ errno = ECONNRESET;
+ rd_snprintf(errstr, errstr_size,
+ "SSL syscall error: %s", rd_strerror(errno));
+ } else
+ rd_snprintf(errstr, errstr_size,
+ "SSL syscall error number: %d: %s", serr2,
+ rd_strerror(errno));
+ return -1;
+
+ case SSL_ERROR_ZERO_RETURN:
+ rd_snprintf(errstr, errstr_size, "Disconnected");
+ return -1;
+
+ default:
+ rd_kafka_ssl_error(NULL, rktrans->rktrans_rkb,
+ errstr, errstr_size);
+ return -1;
+ }
+
+ return 0;
+}
+
+static ssize_t
+rd_kafka_transport_ssl_send (rd_kafka_transport_t *rktrans,
+ rd_slice_t *slice,
+ char *errstr, size_t errstr_size) {
+ ssize_t sum = 0;
+ const void *p;
+ size_t rlen;
+
+ while ((rlen = rd_slice_peeker(slice, &p))) {
+ int r;
+
+ r = SSL_write(rktrans->rktrans_ssl, p, (int)rlen);
+
+ if (unlikely(r <= 0)) {
+ if (rd_kafka_transport_ssl_io_update(rktrans, r,
+ errstr,
+ errstr_size) == -1)
+ return -1;
+ else
+ return sum;
+ }
+
+ /* Update buffer read position */
+ rd_slice_read(slice, NULL, (size_t)r);
+
+ sum += r;
+ /* FIXME: remove this and try again immediately and let
+ * the next SSL_write() call fail instead? */
+ if ((size_t)r < rlen)
+ break;
+
+ }
+ return sum;
+}
+
+static ssize_t
+rd_kafka_transport_ssl_recv (rd_kafka_transport_t *rktrans,
+ rd_buf_t *rbuf, char *errstr, size_t errstr_size) {
+ ssize_t sum = 0;
+ void *p;
+ size_t len;
+
+ while ((len = rd_buf_get_writable(rbuf, &p))) {
+ int r;
+
+ r = SSL_read(rktrans->rktrans_ssl, p, (int)len);
+
+ if (unlikely(r <= 0)) {
+ if (rd_kafka_transport_ssl_io_update(rktrans, r,
+ errstr,
+ errstr_size) == -1)
+ return -1;
+ else
+ return sum;
+ }
+
+ VALGRIND_MAKE_MEM_DEFINED(p, r);
+
+ /* Update buffer write position */
+ rd_buf_write(rbuf, NULL, (size_t)r);
+
+ sum += r;
+
+ /* FIXME: remove this and try again immediately and let
+ * the next SSL_read() call fail instead? */
+ if ((size_t)r < len)
+ break;
+
+ }
+ return sum;
+
+}
+
+
+/**
+ * OpenSSL password query callback
+ *
+ * Locality: application thread
+ */
+static int rd_kafka_transport_ssl_passwd_cb (char *buf, int size, int rwflag,
+ void *userdata) {
+ rd_kafka_t *rk = userdata;
+ int pwlen;
+
+ rd_kafka_dbg(rk, SECURITY, "SSLPASSWD",
+ "Private key file \"%s\" requires password",
+ rk->rk_conf.ssl.key_location);
+
+ if (!rk->rk_conf.ssl.key_password) {
+ rd_kafka_log(rk, LOG_WARNING, "SSLPASSWD",
+ "Private key file \"%s\" requires password but "
+ "no password configured (ssl.key.password)",
+ rk->rk_conf.ssl.key_location);
+ return -1;
+ }
+
+
+ pwlen = (int) strlen(rk->rk_conf.ssl.key_password);
+ memcpy(buf, rk->rk_conf.ssl.key_password, RD_MIN(pwlen, size));
+
+ return pwlen;
+}
+
+/**
+ * Set up SSL for a newly connected connection
+ *
+ * Returns -1 on failure, else 0.
+ */
+static int rd_kafka_transport_ssl_connect (rd_kafka_broker_t *rkb,
+ rd_kafka_transport_t *rktrans,
+ char *errstr, size_t errstr_size) {
+ int r;
+ char name[RD_KAFKA_NODENAME_SIZE];
+ char *t;
+
+ rktrans->rktrans_ssl = SSL_new(rkb->rkb_rk->rk_conf.ssl.ctx);
+ if (!rktrans->rktrans_ssl)
+ goto fail;
+
+ if (!SSL_set_fd(rktrans->rktrans_ssl, rktrans->rktrans_s))
+ goto fail;
+
+#if (OPENSSL_VERSION_NUMBER >= 0x0090806fL) && !defined(OPENSSL_NO_TLSEXT)
+ /* If non-numerical hostname, send it for SNI */
+ rd_snprintf(name, sizeof(name), "%s", rkb->rkb_nodename);
+ if ((t = strrchr(name, ':')))
+ *t = '\0';
+ if (!(/*ipv6*/(strchr(name, ':') &&
+ strspn(name, "0123456789abcdefABCDEF:.[]%") == strlen(name)) ||
+ /*ipv4*/strspn(name, "0123456789.") == strlen(name)) &&
+ !SSL_set_tlsext_host_name(rktrans->rktrans_ssl, name))
+ goto fail;
+#endif
+
+ r = SSL_connect(rktrans->rktrans_ssl);
+ if (r == 1) {
+ /* Connected, highly unlikely since this is a
+ * non-blocking operation. */
+ rd_kafka_transport_connect_done(rktrans, NULL);
+ return 0;
+ }
+
+
+ if (rd_kafka_transport_ssl_io_update(rktrans, r,
+ errstr, errstr_size) == -1)
+ return -1;
+
+ return 0;
+
+ fail:
+ rd_kafka_ssl_error(NULL, rkb, errstr, errstr_size);
+ return -1;
+}
+
+
+static RD_UNUSED int
+rd_kafka_transport_ssl_io_event (rd_kafka_transport_t *rktrans, int events) {
+ int r;
+ char errstr[512];
+
+ if (events & POLLOUT) {
+ r = SSL_write(rktrans->rktrans_ssl, NULL, 0);
+ if (rd_kafka_transport_ssl_io_update(rktrans, r,
+ errstr,
+ sizeof(errstr)) == -1)
+ goto fail;
+ }
+
+ return 0;
+
+ fail:
+ /* Permanent error */
+ rd_kafka_broker_fail(rktrans->rktrans_rkb, LOG_ERR,
+ RD_KAFKA_RESP_ERR__TRANSPORT,
+ "%s", errstr);
+ return -1;
+}
+
+
+/**
+ * Verify SSL handshake was valid.
+ */
+static int rd_kafka_transport_ssl_verify (rd_kafka_transport_t *rktrans) {
+ long int rl;
+ X509 *cert;
+
+ cert = SSL_get_peer_certificate(rktrans->rktrans_ssl);
+ X509_free(cert);
+ if (!cert) {
+ rd_kafka_broker_fail(rktrans->rktrans_rkb, LOG_ERR,
+ RD_KAFKA_RESP_ERR__SSL,
+ "Broker did not provide a certificate");
+ return -1;
+ }
+
+ if ((rl = SSL_get_verify_result(rktrans->rktrans_ssl)) != X509_V_OK) {
+ rd_kafka_broker_fail(rktrans->rktrans_rkb, LOG_ERR,
+ RD_KAFKA_RESP_ERR__SSL,
+ "Failed to verify broker certificate: %s",
+ X509_verify_cert_error_string(rl));
+ return -1;
+ }
+
+ rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SSLVERIFY",
+ "Broker SSL certificate verified");
+ return 0;
+}
+
+/**
+ * SSL handshake handling.
+ * Call repeatedly (based on IO events) until handshake is done.
+ *
+ * Returns -1 on error, 0 if handshake is still in progress, or 1 on completion.
+ */
+static int rd_kafka_transport_ssl_handshake (rd_kafka_transport_t *rktrans) {
+ rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
+ char errstr[512];
+ int r;
+
+ r = SSL_do_handshake(rktrans->rktrans_ssl);
+ if (r == 1) {
+ /* SSL handshake done. Verify. */
+ if (rd_kafka_transport_ssl_verify(rktrans) == -1)
+ return -1;
+
+ rd_kafka_transport_connect_done(rktrans, NULL);
+ return 1;
+
+ } else if (rd_kafka_transport_ssl_io_update(rktrans, r,
+ errstr,
+ sizeof(errstr)) == -1) {
+ rd_kafka_broker_fail(rkb, LOG_ERR, RD_KAFKA_RESP_ERR__SSL,
+ "SSL handshake failed: %s%s", errstr,
+ strstr(errstr, "unexpected message") ?
+ ": client authentication might be "
+ "required (see broker log)" : "");
+ return -1;
+ }
+
+ return 0;
+}
+
+
+/**
+ * Once per rd_kafka_t handle cleanup of OpenSSL
+ *
+ * Locality: any thread
+ *
+ * NOTE: rd_kafka_wrlock() MUST be held
+ */
+void rd_kafka_transport_ssl_ctx_term (rd_kafka_t *rk) {
+ SSL_CTX_free(rk->rk_conf.ssl.ctx);
+ rk->rk_conf.ssl.ctx = NULL;
+}
+
+/**
+ * Once per rd_kafka_t handle initialization of OpenSSL
+ *
+ * Locality: application thread
+ *
+ * NOTE: rd_kafka_wrlock() MUST be held
+ */
+int rd_kafka_transport_ssl_ctx_init (rd_kafka_t *rk,
+ char *errstr, size_t errstr_size) {
+ int r;
+ SSL_CTX *ctx;
+
+ if (errstr_size > 0)
+ errstr[0] = '\0';
+
+ ctx = SSL_CTX_new(SSLv23_client_method());
+ if (!ctx) {
+ rd_snprintf(errstr, errstr_size,
+ "SSLv23_client_method() failed: ");
+ goto fail;
+ }
+
+#ifdef SSL_OP_NO_SSLv3
+ /* Disable SSLv3 (unsafe) */
+ SSL_CTX_set_options(ctx, SSL_OP_NO_SSLv3);
+#endif
+
+ /* Key file password callback */
+ SSL_CTX_set_default_passwd_cb(ctx, rd_kafka_transport_ssl_passwd_cb);
+ SSL_CTX_set_default_passwd_cb_userdata(ctx, rk);
+
+ /* Ciphers */
+ if (rk->rk_conf.ssl.cipher_suites) {
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "Setting cipher list: %s",
+ rk->rk_conf.ssl.cipher_suites);
+ if (!SSL_CTX_set_cipher_list(ctx,
+ rk->rk_conf.ssl.cipher_suites)) {
+ /* Set a string that will prefix the
+ * the OpenSSL error message (which is lousy)
+ * to make it more meaningful. */
+ rd_snprintf(errstr, errstr_size,
+ "ssl.cipher.suites failed: ");
+ goto fail;
+ }
+ }
+
+
+ if (rk->rk_conf.ssl.ca_location) {
+ /* CA certificate location, either file or directory. */
+ int is_dir = rd_kafka_path_is_dir(rk->rk_conf.ssl.ca_location);
+
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "Loading CA certificate(s) from %s %s",
+ is_dir ? "directory":"file",
+ rk->rk_conf.ssl.ca_location);
+
+ r = SSL_CTX_load_verify_locations(ctx,
+ !is_dir ?
+ rk->rk_conf.ssl.
+ ca_location : NULL,
+ is_dir ?
+ rk->rk_conf.ssl.
+ ca_location : NULL);
+
+ if (r != 1) {
+ rd_snprintf(errstr, errstr_size,
+ "ssl.ca.location failed: ");
+ goto fail;
+ }
+ } else {
+ /* Use default CA certificate paths: ignore failures. */
+ r = SSL_CTX_set_default_verify_paths(ctx);
+ if (r != 1)
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "SSL_CTX_set_default_verify_paths() "
+ "failed: ignoring");
+ }
+
+ if (rk->rk_conf.ssl.crl_location) {
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "Loading CRL from file %s",
+ rk->rk_conf.ssl.crl_location);
+
+ r = SSL_CTX_load_verify_locations(ctx,
+ rk->rk_conf.ssl.crl_location,
+ NULL);
+
+ if (r != 1) {
+ rd_snprintf(errstr, errstr_size,
+ "ssl.crl.location failed: ");
+ goto fail;
+ }
+
+
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "Enabling CRL checks");
+
+ X509_STORE_set_flags(SSL_CTX_get_cert_store(ctx),
+ X509_V_FLAG_CRL_CHECK);
+ }
+
+ if (rk->rk_conf.ssl.cert_location) {
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "Loading certificate from file %s",
+ rk->rk_conf.ssl.cert_location);
+
+ r = SSL_CTX_use_certificate_chain_file(ctx,
+ rk->rk_conf.ssl.cert_location);
+
+ if (r != 1) {
+ rd_snprintf(errstr, errstr_size,
+ "ssl.certificate.location failed: ");
+ goto fail;
+ }
+ }
+
+ if (rk->rk_conf.ssl.key_location) {
+ rd_kafka_dbg(rk, SECURITY, "SSL",
+ "Loading private key file from %s",
+ rk->rk_conf.ssl.key_location);
+
+ r = SSL_CTX_use_PrivateKey_file(ctx,
+ rk->rk_conf.ssl.key_location,
+ SSL_FILETYPE_PEM);
+ if (r != 1) {
+ rd_snprintf(errstr, errstr_size,
+ "ssl.key.location failed: ");
+ goto fail;
+ }
+ }
+
+
+ SSL_CTX_set_mode(ctx, SSL_MODE_ENABLE_PARTIAL_WRITE);
+
+ rk->rk_conf.ssl.ctx = ctx;
+ return 0;
+
+ fail:
+ r = (int)strlen(errstr);
+ rd_kafka_ssl_error(rk, NULL, errstr+r,
+ (int)errstr_size > r ? (int)errstr_size - r : 0);
+ SSL_CTX_free(ctx);
+
+ return -1;
+}
+
+
+#endif /* WITH_SSL */
+
+
+ssize_t
+rd_kafka_transport_send (rd_kafka_transport_t *rktrans,
+ rd_slice_t *slice, char *errstr, size_t errstr_size) {
+
+#if WITH_SSL
+ if (rktrans->rktrans_ssl)
+ return rd_kafka_transport_ssl_send(rktrans, slice,
+ errstr, errstr_size);
+ else
+#endif
+ return rd_kafka_transport_socket_send(rktrans, slice,
+ errstr, errstr_size);
+}
+
+
+ssize_t
+rd_kafka_transport_recv (rd_kafka_transport_t *rktrans, rd_buf_t *rbuf,
+ char *errstr, size_t errstr_size) {
+#if WITH_SSL
+ if (rktrans->rktrans_ssl)
+ return rd_kafka_transport_ssl_recv(rktrans, rbuf,
+ errstr, errstr_size);
+ else
+#endif
+ return rd_kafka_transport_socket_recv(rktrans, rbuf,
+ errstr, errstr_size);
+}
+
+
+
+
+/**
+ * Length framed receive handling.
+ * Currently only supports a the following framing:
+ * [int32_t:big_endian_length_of_payload][payload]
+ *
+ * To be used on POLLIN event, will return:
+ * -1: on fatal error (errstr will be updated, *rkbufp remains unset)
+ * 0: still waiting for data (*rkbufp remains unset)
+ * 1: data complete, (buffer returned in *rkbufp)
+ */
+int rd_kafka_transport_framed_recv (rd_kafka_transport_t *rktrans,
+ rd_kafka_buf_t **rkbufp,
+ char *errstr, size_t errstr_size) {
+ rd_kafka_buf_t *rkbuf = rktrans->rktrans_recv_buf;
+ ssize_t r;
+ const int log_decode_errors = LOG_ERR;
+
+ /* States:
+ * !rktrans_recv_buf: initial state; set up buf to receive header.
+ * rkbuf_totlen == 0: awaiting header
+ * rkbuf_totlen > 0: awaiting payload
+ */
+
+ if (!rkbuf) {
+ rkbuf = rd_kafka_buf_new(1, 4/*length field's length*/);
+ /* Set up buffer reader for the length field */
+ rd_buf_write_ensure(&rkbuf->rkbuf_buf, 4, 4);
+ rktrans->rktrans_recv_buf = rkbuf;
+ }
+
+
+ r = rd_kafka_transport_recv(rktrans, &rkbuf->rkbuf_buf,
+ errstr, errstr_size);
+ if (r == 0)
+ return 0;
+ else if (r == -1)
+ return -1;
+
+ if (rkbuf->rkbuf_totlen == 0) {
+ /* Frame length not known yet. */
+ int32_t frame_len;
+
+ if (rd_buf_write_pos(&rkbuf->rkbuf_buf) < sizeof(frame_len)) {
+ /* Wait for entire frame header. */
+ return 0;
+ }
+
+ /* Initialize reader */
+ rd_slice_init(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf, 0, 4);
+
+ /* Reader header: payload length */
+ rd_kafka_buf_read_i32(rkbuf, &frame_len);
+
+ if (frame_len < 0 ||
+ frame_len > rktrans->rktrans_rkb->
+ rkb_rk->rk_conf.recv_max_msg_size) {
+ rd_snprintf(errstr, errstr_size,
+ "Invalid frame size %"PRId32, frame_len);
+ return -1;
+ }
+
+ rkbuf->rkbuf_totlen = 4 + frame_len;
+ if (frame_len == 0) {
+ /* Payload is empty, we're done. */
+ rktrans->rktrans_recv_buf = NULL;
+ *rkbufp = rkbuf;
+ return 1;
+ }
+
+ /* Allocate memory to hold entire frame payload in contigious
+ * memory. */
+ rd_buf_write_ensure_contig(&rkbuf->rkbuf_buf, frame_len);
+
+ /* Try reading directly, there is probably more data available*/
+ return rd_kafka_transport_framed_recv(rktrans, rkbufp,
+ errstr, errstr_size);
+ }
+
+ if (rd_buf_write_pos(&rkbuf->rkbuf_buf) == rkbuf->rkbuf_totlen) {
+ /* Payload is complete. */
+ rktrans->rktrans_recv_buf = NULL;
+ *rkbufp = rkbuf;
+ return 1;
+ }
+
+ /* Wait for more data */
+ return 0;
+
+ err_parse:
+ if (rkbuf)
+ rd_kafka_buf_destroy(rkbuf);
+ rd_snprintf(errstr, errstr_size, "Frame header parsing failed: %s",
+ rd_kafka_err2str(rkbuf->rkbuf_err));
+ return -1;
+}
+
+
+/**
+ * TCP connection established.
+ * Set up socket options, SSL, etc.
+ *
+ * Locality: broker thread
+ */
+static void rd_kafka_transport_connected (rd_kafka_transport_t *rktrans) {
+ rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
+ unsigned int slen;
+
+ rd_rkb_dbg(rkb, BROKER, "CONNECT",
+ "Connected to %s",
+ rd_sockaddr2str(rkb->rkb_addr_last,
+ RD_SOCKADDR2STR_F_PORT |
+ RD_SOCKADDR2STR_F_FAMILY));
+
+ /* Set socket send & receive buffer sizes if configuerd */
+ if (rkb->rkb_rk->rk_conf.socket_sndbuf_size != 0) {
+ if (setsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_SNDBUF,
+ (void *)&rkb->rkb_rk->rk_conf.socket_sndbuf_size,
+ sizeof(rkb->rkb_rk->rk_conf.
+ socket_sndbuf_size)) == SOCKET_ERROR)
+ rd_rkb_log(rkb, LOG_WARNING, "SNDBUF",
+ "Failed to set socket send "
+ "buffer size to %i: %s",
+ rkb->rkb_rk->rk_conf.socket_sndbuf_size,
+ socket_strerror(socket_errno));
+ }
+
+ if (rkb->rkb_rk->rk_conf.socket_rcvbuf_size != 0) {
+ if (setsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_RCVBUF,
+ (void *)&rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
+ sizeof(rkb->rkb_rk->rk_conf.
+ socket_rcvbuf_size)) == SOCKET_ERROR)
+ rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
+ "Failed to set socket receive "
+ "buffer size to %i: %s",
+ rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
+ socket_strerror(socket_errno));
+ }
+
+ /* Get send and receive buffer sizes to allow limiting
+ * the total number of bytes passed with iovecs to sendmsg()
+ * and recvmsg(). */
+ slen = sizeof(rktrans->rktrans_rcvbuf_size);
+ if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_RCVBUF,
+ (void *)&rktrans->rktrans_rcvbuf_size,
+ &slen) == SOCKET_ERROR) {
+ rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
+ "Failed to get socket receive "
+ "buffer size: %s: assuming 1MB",
+ socket_strerror(socket_errno));
+ rktrans->rktrans_rcvbuf_size = 1024*1024;
+ } else if (rktrans->rktrans_rcvbuf_size < 1024 * 64)
+ rktrans->rktrans_rcvbuf_size = 1024*64; /* Use at least 64KB */
+
+ slen = sizeof(rktrans->rktrans_sndbuf_size);
+ if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_SNDBUF,
+ (void *)&rktrans->rktrans_sndbuf_size,
+ &slen) == SOCKET_ERROR) {
+ rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
+ "Failed to get socket send "
+ "buffer size: %s: assuming 1MB",
+ socket_strerror(socket_errno));
+ rktrans->rktrans_sndbuf_size = 1024*1024;
+ } else if (rktrans->rktrans_sndbuf_size < 1024 * 64)
+ rktrans->rktrans_sndbuf_size = 1024*64; /* Use at least 64KB */
+
+
+#ifdef TCP_NODELAY
+ if (rkb->rkb_rk->rk_conf.socket_nagle_disable) {
+ int one = 1;
+ if (setsockopt(rktrans->rktrans_s, IPPROTO_TCP, TCP_NODELAY,
+ (void *)&one, sizeof(one)) == SOCKET_ERROR)
+ rd_rkb_log(rkb, LOG_WARNING, "NAGLE",
+ "Failed to disable Nagle (TCP_NODELAY) "
+ "on socket %d: %s",
+ socket_strerror(socket_errno));
+ }
+#endif
+
+
+#if WITH_SSL
+ if (rkb->rkb_proto == RD_KAFKA_PROTO_SSL ||
+ rkb->rkb_proto == RD_KAFKA_PROTO_SASL_SSL) {
+ char errstr[512];
+
+ /* Set up SSL connection.
+ * This is also an asynchronous operation so dont
+ * propagate to broker_connect_done() just yet. */
+ if (rd_kafka_transport_ssl_connect(rkb, rktrans,
+ errstr,
+ sizeof(errstr)) == -1) {
+ rd_kafka_transport_connect_done(rktrans, errstr);
+ return;
+ }
+ return;
+ }
+#endif
+
+ /* Propagate connect success */
+ rd_kafka_transport_connect_done(rktrans, NULL);
+}
+
+
+
+/**
+ * @brief the kernel SO_ERROR in \p errp for the given transport.
+ * @returns 0 if getsockopt() was succesful (and \p and errp can be trusted),
+ * else -1 in which case \p errp 's value is undefined.
+ */
+static int rd_kafka_transport_get_socket_error (rd_kafka_transport_t *rktrans,
+ int *errp) {
+ socklen_t intlen = sizeof(*errp);
+
+ if (getsockopt(rktrans->rktrans_s, SOL_SOCKET,
+ SO_ERROR, (void *)errp, &intlen) == -1) {
+ rd_rkb_dbg(rktrans->rktrans_rkb, BROKER, "SO_ERROR",
+ "Failed to get socket error: %s",
+ socket_strerror(socket_errno));
+ return -1;
+ }
+
+ return 0;
+}
+
+
+/**
+ * IO event handler.
+ *
+ * Locality: broker thread
+ */
+static void rd_kafka_transport_io_event (rd_kafka_transport_t *rktrans,
+ int events) {
+ char errstr[512];
+ int r;
+ rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
+
+ switch (rkb->rkb_state)
+ {
+ case RD_KAFKA_BROKER_STATE_CONNECT:
+#if WITH_SSL
+ if (rktrans->rktrans_ssl) {
+ /* Currently setting up SSL connection:
+ * perform handshake. */
+ rd_kafka_transport_ssl_handshake(rktrans);
+ return;
+ }
+#endif
+
+ /* Asynchronous connect finished, read status. */
+ if (!(events & (POLLOUT|POLLERR|POLLHUP)))
+ return;
+
+ if (rd_kafka_transport_get_socket_error(rktrans, &r) == -1) {
+ rd_kafka_broker_fail(
+ rkb, LOG_ERR, RD_KAFKA_RESP_ERR__TRANSPORT,
+ "Connect to %s failed: "
+ "unable to get status from "
+ "socket %d: %s",
+ rd_sockaddr2str(rkb->rkb_addr_last,
+ RD_SOCKADDR2STR_F_PORT |
+ RD_SOCKADDR2STR_F_FAMILY),
+ rktrans->rktrans_s,
+ rd_strerror(socket_errno));
+ } else if (r != 0) {
+ /* Connect failed */
+ errno = r;
+ rd_snprintf(errstr, sizeof(errstr),
+ "Connect to %s failed: %s",
+ rd_sockaddr2str(rkb->rkb_addr_last,
+ RD_SOCKADDR2STR_F_PORT |
+ RD_SOCKADDR2STR_F_FAMILY),
+ rd_strerror(r));
+
+ rd_kafka_transport_connect_done(rktrans, errstr);
+ } else {
+ /* Connect succeeded */
+ rd_kafka_transport_connected(rktrans);
+ }
+ break;
+
+ case RD_KAFKA_BROKER_STATE_AUTH:
+ /* SASL handshake */
+ if (rd_kafka_sasl_io_event(rktrans, events,
+ errstr, sizeof(errstr)) == -1) {
+ errno = EINVAL;
+ rd_kafka_broker_fail(rkb, LOG_ERR,
+ RD_KAFKA_RESP_ERR__AUTHENTICATION,
+ "SASL authentication failure: %s",
+ errstr);
+ return;
+ }
+ break;
+
+ case RD_KAFKA_BROKER_STATE_APIVERSION_QUERY:
+ case RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE:
+ case RD_KAFKA_BROKER_STATE_UP:
+ case RD_KAFKA_BROKER_STATE_UPDATE:
+
+ if (events & POLLIN) {
+ while (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP &&
+ rd_kafka_recv(rkb) > 0)
+ ;
+ }
+
+ if (events & POLLHUP) {
+ rd_kafka_broker_fail(rkb,
+ rkb->rkb_rk->rk_conf.
+ log_connection_close ?
+ LOG_NOTICE : LOG_DEBUG,
+ RD_KAFKA_RESP_ERR__TRANSPORT,
+ "Connection closed");
+ return;
+ }
+
+ if (events & POLLOUT) {
+ while (rd_kafka_send(rkb) > 0)
+ ;
+ }
+ break;
+
+ case RD_KAFKA_BROKER_STATE_INIT:
+ case RD_KAFKA_BROKER_STATE_DOWN:
+ rd_kafka_assert(rkb->rkb_rk, !*"bad state");
+ }
+}
+
+
+/**
+ * Poll and serve IOs
+ *
+ * Locality: broker thread
+ */
+void rd_kafka_transport_io_serve (rd_kafka_transport_t *rktrans,
+ int timeout_ms) {
+ rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
+ int events;
+
+ if (rd_kafka_bufq_cnt(&rkb->rkb_waitresps) < rkb->rkb_max_inflight &&
+ rd_kafka_bufq_cnt(&rkb->rkb_outbufs) > 0)
+ rd_kafka_transport_poll_set(rkb->rkb_transport, POLLOUT);
+
+ if ((events = rd_kafka_transport_poll(rktrans, timeout_ms)) <= 0)
+ return;
+
+ rd_kafka_transport_poll_clear(rktrans, POLLOUT);
+
+ rd_kafka_transport_io_event(rktrans, events);
+}
+
+
+/**
+ * Initiate asynchronous connection attempt.
+ *
+ * Locality: broker thread
+ */
+rd_kafka_transport_t *rd_kafka_transport_connect (rd_kafka_broker_t *rkb,
+ const rd_sockaddr_inx_t *sinx,
+ char *errstr,
+ size_t errstr_size) {
+ rd_kafka_transport_t *rktrans;
+ int s = -1;
+ int on = 1;
+ int r;
+
+ rkb->rkb_addr_last = sinx;
+
+ s = rkb->rkb_rk->rk_conf.socket_cb(sinx->in.sin_family,
+ SOCK_STREAM, IPPROTO_TCP,
+ rkb->rkb_rk->rk_conf.opaque);
+ if (s == -1) {
+ rd_snprintf(errstr, errstr_size, "Failed to create socket: %s",
+ socket_strerror(socket_errno));
+ return NULL;
+ }
+
+
+#ifdef SO_NOSIGPIPE
+ /* Disable SIGPIPE signalling for this socket on OSX */
+ if (setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, &on, sizeof(on)) == -1)
+ rd_rkb_dbg(rkb, BROKER, "SOCKET",
+ "Failed to set SO_NOSIGPIPE: %s",
+ socket_strerror(socket_errno));
+#endif
+
+ /* Enable TCP keep-alives, if configured. */
+ if (rkb->rkb_rk->rk_conf.socket_keepalive) {
+#ifdef SO_KEEPALIVE
+ if (setsockopt(s, SOL_SOCKET, SO_KEEPALIVE,
+ (void *)&on, sizeof(on)) == SOCKET_ERROR)
+ rd_rkb_dbg(rkb, BROKER, "SOCKET",
+ "Failed to set SO_KEEPALIVE: %s",
+ socket_strerror(socket_errno));
+#else
+ rd_rkb_dbg(rkb, BROKER, "SOCKET",
+ "System does not support "
+ "socket.keepalive.enable (SO_KEEPALIVE)");
+#endif
+ }
+
+ /* Set the socket to non-blocking */
+ if ((r = rd_fd_set_nonblocking(s))) {
+ rd_snprintf(errstr, errstr_size,
+ "Failed to set socket non-blocking: %s",
+ socket_strerror(r));
+ goto err;
+ }
+
+ rd_rkb_dbg(rkb, BROKER, "CONNECT", "Connecting to %s (%s) "
+ "with socket %i",
+ rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_FAMILY |
+ RD_SOCKADDR2STR_F_PORT),
+ rd_kafka_secproto_names[rkb->rkb_proto], s);
+
+ /* Connect to broker */
+ if (rkb->rkb_rk->rk_conf.connect_cb) {
+ r = rkb->rkb_rk->rk_conf.connect_cb(
+ s, (struct sockaddr *)sinx, RD_SOCKADDR_INX_LEN(sinx),
+ rkb->rkb_name, rkb->rkb_rk->rk_conf.opaque);
+ } else {
+ if (connect(s, (struct sockaddr *)sinx,
+ RD_SOCKADDR_INX_LEN(sinx)) == SOCKET_ERROR &&
+ (socket_errno != EINPROGRESS
+#ifdef _MSC_VER
+ && socket_errno != WSAEWOULDBLOCK
+#endif
+ ))
+ r = socket_errno;
+ else
+ r = 0;
+ }
+
+ if (r != 0) {
+ rd_rkb_dbg(rkb, BROKER, "CONNECT",
+ "couldn't connect to %s: %s (%i)",
+ rd_sockaddr2str(sinx,
+ RD_SOCKADDR2STR_F_PORT |
+ RD_SOCKADDR2STR_F_FAMILY),
+ socket_strerror(r), r);
+ rd_snprintf(errstr, errstr_size,
+ "Failed to connect to broker at %s: %s",
+ rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_NICE),
+ socket_strerror(r));
+ goto err;
+ }
+
+ /* Create transport handle */
+ rktrans = rd_calloc(1, sizeof(*rktrans));
+ rktrans->rktrans_rkb = rkb;
+ rktrans->rktrans_s = s;
+ rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt++].fd = s;
+ if (rkb->rkb_wakeup_fd[0] != -1) {
+ rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt].events = POLLIN;
+ rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt++].fd = rkb->rkb_wakeup_fd[0];
+ }
+
+
+ /* Poll writability to trigger on connection success/failure. */
+ rd_kafka_transport_poll_set(rktrans, POLLOUT);
+
+ return rktrans;
+
+ err:
+ if (s != -1)
+ rd_kafka_transport_close0(rkb->rkb_rk, s);
+
+ return NULL;
+}
+
+
+
+void rd_kafka_transport_poll_set(rd_kafka_transport_t *rktrans, int event) {
+ rktrans->rktrans_pfd[0].events |= event;
+}
+
+void rd_kafka_transport_poll_clear(rd_kafka_transport_t *rktrans, int event) {
+ rktrans->rktrans_pfd[0].events &= ~event;
+}
+
+
+int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout) {
+ int r;
+#ifndef _MSC_VER
+ r = poll(rktrans->rktrans_pfd, rktrans->rktrans_pfd_cnt, tmout);
+ if (r <= 0)
+ return r;
+#else
+ r = WSAPoll(rktrans->rktrans_pfd, rktrans->rktrans_pfd_cnt, tmout);
+ if (r == 0) {
+ /* Workaround for broken WSAPoll() while connecting:
+ * failed connection attempts are not indicated at all by WSAPoll()
+ * so we need to check the socket error when Poll returns 0.
+ * Issue #525 */
+ r = ECONNRESET;
+ if (unlikely(rktrans->rktrans_rkb->rkb_state ==
+ RD_KAFKA_BROKER_STATE_CONNECT &&
+ (rd_kafka_transport_get_socket_error(rktrans,
+ &r) == -1 ||
+ r != 0))) {
+ char errstr[512];
+ errno = r;
+ rd_snprintf(errstr, sizeof(errstr),
+ "Connect to %s failed: %s",
+ rd_sockaddr2str(rktrans->rktrans_rkb->
+ rkb_addr_last,
+ RD_SOCKADDR2STR_F_PORT |
+ RD_SOCKADDR2STR_F_FAMILY),
+ socket_strerror(r));
+ rd_kafka_transport_connect_done(rktrans, errstr);
+ return -1;
+ } else
+ return 0;
+ } else if (r == SOCKET_ERROR)
+ return -1;
+#endif
+ rd_atomic64_add(&rktrans->rktrans_rkb->rkb_c.wakeups, 1);
+
+ if (rktrans->rktrans_pfd[1].revents & POLLIN) {
+ /* Read wake-up fd data and throw away, just used for wake-ups*/
+ char buf[512];
+ if (rd_read((int)rktrans->rktrans_pfd[1].fd,
+ buf, sizeof(buf)) == -1) {
+ /* Ignore warning */
+ }
+ }
+
+ return rktrans->rktrans_pfd[0].revents;
+}
+
+
+
+
+
+#if 0
+/**
+ * Global cleanup.
+ * This is dangerous and SHOULD NOT be called since it will rip
+ * the rug from under the application if it uses any of this functionality
+ * in its own code. This means we might leak some memory on exit.
+ */
+void rd_kafka_transport_term (void) {
+#ifdef _MSC_VER
+ (void)WSACleanup(); /* FIXME: dangerous */
+#endif
+}
+#endif
+
+void rd_kafka_transport_init(void) {
+#ifdef _MSC_VER
+ WSADATA d;
+ (void)WSAStartup(MAKEWORD(2, 2), &d);
+#endif
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.h b/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.h
new file mode 100644
index 0000000..fcd2580
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdkafka_transport.h
@@ -0,0 +1,72 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2015, 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.
+ */
+
+#pragma once
+
+#ifndef _MSC_VER
+#include <poll.h>
+#endif
+
+#include "rdbuf.h"
+#include "rdaddr.h"
+
+typedef struct rd_kafka_transport_s rd_kafka_transport_t;
+
+void rd_kafka_transport_io_serve (rd_kafka_transport_t *rktrans,
+ int timeout_ms);
+
+ssize_t rd_kafka_transport_send (rd_kafka_transport_t *rktrans,
+ rd_slice_t *slice,
+ char *errstr, size_t errstr_size);
+ssize_t rd_kafka_transport_recv (rd_kafka_transport_t *rktrans,
+ rd_buf_t *rbuf,
+ char *errstr, size_t errstr_size);
+int rd_kafka_transport_framed_recv (rd_kafka_transport_t *rktrans,
+ rd_kafka_buf_t **rkbufp,
+ char *errstr, size_t errstr_size);
+struct rd_kafka_broker_s;
+rd_kafka_transport_t *rd_kafka_transport_connect(struct rd_kafka_broker_s *rkb, const rd_sockaddr_inx_t *sinx,
+ char *errstr, size_t errstr_size);
+void rd_kafka_transport_connect_done (rd_kafka_transport_t *rktrans,
+ char *errstr);
+
+void rd_kafka_transport_close(rd_kafka_transport_t *rktrans);
+void rd_kafka_transport_poll_set(rd_kafka_transport_t *rktrans, int event);
+void rd_kafka_transport_poll_clear(rd_kafka_transport_t *rktrans, int event);
+int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout);
+
+#if WITH_SSL
+void rd_kafka_transport_ssl_ctx_term (rd_kafka_t *rk);
+int rd_kafka_transport_ssl_ctx_init (rd_kafka_t *rk,
+ char *errstr, size_t errstr_size);
+
+void rd_kafka_transport_ssl_term (void);
+void rd_kafka_transport_ssl_init (void);
+#endif
+void rd_kafka_transport_term (void);
+void rd_kafka_transport_init(void);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdkafka_transport_int.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_transport_int.h b/thirdparty/librdkafka-0.11.1/src/rdkafka_transport_int.h
new file mode 100644
index 0000000..8ae79b4
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdkafka_transport_int.h
@@ -0,0 +1,84 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2015, 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.
+ */
+#pragma once
+
+/* This header file is to be used by .c files needing access to the
+ * rd_kafka_transport_t struct internals. */
+
+#include "rdkafka_sasl.h"
+
+#if WITH_SSL
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+#endif
+
+struct rd_kafka_transport_s {
+ int rktrans_s;
+
+ rd_kafka_broker_t *rktrans_rkb;
+
+#if WITH_SSL
+ SSL *rktrans_ssl;
+#endif
+
+ struct {
+ void *state; /* SASL implementation
+ * state handle */
+
+ int complete; /* Auth was completed early
+ * from the client's perspective
+ * (but we might still have to
+ * wait for server reply). */
+
+ /* SASL framing buffers */
+ struct msghdr msg;
+ struct iovec iov[2];
+
+ char *recv_buf;
+ int recv_of; /* Received byte count */
+ int recv_len; /* Expected receive length for
+ * current frame. */
+ } rktrans_sasl;
+
+ rd_kafka_buf_t *rktrans_recv_buf; /* Used with framed_recvmsg */
+
+ /* Two pollable fds:
+ * - TCP socket
+ * - wake-up fd
+ */
+#ifndef _MSC_VER
+ struct pollfd rktrans_pfd[2];
+#else
+ WSAPOLLFD rktrans_pfd[2];
+#endif
+ int rktrans_pfd_cnt;
+
+ size_t rktrans_rcvbuf_size; /**< Socket receive buffer size */
+ size_t rktrans_sndbuf_size; /**< Socket send buffer size */
+};
+
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdlist.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdlist.c b/thirdparty/librdkafka-0.11.1/src/rdlist.c
new file mode 100644
index 0000000..11cf14d
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdlist.c
@@ -0,0 +1,313 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2012-2015, 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 "rd.h"
+#include "rdlist.h"
+
+
+void rd_list_dump (const char *what, const rd_list_t *rl) {
+ int i;
+ printf("%s: (rd_list_t*)%p cnt %d, size %d, elems %p:\n",
+ what, rl, rl->rl_cnt, rl->rl_size, rl->rl_elems);
+ for (i = 0 ; i < rl->rl_cnt ; i++)
+ printf(" #%d: %p at &%p\n", i,
+ rl->rl_elems[i], &rl->rl_elems[i]);
+}
+
+void rd_list_grow (rd_list_t *rl, size_t size) {
+ rd_assert(!(rl->rl_flags & RD_LIST_F_FIXED_SIZE));
+ rl->rl_size += (int)size;
+ if (unlikely(rl->rl_size == 0))
+ return; /* avoid zero allocations */
+ rl->rl_elems = rd_realloc(rl->rl_elems,
+ sizeof(*rl->rl_elems) * rl->rl_size);
+}
+
+rd_list_t *
+rd_list_init (rd_list_t *rl, int initial_size, void (*free_cb) (void *)) {
+ memset(rl, 0, sizeof(*rl));
+
+ if (initial_size > 0)
+ rd_list_grow(rl, initial_size);
+
+ rl->rl_free_cb = free_cb;
+
+ return rl;
+}
+
+rd_list_t *rd_list_new (int initial_size, void (*free_cb) (void *)) {
+ rd_list_t *rl = malloc(sizeof(*rl));
+ rd_list_init(rl, initial_size, free_cb);
+ rl->rl_flags |= RD_LIST_F_ALLOCATED;
+ return rl;
+}
+
+void rd_list_prealloc_elems (rd_list_t *rl, size_t elemsize, size_t size) {
+ size_t allocsize;
+ char *p;
+ size_t i;
+
+ rd_assert(!rl->rl_elems);
+
+ /* Allocation layout:
+ * void *ptrs[cnt];
+ * elems[elemsize][cnt];
+ */
+
+ allocsize = (sizeof(void *) * size) + (elemsize * size);
+ rl->rl_elems = rd_malloc(allocsize);
+
+ /* p points to first element's memory. */
+ p = (char *)&rl->rl_elems[size];
+
+ /* Pointer -> elem mapping */
+ for (i = 0 ; i < size ; i++, p += elemsize)
+ rl->rl_elems[i] = p;
+
+ rl->rl_size = (int)size;
+ rl->rl_cnt = 0;
+ rl->rl_flags |= RD_LIST_F_FIXED_SIZE;
+}
+
+
+void rd_list_free_cb (rd_list_t *rl, void *ptr) {
+ if (rl->rl_free_cb && ptr)
+ rl->rl_free_cb(ptr);
+}
+
+
+void *rd_list_add (rd_list_t *rl, void *elem) {
+ if (rl->rl_cnt == rl->rl_size)
+ rd_list_grow(rl, rl->rl_size ? rl->rl_size * 2 : 16);
+ rl->rl_flags &= ~RD_LIST_F_SORTED;
+ if (elem)
+ rl->rl_elems[rl->rl_cnt] = elem;
+ return rl->rl_elems[rl->rl_cnt++];
+}
+
+static void rd_list_remove0 (rd_list_t *rl, int idx) {
+ rd_assert(idx < rl->rl_cnt);
+
+ if (idx + 1 < rl->rl_cnt)
+ memmove(&rl->rl_elems[idx],
+ &rl->rl_elems[idx+1],
+ sizeof(*rl->rl_elems) * (rl->rl_cnt - (idx+1)));
+ rl->rl_cnt--;
+}
+
+void *rd_list_remove (rd_list_t *rl, void *match_elem) {
+ void *elem;
+ int i;
+
+ RD_LIST_FOREACH(elem, rl, i) {
+ if (elem == match_elem) {
+ rd_list_remove0(rl, i);
+ return elem;
+ }
+ }
+
+ return NULL;
+}
+
+
+void *rd_list_remove_cmp (rd_list_t *rl, void *match_elem,
+ int (*cmp) (void *_a, void *_b)) {
+ void *elem;
+ int i;
+
+ RD_LIST_FOREACH(elem, rl, i) {
+ if (match_elem == cmp ||
+ !cmp(elem, match_elem)) {
+ rd_list_remove0(rl, i);
+ return elem;
+ }
+ }
+
+ return NULL;
+}
+
+
+/**
+ * Trampoline to avoid the double pointers in callbacks.
+ *
+ * rl_elems is a **, but to avoid having the application do the cumbersome
+ * ** -> * casting we wrap this here and provide a simple * pointer to the
+ * the callbacks.
+ *
+ * This is true for all list comparator uses, i.e., both sort() and find().
+ */
+static RD_TLS int (*rd_list_cmp_curr) (const void *, const void *);
+
+static RD_INLINE
+int rd_list_cmp_trampoline (const void *_a, const void *_b) {
+ const void *a = *(const void **)_a, *b = *(const void **)_b;
+
+ return rd_list_cmp_curr(a, b);
+}
+
+void rd_list_sort (rd_list_t *rl, int (*cmp) (const void *, const void *)) {
+ rd_list_cmp_curr = cmp;
+ qsort(rl->rl_elems, rl->rl_cnt, sizeof(*rl->rl_elems),
+ rd_list_cmp_trampoline);
+ rl->rl_flags |= RD_LIST_F_SORTED;
+}
+
+void rd_list_clear (rd_list_t *rl) {
+ rl->rl_cnt = 0;
+ rl->rl_flags &= ~RD_LIST_F_SORTED;
+}
+
+
+void rd_list_destroy (rd_list_t *rl) {
+
+ if (rl->rl_elems) {
+ int i;
+ if (rl->rl_free_cb) {
+ for (i = 0 ; i < rl->rl_cnt ; i++)
+ if (rl->rl_elems[i])
+ rl->rl_free_cb(rl->rl_elems[i]);
+ }
+
+ rd_free(rl->rl_elems);
+ }
+
+ if (rl->rl_flags & RD_LIST_F_ALLOCATED)
+ rd_free(rl);
+}
+
+
+void *rd_list_elem (const rd_list_t *rl, int idx) {
+ if (likely(idx < rl->rl_cnt))
+ return (void *)rl->rl_elems[idx];
+ return NULL;
+}
+
+void *rd_list_find (const rd_list_t *rl, const void *match,
+ int (*cmp) (const void *, const void *)) {
+ int i;
+ const void *elem;
+
+ if (rl->rl_flags & RD_LIST_F_SORTED) {
+ void **r;
+ rd_list_cmp_curr = cmp;
+ r = bsearch(&match/*ptrptr to match elems*/,
+ rl->rl_elems, rl->rl_cnt,
+ sizeof(*rl->rl_elems), rd_list_cmp_trampoline);
+ return r ? *r : NULL;
+ }
+
+ RD_LIST_FOREACH(elem, rl, i) {
+ if (!cmp(match, elem))
+ return (void *)elem;
+ }
+
+ return NULL;
+}
+
+
+int rd_list_cmp (const rd_list_t *a, rd_list_t *b,
+ int (*cmp) (const void *, const void *)) {
+ int i;
+
+ i = a->rl_cnt - b->rl_cnt;
+ if (i)
+ return i;
+
+ for (i = 0 ; i < a->rl_cnt ; i++) {
+ int r = cmp(a->rl_elems[i], b->rl_elems[i]);
+ if (r)
+ return r;
+ }
+
+ return 0;
+}
+
+
+/**
+ * @brief Simple element pointer comparator
+ */
+int rd_list_cmp_ptr (const void *a, const void *b) {
+ if (a < b)
+ return -1;
+ else if (a > b)
+ return 1;
+ return 0;
+}
+
+
+void rd_list_apply (rd_list_t *rl,
+ int (*cb) (void *elem, void *opaque), void *opaque) {
+ void *elem;
+ int i;
+
+ RD_LIST_FOREACH(elem, rl, i) {
+ if (!cb(elem, opaque)) {
+ rd_list_remove0(rl, i);
+ i--;
+ }
+ }
+
+ return;
+}
+
+
+/**
+ * @brief Default element copier that simply assigns the original pointer.
+ */
+static void *rd_list_nocopy_ptr (const void *elem, void *opaque) {
+ return (void *)elem;
+}
+
+
+rd_list_t *rd_list_copy (const rd_list_t *src,
+ void *(*copy_cb) (const void *elem, void *opaque),
+ void *opaque) {
+ rd_list_t *dst;
+
+ dst = rd_list_new(src->rl_cnt, src->rl_free_cb);
+
+ rd_list_copy_to(dst, src, copy_cb, opaque);
+ return dst;
+}
+
+
+void rd_list_copy_to (rd_list_t *dst, const rd_list_t *src,
+ void *(*copy_cb) (const void *elem, void *opaque),
+ void *opaque) {
+ void *elem;
+ int i;
+
+ if (!copy_cb)
+ copy_cb = rd_list_nocopy_ptr;
+
+ RD_LIST_FOREACH(elem, src, i) {
+ void *celem = copy_cb(elem, opaque);
+ if (celem)
+ rd_list_add(dst, celem);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdlist.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdlist.h b/thirdparty/librdkafka-0.11.1/src/rdlist.h
new file mode 100644
index 0000000..27233e3
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdlist.h
@@ -0,0 +1,246 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2012-2015, 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.
+ */
+
+#pragma once
+
+
+/**
+ *
+ * Simple light-weight append-only list to be used as a collection convenience.
+ *
+ */
+
+typedef struct rd_list_s {
+ int rl_size;
+ int rl_cnt;
+ void **rl_elems;
+ void (*rl_free_cb) (void *);
+ int rl_flags;
+#define RD_LIST_F_ALLOCATED 0x1 /* The rd_list_t is allocated,
+ * will be free on destroy() */
+#define RD_LIST_F_SORTED 0x2 /* Set by sort(), cleared by any mutations.
+ * When this flag is set bsearch() is used
+ * by find(), otherwise a linear search. */
+#define RD_LIST_F_FIXED_SIZE 0x4 /* Assert on grow */
+#define RD_LIST_F_UNIQUE 0x8 /* Don't allow duplicates:
+ * ONLY ENFORCED BY CALLER. */
+} rd_list_t;
+
+
+/**
+ * @brief Initialize a list, preallocate space for 'initial_size' elements
+ * (optional).
+ * List elements will optionally be freed by \p free_cb.
+ *
+ * @returns \p rl
+ */
+rd_list_t *
+rd_list_init (rd_list_t *rl, int initial_size, void (*free_cb) (void *));
+
+
+/**
+ * Allocate a new list pointer and initialize it according to rd_list_init().
+ *
+ * Use rd_list_destroy() to free.
+ */
+rd_list_t *rd_list_new (int initial_size, void (*free_cb) (void *));
+
+
+/**
+ * @brief Prepare list to for an additional \p size elements.
+ * This is an optimization to avoid incremental grows.
+ */
+void rd_list_grow (rd_list_t *rl, size_t size);
+
+/**
+ * @brief Preallocate elements to avoid having to pass an allocated pointer to
+ * rd_list_add(), instead pass NULL to rd_list_add() and use the returned
+ * pointer as the element.
+ *
+ * @param elemsize element size
+ * @param size number of elements
+ *
+ * @remark Preallocated element lists can't grow past \p size.
+ */
+void rd_list_prealloc_elems (rd_list_t *rl, size_t elemsize, size_t size);
+
+
+/**
+ * @brief Free a pointer using the list's free_cb
+ *
+ * @remark If no free_cb is set, or \p ptr is NULL, dont do anything
+ *
+ * Typical use is rd_list_free_cb(rd_list_remove_cmp(....));
+ */
+void rd_list_free_cb (rd_list_t *rl, void *ptr);
+
+
+/**
+ * @brief Append element to list
+ *
+ * @returns \p elem. If \p elem is NULL the default element for that index
+ * will be returned (for use with set_elems).
+ */
+void *rd_list_add (rd_list_t *rl, void *elem);
+
+
+/**
+ * Remove element from list.
+ * This is a slow O(n) + memmove operation.
+ * Returns the removed element.
+ */
+void *rd_list_remove (rd_list_t *rl, void *match_elem);
+
+/**
+ * Remove element from list using comparator.
+ * See rd_list_remove()
+ */
+void *rd_list_remove_cmp (rd_list_t *rl, void *match_elem,
+ int (*cmp) (void *_a, void *_b));
+
+/**
+ * Sort list using comparator
+ */
+void rd_list_sort (rd_list_t *rl, int (*cmp) (const void *, const void *));
+
+
+/**
+ * Empties the list (but does not free any memory)
+ */
+void rd_list_clear (rd_list_t *rl);
+
+
+/**
+ * Empties the list, frees the element array, and optionally frees
+ * each element using the registered \c rl->rl_free_cb.
+ *
+ * If the list was previously allocated with rd_list_new() it will be freed.
+ */
+void rd_list_destroy (rd_list_t *rl);
+
+
+/**
+ * Returns the element at index 'idx', or NULL if out of range.
+ *
+ * Typical iteration is:
+ * int i = 0;
+ * my_type_t *obj;
+ * while ((obj = rd_list_elem(rl, i++)))
+ * do_something(obj);
+ */
+void *rd_list_elem (const rd_list_t *rl, int idx);
+
+#define RD_LIST_FOREACH(elem,listp,idx) \
+ for (idx = 0 ; (elem = rd_list_elem(listp, idx)) ; idx++)
+
+#define RD_LIST_FOREACH_REVERSE(elem,listp,idx) \
+ for (idx = (listp)->rl_cnt-1 ; \
+ idx >= 0 && (elem = rd_list_elem(listp, idx)) ; idx--)
+
+/**
+ * Returns the number of elements in list.
+ */
+static RD_INLINE RD_UNUSED int rd_list_cnt (const rd_list_t *rl) {
+ return rl->rl_cnt;
+}
+
+
+/**
+ * Returns true if list is empty
+ */
+#define rd_list_empty(rl) (rd_list_cnt(rl) == 0)
+
+
+
+/**
+ * Find element using comparator
+ * 'match' will be the first argument to 'cmp', and each element (up to a match)
+ * will be the second argument to 'cmp'.
+ */
+void *rd_list_find (const rd_list_t *rl, const void *match,
+ int (*cmp) (const void *, const void *));
+
+
+
+/**
+ * @brief Compare list \p a to \p b.
+ *
+ * @returns < 0 if a was "lesser" than b,
+ * > 0 if a was "greater" than b,
+ * 0 if a and b are equal.
+ */
+int rd_list_cmp (const rd_list_t *a, rd_list_t *b,
+ int (*cmp) (const void *, const void *));
+
+/**
+ * @brief Simple element pointer comparator
+ */
+int rd_list_cmp_ptr (const void *a, const void *b);
+
+
+/**
+ * @brief Apply \p cb to each element in list, if \p cb returns 0
+ * the element will be removed (but not freed).
+ */
+void rd_list_apply (rd_list_t *rl,
+ int (*cb) (void *elem, void *opaque), void *opaque);
+
+
+
+/**
+ * @brief Copy list \p src, returning a new list,
+ * using optional \p copy_cb (per elem)
+ */
+rd_list_t *rd_list_copy (const rd_list_t *src,
+ void *(*copy_cb) (const void *elem, void *opaque),
+ void *opaque);
+
+
+/**
+ * @brief Copy list \p src to \p dst using optional \p copy_cb (per elem)
+ * @remark The destination list is not initialized or copied by this function.
+ * @remark copy_cb() may return NULL in which case no element is added,
+ * but the copy callback might have done so itself.
+ */
+void rd_list_copy_to (rd_list_t *dst, const rd_list_t *src,
+ void *(*copy_cb) (const void *elem, void *opaque),
+ void *opaque);
+
+/**
+ * @brief String copier for rd_list_copy()
+ */
+static RD_UNUSED
+void *rd_list_string_copy (const void *elem, void *opaque) {
+ return rd_strdup((const char *)elem);
+}
+
+
+/**
+ * Debugging: Print list to stdout.
+ */
+void rd_list_dump (const char *what, const rd_list_t *rl);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdlog.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdlog.c b/thirdparty/librdkafka-0.11.1/src/rdlog.c
new file mode 100644
index 0000000..3f0d29a
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdlog.c
@@ -0,0 +1,89 @@
+/*
+ * librd - Rapid Development 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.
+ */
+
+#include "rdkafka_int.h"
+#include "rdlog.h"
+
+#include <stdarg.h>
+#include <string.h>
+#include <ctype.h>
+
+
+
+
+void rd_hexdump (FILE *fp, const char *name, const void *ptr, size_t len) {
+ const char *p = (const char *)ptr;
+ size_t of = 0;
+
+
+ if (name)
+ fprintf(fp, "%s hexdump (%"PRIusz" bytes):\n", name, len);
+
+ for (of = 0 ; of < len ; of += 16) {
+ char hexen[16*3+1];
+ char charen[16+1];
+ int hof = 0;
+
+ int cof = 0;
+ unsigned int i;
+
+ for (i = (unsigned int)of ; i < (unsigned int)of + 16 && i < len ; i++) {
+ hof += rd_snprintf(hexen+hof, sizeof(hexen)-hof,
+ "%02x ",
+ p[i] & 0xff);
+ cof += rd_snprintf(charen+cof, sizeof(charen)-cof, "%c",
+ isprint((int)p[i]) ? p[i] : '.');
+ }
+ fprintf(fp, "%08zx: %-48s %-16s\n",
+ of, hexen, charen);
+ }
+}
+
+
+void rd_iov_print (const char *what, int iov_idx, const struct iovec *iov,
+ int hexdump) {
+ printf("%s: iov #%i: %"PRIusz"\n", what, iov_idx,
+ (size_t)iov->iov_len);
+ if (hexdump)
+ rd_hexdump(stdout, what, iov->iov_base, iov->iov_len);
+}
+
+
+void rd_msghdr_print (const char *what, const struct msghdr *msg,
+ int hexdump) {
+ int i;
+ size_t len = 0;
+
+ printf("%s: iovlen %"PRIusz"\n", what, (size_t)msg->msg_iovlen);
+
+ for (i = 0 ; i < (int)msg->msg_iovlen ; i++) {
+ rd_iov_print(what, i, &msg->msg_iov[i], hexdump);
+ len += msg->msg_iov[i].iov_len;
+ }
+ printf("%s: ^ message was %"PRIusz" bytes in total\n", what, len);
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdlog.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdlog.h b/thirdparty/librdkafka-0.11.1/src/rdlog.h
new file mode 100644
index 0000000..95066e2
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdlog.h
@@ -0,0 +1,37 @@
+/*
+ * librd - Rapid Development 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.
+ */
+
+#pragma once
+
+void rd_hexdump (FILE *fp, const char *name, const void *ptr, size_t len);
+
+void rd_iov_print (const char *what, int iov_idx, const struct iovec *iov,
+ int hexdump);
+struct msghdr;
+void rd_msghdr_print (const char *what, const struct msghdr *msg,
+ int hexdump);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdports.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdports.c b/thirdparty/librdkafka-0.11.1/src/rdports.c
new file mode 100644
index 0000000..a34195b
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdports.c
@@ -0,0 +1,60 @@
+/*
+* librdkafka - Apache Kafka C library
+*
+* Copyright (c) 2016 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.
+*/
+
+/**
+ * System portability
+ */
+
+#include "rd.h"
+
+
+#include <stdlib.h>
+
+/**
+ * qsort_r substitute
+ * This nicely explains why we wont bother with the native implementation
+ * on Win32 (qsort_s), OSX/FreeBSD (qsort_r with diff args):
+ * http://forum.theorex.tech/t/different-declarations-of-qsort-r-on-mac-and-linux/93/2
+ */
+static RD_TLS int (*rd_qsort_r_cmp) (const void *, const void *, void *);
+static RD_TLS void *rd_qsort_r_arg;
+
+static RD_UNUSED
+int rd_qsort_r_trampoline (const void *a, const void *b) {
+ return rd_qsort_r_cmp(a, b, rd_qsort_r_arg);
+}
+
+void rd_qsort_r (void *base, size_t nmemb, size_t size,
+ int (*compar)(const void *, const void *, void *),
+ void *arg) {
+ rd_qsort_r_cmp = compar;
+ rd_qsort_r_arg = arg;
+ qsort(base, nmemb, size, rd_qsort_r_trampoline);
+ rd_qsort_r_cmp = NULL;
+ rd_qsort_r_arg = NULL;
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/9f66960e/thirdparty/librdkafka-0.11.1/src/rdports.h
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.1/src/rdports.h b/thirdparty/librdkafka-0.11.1/src/rdports.h
new file mode 100644
index 0000000..44cef55
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.1/src/rdports.h
@@ -0,0 +1,33 @@
+/*
+* librdkafka - Apache Kafka C library
+*
+* Copyright (c) 2016 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.
+*/
+#pragma once
+
+
+void rd_qsort_r (void *base, size_t nmemb, size_t size,
+ int (*compar)(const void *, const void *, void *),
+ void *arg);