<commit_before>/*
 * Copyright (c) 2017, <NAME>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 * * Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * * 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.
 *
 */

#ifndef CPP_KAFKA_CONSUMER_H
#define CPP_KAFKA_CONSUMER_H

#include <vector>
#include <string>
#include <chrono>
#include <functional>
#include "kafka_handle_base.h"
#include "queue.h"
#include "macros.h"
#include "error.h"
#include "detail/callback_invoker.h"

namespace cppkafka {

class TopicConfiguration;

/**
 * \brief High level kafka consumer class
 *
 * Wrapper for the high level consumer API provided by rdkafka. Most methods are just
 * a one to one mapping to rdkafka functions.
 *
 * This class allows hooking up to assignments/revocations via callbacks.
 *
 * Semi-simple code showing how to use this class
 *
 * \code
 * // Create a configuration and set the group.id and broker list fields
 * Configuration config = {
 *     { "metadata.broker.list", "127.0.0.1:9092" },
 *     { "group.id", "foo" }
 * };
 *
 * // Create a consumer
 * Consumer consumer(config);
 *
 * // Set the assignment callback
 * consumer.set_assignment_callback([&](TopicPartitionList& topic_partitions) {
 *     // Here you could fetch offsets and do something, altering the offsets on the
 *     // topic_partitions vector if needed
 *     cout << "Got assigned " << topic_partitions.size() << " partitions!" << endl;
 * });
 *
 * // Set the revocation callback
 * consumer.set_revocation_callback([&](const TopicPartitionList& topic_partitions) {
 *     cout << topic_partitions.size() << " partitions revoked!" << endl;
 * });
 *
 * // Subscribe
 * consumer.subscribe({ "my_topic" });
 * while (true) {
 *     // Poll. This will optionally return a message. It's necessary to check if it's a valid
 *     // one before using it
 *     Message msg = consumer.poll();
 *     if (msg) {
 *         if (!msg.get_error()) {
 *             // It's an actual message. Get the payload and print it to stdout
 *             cout << msg.get_payload().as_string() << endl;
 *         }
 *         else if (!msg.is_eof()) {
 *             // Is it an error notification, handle it.
 *             // This is explicitly skipping EOF notifications as they're not actually errors,
 *             // but that's how rdkafka provides them
 *         }
 *     }
 * }
 * \endcode
 */
class CPPKAFKA_API Consumer : public KafkaHandleBase {
public:
    using AssignmentCallback = std::function<void(TopicPartitionList&)>;
    using RevocationCallback = std::function<void(const TopicPartitionList&)>;
    using RebalanceErrorCallback = std::function<void(Error)>;
    using KafkaHandleBase::pause;

    /**
     * \brief Creates an instance of a consumer.
     *
     * Note that the configuration *must contain* the group.id attribute set or this<commit_msg>Bring base class  into scope<commit_after>/*
 * Copyright (c) 2017, <NAME>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 * * Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * * 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.
 *
 */

#ifndef CPP_KAFKA_CONSUMER_H
#define CPP_KAFKA_CONSUMER_H

#include <vector>
#include <string>
#include <chrono>
#include <functional>
#include "kafka_handle_base.h"
#include "queue.h"
#include "macros.h"
#include "error.h"
#include "detail/callback_invoker.h"

namespace cppkafka {

class TopicConfiguration;

/**
 * \brief High level kafka consumer class
 *
 * Wrapper for the high level consumer API provided by rdkafka. Most methods are just
 * a one to one mapping to rdkafka functions.
 *
 * This class allows hooking up to assignments/revocations via callbacks.
 *
 * Semi-simple code showing how to use this class
 *
 * \code
 * // Create a configuration and set the group.id and broker list fields
 * Configuration config = {
 *     { "metadata.broker.list", "127.0.0.1:9092" },
 *     { "group.id", "foo" }
 * };
 *
 * // Create a consumer
 * Consumer consumer(config);
 *
 * // Set the assignment callback
 * consumer.set_assignment_callback([&](TopicPartitionList& topic_partitions) {
 *     // Here you could fetch offsets and do something, altering the offsets on the
 *     // topic_partitions vector if needed
 *     cout << "Got assigned " << topic_partitions.size() << " partitions!" << endl;
 * });
 *
 * // Set the revocation callback
 * consumer.set_revocation_callback([&](const TopicPartitionList& topic_partitions) {
 *     cout << topic_partitions.size() << " partitions revoked!" << endl;
 * });
 *
 * // Subscribe
 * consumer.subscribe({ "my_topic" });
 * while (true) {
 *     // Poll. This will optionally return a message. It's necessary to check if it's a valid
 *     // one before using it
 *     Message msg = consumer.poll();
 *     if (msg) {
 *         if (!msg.get_error()) {
 *             // It's an actual message. Get the payload and print it to stdout
 *             cout << msg.get_payload().as_string() << endl;
 *         }
 *         else if (!msg.is_eof()) {
 *             // Is it an error notification, handle it.
 *             // This is explicitly skipping EOF notifications as they're not actually errors,
 *             // but that's how rdkafka provides them
 *         }
 *     }
 * }
 * \endcode
 */
class CPPKAFKA_API Consumer : public KafkaHandleBase {
public:
    using AssignmentCallback = std::function<void(TopicPartitionList&)>;
    using RevocationCallback = std::function<void(const TopicPartitionList&)>;
    using RebalanceErrorCallback = std::function<void(Error)>;
    using KafkaHandleBase::pause;
    using KafkaHandleBase::resume;

    /**
     * \brief Creates an instance of a consumer.
     *
     * Note that the configuration *must contain* the group.id attribute set or this
{'fraction_non_alphanumeric': 0.09631552917903066, 'fraction_numerical': 0.0034619188921859545, 'mean_word_length': 3.4105779716466738, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5771744', 'n_tokens_mistral': 2337, 'n_tokens_neox': 2166, 'n_words': 879}