| /* |
| * 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" |
| |
| RdKafka::Consumer::~Consumer () {} |
| |
| RdKafka::Consumer *RdKafka::Consumer::create (RdKafka::Conf *conf, |
| std::string &errstr) { |
| char errbuf[512]; |
| RdKafka::ConfImpl *confimpl = dynamic_cast<RdKafka::ConfImpl *>(conf); |
| RdKafka::ConsumerImpl *rkc = new RdKafka::ConsumerImpl(); |
| rd_kafka_conf_t *rk_conf = NULL; |
| |
| if (confimpl) { |
| if (!confimpl->rk_conf_) { |
| errstr = "Requires RdKafka::Conf::CONF_GLOBAL object"; |
| delete rkc; |
| return NULL; |
| } |
| |
| rkc->set_common_config(confimpl); |
| |
| rk_conf = rd_kafka_conf_dup(confimpl->rk_conf_); |
| } |
| |
| rd_kafka_t *rk; |
| if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, rk_conf, |
| errbuf, sizeof(errbuf)))) { |
| errstr = errbuf; |
| delete rkc; |
| return NULL; |
| } |
| |
| rkc->rk_ = rk; |
| |
| |
| return rkc; |
| } |
| |
| int64_t RdKafka::Consumer::OffsetTail (int64_t offset) { |
| return RD_KAFKA_OFFSET_TAIL(offset); |
| } |
| |
| RdKafka::ErrorCode RdKafka::ConsumerImpl::start (Topic *topic, |
| int32_t partition, |
| int64_t offset) { |
| RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); |
| |
| if (rd_kafka_consume_start(topicimpl->rkt_, partition, offset) == -1) |
| return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); |
| |
| return RdKafka::ERR_NO_ERROR; |
| } |
| |
| |
| RdKafka::ErrorCode RdKafka::ConsumerImpl::start (Topic *topic, |
| int32_t partition, |
| int64_t offset, |
| Queue *queue) { |
| RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); |
| RdKafka::QueueImpl *queueimpl = dynamic_cast<RdKafka::QueueImpl *>(queue); |
| |
| if (rd_kafka_consume_start_queue(topicimpl->rkt_, partition, offset, |
| queueimpl->queue_) == -1) |
| return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); |
| |
| return RdKafka::ERR_NO_ERROR; |
| } |
| |
| |
| RdKafka::ErrorCode RdKafka::ConsumerImpl::stop (Topic *topic, |
| int32_t partition) { |
| RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); |
| |
| if (rd_kafka_consume_stop(topicimpl->rkt_, partition) == -1) |
| return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); |
| |
| return RdKafka::ERR_NO_ERROR; |
| } |
| |
| RdKafka::ErrorCode RdKafka::ConsumerImpl::seek (Topic *topic, |
| int32_t partition, |
| int64_t offset, |
| int timeout_ms) { |
| RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); |
| |
| if (rd_kafka_seek(topicimpl->rkt_, partition, offset, timeout_ms) == -1) |
| return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); |
| |
| return RdKafka::ERR_NO_ERROR; |
| } |
| |
| RdKafka::Message *RdKafka::ConsumerImpl::consume (Topic *topic, |
| int32_t partition, |
| int timeout_ms) { |
| RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); |
| rd_kafka_message_t *rkmessage; |
| |
| rkmessage = rd_kafka_consume(topicimpl->rkt_, partition, timeout_ms); |
| if (!rkmessage) |
| return new RdKafka::MessageImpl(topic, |
| static_cast<RdKafka::ErrorCode> |
| (rd_kafka_last_error())); |
| |
| return new RdKafka::MessageImpl(topic, rkmessage); |
| } |
| |
| namespace { |
| /* Helper struct for `consume_callback'. |
| * Encapsulates the values we need in order to call `rd_kafka_consume_callback' |
| * and keep track of the C++ callback function and `opaque' value. |
| */ |
| struct ConsumerImplCallback { |
| ConsumerImplCallback(RdKafka::Topic* topic, RdKafka::ConsumeCb* cb, void* data) |
| : topic(topic), cb_cls(cb), cb_data(data) { |
| } |
| /* This function is the one we give to `rd_kafka_consume_callback', with |
| * the `opaque' pointer pointing to an instance of this struct, in which |
| * we can find the C++ callback and `cb_data'. |
| */ |
| static void consume_cb_trampoline(rd_kafka_message_t *msg, void *opaque) { |
| ConsumerImplCallback *instance = static_cast<ConsumerImplCallback*>(opaque); |
| RdKafka::MessageImpl message(instance->topic, msg, false /*don't free*/); |
| instance->cb_cls->consume_cb(message, instance->cb_data); |
| } |
| RdKafka::Topic *topic; |
| RdKafka::ConsumeCb *cb_cls; |
| void *cb_data; |
| }; |
| } |
| |
| int RdKafka::ConsumerImpl::consume_callback (RdKafka::Topic* topic, |
| int32_t partition, |
| int timeout_ms, |
| RdKafka::ConsumeCb *consume_cb, |
| void *opaque) { |
| RdKafka::TopicImpl *topicimpl = static_cast<RdKafka::TopicImpl *>(topic); |
| ConsumerImplCallback context(topic, consume_cb, opaque); |
| return rd_kafka_consume_callback(topicimpl->rkt_, partition, timeout_ms, |
| &ConsumerImplCallback::consume_cb_trampoline, &context); |
| } |
| |
| |
| RdKafka::Message *RdKafka::ConsumerImpl::consume (Queue *queue, |
| int timeout_ms) { |
| RdKafka::QueueImpl *queueimpl = dynamic_cast<RdKafka::QueueImpl *>(queue); |
| rd_kafka_message_t *rkmessage; |
| |
| rkmessage = rd_kafka_consume_queue(queueimpl->queue_, timeout_ms); |
| if (!rkmessage) |
| return new RdKafka::MessageImpl(NULL, |
| static_cast<RdKafka::ErrorCode> |
| (rd_kafka_last_error())); |
| /* |
| * Recover our Topic * from the topic conf's opaque field, which we |
| * set in RdKafka::Topic::create() for just this kind of situation. |
| */ |
| void *opaque = rd_kafka_topic_opaque(rkmessage->rkt); |
| Topic *topic = static_cast<Topic *>(opaque); |
| |
| return new RdKafka::MessageImpl(topic, rkmessage); |
| } |
| |
| namespace { |
| /* Helper struct for `consume_callback' with a Queue. |
| * Encapsulates the values we need in order to call `rd_kafka_consume_callback' |
| * and keep track of the C++ callback function and `opaque' value. |
| */ |
| struct ConsumerImplQueueCallback { |
| ConsumerImplQueueCallback(RdKafka::ConsumeCb *cb, void *data) |
| : cb_cls(cb), cb_data(data) { |
| } |
| /* This function is the one we give to `rd_kafka_consume_callback', with |
| * the `opaque' pointer pointing to an instance of this struct, in which |
| * we can find the C++ callback and `cb_data'. |
| */ |
| static void consume_cb_trampoline(rd_kafka_message_t *msg, void *opaque) { |
| ConsumerImplQueueCallback *instance = static_cast<ConsumerImplQueueCallback *>(opaque); |
| /* |
| * Recover our Topic * from the topic conf's opaque field, which we |
| * set in RdKafka::Topic::create() for just this kind of situation. |
| */ |
| void *topic_opaque = rd_kafka_topic_opaque(msg->rkt); |
| RdKafka::Topic *topic = static_cast<RdKafka::Topic *>(topic_opaque); |
| RdKafka::MessageImpl message(topic, msg, false /*don't free*/); |
| instance->cb_cls->consume_cb(message, instance->cb_data); |
| } |
| RdKafka::ConsumeCb *cb_cls; |
| void *cb_data; |
| }; |
| } |
| |
| int RdKafka::ConsumerImpl::consume_callback (Queue *queue, |
| int timeout_ms, |
| RdKafka::ConsumeCb *consume_cb, |
| void *opaque) { |
| RdKafka::QueueImpl *queueimpl = dynamic_cast<RdKafka::QueueImpl *>(queue); |
| ConsumerImplQueueCallback context(consume_cb, opaque); |
| return rd_kafka_consume_callback_queue(queueimpl->queue_, timeout_ms, |
| &ConsumerImplQueueCallback::consume_cb_trampoline, |
| &context); |
| } |
