intial polling version

Author: accelerated

Diff for: include/cppkafka/consumer.h

@@ -35,7 +35,7 @@
 #include <chrono>
 #include <functional>
 #include "kafka_handle_base.h"
-#include "message.h"
+#include "queue.h"
 #include "macros.h"
 #include "error.h"
 
@@ -54,7 +54,7 @@ class TopicConfiguration;
  * Semi-simple code showing how to use this class
  *
  * \code
- * // Create a configuration and set the group.id and broker list fields 
+ * // 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" }
@@ -74,13 +74,13 @@ class TopicConfiguration;
  * consumer.set_revocation_callback([&](const TopicPartitionList& topic_partitions) {
  *     cout << topic_partitions.size() << " partitions revoked!" << endl;
  * });
- * 
- * // Subscribe 
+ *
+ * // 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(); 
+ *     Message msg = consumer.poll();
  *     if (msg) {
  *         if (!msg.get_error()) {
  *             // It's an actual message. Get the payload and print it to stdout
@@ -103,12 +103,12 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
 
     /**
      * \brief Creates an instance of a consumer.
-     * 
-     * Note that the configuration *must contain* the group.id attribute set or this 
+     *
+     * Note that the configuration *must contain* the group.id attribute set or this
      * will throw.
      *
      * \param config The configuration to be used
-     */ 
+     */
     Consumer(Configuration config);
     Consumer(const Consumer&) = delete;
     Consumer(Consumer&&) = delete;
@@ -124,7 +124,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
 
     /**
      * \brief Sets the topic/partition assignment callback
-     * 
+     *
      * The Consumer class will use rd_kafka_conf_set_rebalance_cb and will handle the
      * rebalance, converting from rdkafka topic partition list handles into TopicPartitionList
      * and executing the assignment/revocation/rebalance_error callbacks.
@@ -138,7 +138,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
 
     /**
      * \brief Sets the topic/partition revocation callback
-     * 
+     *
      * The Consumer class will use rd_kafka_conf_set_rebalance_cb and will handle the
      * rebalance, converting from rdkafka topic partition list handles into TopicPartitionList
      * and executing the assignment/revocation/rebalance_error callbacks.
@@ -153,7 +153,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
 
     /**
      * \brief Sets the rebalance error callback
-     * 
+     *
      * The Consumer class will use rd_kafka_conf_set_rebalance_cb and will handle the
      * rebalance, converting from rdkafka topic partition list handles into TopicPartitionList
      * and executing the assignment/revocation/rebalance_error callbacks.
@@ -188,9 +188,9 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
     /**
      * \brief Unassigns the current topic/partition assignment
      *
-     * This translates into a call to rd_kafka_assign using a null as the topic partition list 
+     * This translates into a call to rd_kafka_assign using a null as the topic partition list
      * parameter
-     */ 
+     */
     void unassign();
 
     /**
@@ -252,7 +252,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
      *
      * This translates into a call to rd_kafka_get_watermark_offsets
      *
-     * \param topic_partition The topic/partition to get the offsets from 
+     * \param topic_partition The topic/partition to get the offsets from
      */
     OffsetTuple get_offsets(const TopicPartition& topic_partition) const;
 
@@ -316,16 +316,16 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
      * This will call rd_kafka_consumer_poll.
      *
      * Note that you need to call poll periodically as a keep alive mechanism, otherwise the broker
-     * will think this consumer is down and will trigger a rebalance (if using dynamic 
+     * will think this consumer is down and will trigger a rebalance (if using dynamic
      * subscription).
      *
      * The timeout used on this call will be the one configured via Consumer::set_timeout.
      *
      * The returned message *might* be empty. If's necessary to check that it's a valid one before
      * using it:
-     * 
+     *
      * \code
-     * Message msg = consumer.poll(); 
+     * Message msg = consumer.poll();
      * if (msg) {
      *     // It's a valid message!
      * }
@@ -361,6 +361,38 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
      * \param timeout The timeout for this operation
      */
     std::vector<Message> poll_batch(size_t max_batch_size, std::chrono::milliseconds timeout);
+    
+    /**
+     * \brief Get the global event queue servicing this consumer corresponding to
+     *        rd_kafka_queue_get_main and which is polled via rd_kafka_poll
+     *
+     * \return A Queue object
+     *
+     * \remark Note that this call will disable forwarding to the consumer_queue.
+     *         To restore forwarding (if desired) call Queue::forward_to_queue(consumer_queue)
+     */
+    Queue get_main_queue() const;
+    
+    /**
+     * \brief Get the consumer group queue servicing corresponding to
+     *        rd_kafka_queue_get_consumer and which is polled via rd_kafka_consumer_poll
+     *
+     * \return A Queue object
+     */
+    Queue get_consumer_queue() const;
+    
+    /**
+     * \brief Get the queue belonging to this partition. If the consumer is not assigned to this
+     *        partition, an empty queue will be returned
+     *
+     * \param partition The partition object
+     *
+     * \return A Queue object
+     *
+     * \remark Note that this call will disable forwarding to the consumer_queue.
+     *         To restore forwarding (if desired) call Queue::forward_to_queue(consumer_queue)
+     */
+    Queue get_partition_queue(const TopicPartition& partition) const;
 private:
     static void rebalance_proxy(rd_kafka_t *handle, rd_kafka_resp_err_t error,
                                 rd_kafka_topic_partition_list_t *partitions, void *opaque);

Diff for: include/cppkafka/cppkafka.h

@@ -46,6 +46,7 @@
 #include <cppkafka/message_builder.h>
 #include <cppkafka/metadata.h>
 #include <cppkafka/producer.h>
+#include <cppkafka/queue.h>
 #include <cppkafka/topic.h>
 #include <cppkafka/topic_configuration.h>
 #include <cppkafka/topic_partition.h>
@@ -55,5 +56,6 @@
 #include <cppkafka/utils/buffered_producer.h>
 #include <cppkafka/utils/compacted_topic_processor.h>
 #include <cppkafka/utils/consumer_dispatcher.h>
+#include <cppkafka/utils/roundrobin_poll_adapter.h>
 
 #endif

Diff for: include/cppkafka/exceptions.h

@@ -122,6 +122,18 @@ class CPPKAFKA_API ConsumerException : public Exception {
     Error error_;
 };
 
+/**
+ * Queue exception for rd_kafka_queue_t errors
+ */
+class CPPKAFKA_API QueueException : public Exception {
+public:
+    QueueException(Error error);
+
+    Error get_error() const;
+private:
+    Error error_;
+};
+
 } // cppkafka
 
 #endif // CPPKAFKA_EXCEPTIONS_H

Diff for: include/cppkafka/message.h

@@ -177,6 +177,8 @@ class CPPKAFKA_API Message {
     Buffer key_;
 };
 
+using MessageList = std::vector<Message>;
+
 /**
  * Represents a message's timestamp
  */

Diff for: include/cppkafka/queue.h

@@ -0,0 +1,183 @@
+/*
+ * Copyright (c) 2017, Matias Fontanini
+ * 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.
+ *
+ */
+
+#include <vector>
+#include <memory>
+#include <boost/optional.hpp>
+#include <librdkafka/rdkafka.h>
+#include "macros.h"
+#include "message.h"
+
+#ifndef CPPKAFKA_QUEUE_H
+#define CPPKAFKA_QUEUE_H
+
+namespace cppkafka {
+/**
+ * \brief Represents a rdkafka queue
+ *
+ * This is a simple wrapper over a rd_kafka_queue_t*
+ */
+class CPPKAFKA_API Queue {
+public:
+    /**
+     * \brief Creates a Queue object that doesn't take ownership of the handle
+     *
+     * \param handle The handle to be used
+     */
+    static Queue make_non_owning(rd_kafka_queue_t* handle);
+    
+    /**
+     * \brief Constructs an empty queue
+     *
+     * Note that using any methods except Queue::get_handle on an empty queue is undefined
+     * behavior
+     */
+    Queue();
+
+    /**
+     * \brief Constructs a queue using a handle
+     *
+     * This will take ownership of the handle
+     *
+     * \param handle The handle to be used
+     */
+    Queue(rd_kafka_queue_t* handle);
+
+    /**
+     * Returns the rdkakfa handle
+     */
+    rd_kafka_queue_t* get_handle() const;
+    
+    /**
+     * \brief Returns the length of the queue
+     *
+     * This translates to a call to rd_kafka_queue_length
+     */
+    size_t get_length() const;
+    
+    /**
+     * \brief Forward to another queue
+     *
+     * This translates to a call to rd_kafka_queue_forward
+     */
+    void forward_to_queue(const Queue& forward_queue) const;
+    
+    /**
+     * \brief Disable forwarding to another queue
+     *
+     * This translates to a call to rd_kafka_queue_forward(NULL)
+     */
+    void disable_queue_forwarding() const;
+    
+    /**
+     * \brief Sets the timeout for consume operations
+     *
+     * This timeout is applied when calling consume()
+     *
+     * \param timeout The timeout to be set
+     */
+    void set_consume_timeout(std::chrono::milliseconds timeout);
+    
+    /**
+     * Gets the configured timeout.
+     *
+     * \sa Queue::set_timeout
+     */
+    std::chrono::milliseconds get_consume_timeout() const;
+    
+    /**
+     * \brief Consume a message from this queue
+     *
+     * This translates to a call to rd_kafka_consume_queue using the configured timeout for this object
+     *
+     * \return A message
+     */
+    Message consume() const;
+
+    /**
+     * \brief Consume a message from this queue
+     *
+     * Same as consume() but the specified timeout will be used instead of the configured one
+     *
+     * \param timeout The timeout to be used on this call
+     *
+     * \return A message
+     */
+    Message consume(std::chrono::milliseconds timeout) const;
+    
+    /**
+     * \brief Consumes a batch of messages from this queue
+     *
+     * This translates to a call to rd_kafka_consume_batch_queue using the configured timeout for this object
+     *
+     * \param max_batch_size The max number of messages to consume if available
+     *
+     * \return A list of messages. Could be empty if there's nothing to consume
+     */
+    MessageList consume_batch(size_t max_batch_size) const;
+    
+    /**
+     * \brief Consumes a batch of messages from this queue
+     *
+     * Same as Queue::consume_batch(size_t) but the specified timeout will be used instead of the configured one
+     *
+     * \param max_batch_size The max number of messages to consume if available
+     *
+     * \param timeout The timeout to be used on this call
+     *
+     * \return A list of messages. Could be empty if there's nothing to consume
+     */
+    MessageList consume_batch(size_t max_batch_size, std::chrono::milliseconds timeout) const;
+    
+    /**
+     * Indicates whether this queue is valid (not null)
+     */
+    explicit operator bool() const {
+        return handle_ != nullptr;
+    }
+    
+private:
+    static const std::chrono::milliseconds DEFAULT_TIMEOUT;
+    
+    using HandlePtr = std::unique_ptr<rd_kafka_queue_t, decltype(&rd_kafka_queue_destroy)>;
+
+    struct NonOwningTag { };
+
+    Queue(rd_kafka_queue_t* handle, NonOwningTag);
+
+    // Members
+    HandlePtr                   handle_;
+    std::chrono::milliseconds   timeout_ms_;
+};
+
+using QueueList = std::vector<Queue>;
+
+} // cppkafka
+
+#endif //CPPKAFKA_QUEUE_H