/* * librdkafka - Apache Kafka C/C++ library * * Copyright (c) 2014-2022, 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 #include #include #include #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(rkt_opaque); std::string key(static_cast(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(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(conf); rd_kafka_topic_t *rkt; rd_kafka_topic_conf_t *rkt_conf; rd_kafka_t *rk = dynamic_cast(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(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; }