1 files changed, 124 insertions, 0 deletions
diff --git a/fluent-bit/lib/librdkafka-2.1.0/src-cpp/TopicImpl.cpp b/fluent-bit/lib/librdkafka-2.1.0/src-cpp/TopicImpl.cpp
new file mode 100644
index 000000000..bf9734df9
--- /dev/null
+++ b/fluent-bit/lib/librdkafka-2.1.0/src-cpp/TopicImpl.cpp
@@ -0,0 +1,124 @@
+/*
+ * librdkafka - Apache Kafka C/C++ library
+ *
+ * Copyright (c) 2014 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 <iostream>
+#include <string>
+#include <list>
+#include <cerrno>
+
+#include "rdkafkacpp_int.h"
+
+const int32_t RdKafka::Topic::PARTITION_UA = RD_KAFKA_PARTITION_UA;
+
+const int64_t RdKafka::Topic::OFFSET_BEGINNING = RD_KAFKA_OFFSET_BEGINNING;
+
+const int64_t RdKafka::Topic::OFFSET_END = RD_KAFKA_OFFSET_END;
+
+const int64_t RdKafka::Topic::OFFSET_STORED = RD_KAFKA_OFFSET_STORED;
+
+const int64_t RdKafka::Topic::OFFSET_INVALID = RD_KAFKA_OFFSET_INVALID;
+
+RdKafka::Topic::~Topic() {
+}
+
+static int32_t partitioner_cb_trampoline(const rd_kafka_topic_t *rkt,
+                                         const void *keydata,
+                                         size_t keylen,
+                                         int32_t partition_cnt,
+                                         void *rkt_opaque,
+                                         void *msg_opaque) {
+  RdKafka::TopicImpl *topicimpl = static_cast<RdKafka::TopicImpl *>(rkt_opaque);
+  std::string key(static_cast<const char *>(keydata), keylen);
+  return topicimpl->partitioner_cb_->partitioner_cb(topicimpl, &key,
+                                                    partition_cnt, msg_opaque);
+}
+
+static int32_t partitioner_kp_cb_trampoline(const rd_kafka_topic_t *rkt,
+                                            const void *keydata,
+                                            size_t keylen,
+                                            int32_t partition_cnt,
+                                            void *rkt_opaque,
+                                            void *msg_opaque) {
+  RdKafka::TopicImpl *topicimpl = static_cast<RdKafka::TopicImpl *>(rkt_opaque);
+  return topicimpl->partitioner_kp_cb_->partitioner_cb(
+      topicimpl, keydata, keylen, partition_cnt, msg_opaque);
+}
+
+
+
+RdKafka::Topic *RdKafka::Topic::create(Handle *base,
+                                       const std::string &topic_str,
+                                       const Conf *conf,
+                                       std::string &errstr) {
+  const RdKafka::ConfImpl *confimpl =
+      static_cast<const RdKafka::ConfImpl *>(conf);
+  rd_kafka_topic_t *rkt;
+  rd_kafka_topic_conf_t *rkt_conf;
+  rd_kafka_t *rk = dynamic_cast<HandleImpl *>(base)->rk_;
+
+  RdKafka::TopicImpl *topic = new RdKafka::TopicImpl();
+
+  if (!confimpl) {
+    /* Reuse default topic config, but we need our own copy to
+     * set the topic opaque. */
+    rkt_conf = rd_kafka_default_topic_conf_dup(rk);
+  } else {
+    /* Make a copy of conf struct to allow Conf reuse. */
+    rkt_conf = rd_kafka_topic_conf_dup(confimpl->rkt_conf_);
+  }
+
+  /* Set topic opaque to the topic so that we can reach our topic object
+   * from whatever callbacks get registered.
+   * The application itself will not need these opaques since their
+   * callbacks are class based. */
+  rd_kafka_topic_conf_set_opaque(rkt_conf, static_cast<void *>(topic));
+
+  if (confimpl) {
+    if (confimpl->partitioner_cb_) {
+      rd_kafka_topic_conf_set_partitioner_cb(rkt_conf,
+                                             partitioner_cb_trampoline);
+      topic->partitioner_cb_ = confimpl->partitioner_cb_;
+    } else if (confimpl->partitioner_kp_cb_) {
+      rd_kafka_topic_conf_set_partitioner_cb(rkt_conf,
+                                             partitioner_kp_cb_trampoline);
+      topic->partitioner_kp_cb_ = confimpl->partitioner_kp_cb_;
+    }
+  }
+
+
+  if (!(rkt = rd_kafka_topic_new(rk, topic_str.c_str(), rkt_conf))) {
+    errstr = rd_kafka_err2str(rd_kafka_last_error());
+    delete topic;
+    rd_kafka_topic_conf_destroy(rkt_conf);
+    return NULL;
+  }
+
+  topic->rkt_ = rkt;
+
+  return topic;
+}