| /* |
| * 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, |
| Conf *conf, |
| std::string &errstr) { |
| RdKafka::ConfImpl *confimpl = static_cast<RdKafka::ConfImpl *>(conf); |
| rd_kafka_topic_t *rkt; |
| rd_kafka_topic_conf_t *rkt_conf; |
| |
| RdKafka::TopicImpl *topic = new RdKafka::TopicImpl(); |
| |
| if (!confimpl) |
| rkt_conf = rd_kafka_topic_conf_new(); |
| 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(dynamic_cast<HandleImpl*>(base)->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; |
| |
| } |
| |