intial polling version

Author: accelerated

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"
 
@@ -97,6 +97,15 @@ class TopicConfiguration;
  */
 class CPPKAFKA_API Consumer : public KafkaHandleBase {
 public:
+    /**
+     * \brief Specifies what type of consumption strategy this consumer will use
+     */
+    enum PollStrategy {
+        POLL_COMMON_QUEUE = 0,       // all messages and events are forwarded to a global queue (default)
+                                     // this results in batch consumption of assigned partitions
+        POLL_PARTITION_QUEUES_RR = 1 // poll assigned partitions in round-robin fashion
+                                     // messages are no longer forwarded to a global queue.
+    };
     using AssignmentCallback = std::function<void(TopicPartitionList&)>;
     using RevocationCallback = std::function<void(const TopicPartitionList&)>;
     using RebalanceErrorCallback = std::function<void(Error)>;
@@ -110,6 +119,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
      * \param config The configuration to be used
      */ 
     Consumer(Configuration config);
+    Consumer(Configuration config, PollStrategy strategy);
     Consumer(const Consumer&) = delete;
     Consumer(Consumer&&) = delete;
     Consumer& operator=(const Consumer&) = delete;
@@ -311,9 +321,13 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
     const RebalanceErrorCallback& get_rebalance_error_callback() const;
 
     /**
-     * \brief Polls for new messages
+     * \brief Polls for new messages depending on the configured PollStrategy
      *
-     * This will call rd_kafka_consumer_poll.
+     * If PollStrategy is set to POLL_COMMON_QUEUE (default) this function calls
+     * Queue::consume() on the main consumer queue. If PollStrategy is set to POLL_PARTITION_QUEUES_RR
+     * each invocation of this function will poll one of the assigned partitions in
+     * round-robin fashion by calling its respective Queue::consume() function.
+     * Regardless of the PollStrategy used, this function will also call rd_kafka_poll to serve other events.
      *
      * 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 
@@ -344,23 +358,29 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
     Message poll(std::chrono::milliseconds timeout);
 
     /**
-     * \brief Polls for a batch of messages
+     * \brief Polls for a batch of messages depending on the configured PollStrategy
      *
-     * This can return one or more messages
+     * This can return one or more messages. If PollStrategy is set to POLL_COMMON_QUEUE (default) this function
+     * calls Queue::consume_batch() on the main consumer queue. If PollStrategy is set to POLL_PARTITION_QUEUES_RR
+     * each invocation of this function will poll one of the assigned partitions in round-robin fashion by
+     * calling its respective Queue::consume_batch() function.
+     * Regardless of the PollStrategy used, this function will also call rd_kafka_poll to serve other events.
      *
      * \param max_batch_size The maximum amount of messages expected
      */
-    std::vector<Message> poll_batch(size_t max_batch_size);
+    MessageList poll_batch(size_t max_batch_size);
 
     /**
-     * \brief Polls for a batch of messages
+     * \brief Polls for a batch of messages depending on the configured PollStrategy
      *
-     * This can return one or more messages
+     * Same as the other overload of Consumer::poll but the provided timeout will be used
+     * instead of the one configured on this Consumer.
      *
      * \param max_batch_size The maximum amount of messages expected
      * \param timeout The timeout for this operation
      */
-    std::vector<Message> poll_batch(size_t max_batch_size, std::chrono::milliseconds timeout);
+    MessageList poll_batch(size_t max_batch_size, std::chrono::milliseconds timeout);
+    
 private:
     static void rebalance_proxy(rd_kafka_t *handle, rd_kafka_resp_err_t error,
                                 rd_kafka_topic_partition_list_t *partitions, void *opaque);
@@ -370,9 +390,13 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
     void commit(const TopicPartitionList* topic_partitions, bool async);
     void handle_rebalance(rd_kafka_resp_err_t err, TopicPartitionList& topic_partitions);
 
-    AssignmentCallback assignment_callback_;
-    RevocationCallback revocation_callback_;
-    RebalanceErrorCallback rebalance_error_callback_;
+    AssignmentCallback      assignment_callback_;
+    RevocationCallback      revocation_callback_;
+    RebalanceErrorCallback  rebalance_error_callback_;
+    PollStrategy            poll_strategy_;
+    Queue                   group_queue_;
+    QueueList               partition_queues_;
+    size_t                  queue_index_;
 };
 
 } // cppkafka

include/cppkafka/cppkafka.h

@@ -45,6 +45,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>

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

include/cppkafka/message.h

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

include/cppkafka/queue.h

@@ -0,0 +1,176 @@
+/*
+ * 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 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

src/CMakeLists.txt

@@ -4,6 +4,7 @@ set(SOURCES
     configuration_option.cpp
     exceptions.cpp
     topic.cpp
+    queue.cpp
     buffer.cpp
     message.cpp
     topic_partition.cpp