Add documentation for producer, move classes into Producer

Author: thijsc

.gitignore

@@ -4,3 +4,4 @@ ext/tmp
 ext/librdkafka.*
 *.gem
 .yardoc
+doc

.yardopts

@@ -0,0 +1 @@
+--no-private

lib/rdkafka/ffi.rb

@@ -2,6 +2,7 @@
 require "logger"
 
 module Rdkafka
+  # @private
   module FFI
     extend ::FFI::Library
 
@@ -144,7 +145,7 @@ class TopicPartitionList < ::FFI::Struct
       :void, [:pointer, :pointer, :pointer]
     ) do |client_ptr, message_ptr, opaque_ptr|
       message = Message.new(message_ptr)
-      delivery_handle = Rdkafka::DeliveryHandle.new(message[:_private])
+      delivery_handle = Rdkafka::Producer::DeliveryHandle.new(message[:_private])
       delivery_handle[:pending] = false
       delivery_handle[:response] = message[:err]
       delivery_handle[:partition] = message[:partition]

lib/rdkafka/producer.rb

@@ -1,5 +1,6 @@
 module Rdkafka
   class Producer
+    # @private
     def initialize(native_kafka)
       @closing = false
       @native_kafka = native_kafka
@@ -16,13 +17,28 @@ def initialize(native_kafka)
       @polling_thread.abort_on_exception = true
     end
 
+    # Close this producer and wait for the internal poll queue to empty.
     def close
       # Indicate to polling thread that we're closing
       @closing = true
       # Wait for the polling thread to finish up
       @polling_thread.join
     end
 
+    # Produces a message to a Kafka topic. The message is added to rdkafka's queue, call wait on the returned delivery handle to make sure it is delivered.
+    #
+    # When no partition is specified the underlying Kafka library picks a partition based on the key. If no key is specified, a random partition will be used.
+    # When a timestamp is provided this is used instead of the autogenerated timestamp.
+    #
+    # @param topic [String] The topic to produce to
+    # @param payload [String] The message's payload
+    # @param key [String] The message's key
+    # @param partition [Integer] Optional partition to produce to
+    # @param timestamp [Integer] Optional timestamp of this message
+    #
+    # @raise [RdkafkaError] When adding the message to rdkafka's queue failed
+    #
+    # @return [DeliveryHandle] Delivery handle that can be used to wait for the result of producing this message
     def produce(topic:, payload: nil, key: nil, partition: nil, timestamp: nil)
       # Start by checking and converting the input
 
@@ -73,49 +89,74 @@ def produce(topic:, payload: nil, key: nil, partition: nil, timestamp: nil)
 
       delivery_handle
     end
-  end
-
-  class WaitTimeoutError < RuntimeError; end
-
-  class DeliveryHandle < ::FFI::Struct
-    layout :pending, :bool,
-           :response, :int,
-           :partition, :int,
-           :offset, :int64
 
-    def pending?
-      self[:pending]
-    end
+    # Error that is raised when waiting for a delivery handle to complete
+    # takes longer than the specified timeout.
+    class WaitTimeoutError < RuntimeError; end
+
+    # Handle to wait for a delivery report which is returned when
+    # producing a message.
+    class DeliveryHandle < ::FFI::Struct
+      layout :pending, :bool,
+             :response, :int,
+             :partition, :int,
+             :offset, :int64
+
+      # Whether the delivery handle is still pending.
+      #
+      # @return [Boolean]
+      def pending?
+        self[:pending]
+      end
 
-    # Wait for the delivery report
-    def wait(timeout_in_seconds=60)
-      timeout = if timeout_in_seconds
-                  Time.now.to_i + timeout_in_seconds
-                else
-                  nil
-                end
-      loop do
-        if pending?
-          if timeout && timeout <= Time.now.to_i
-            raise WaitTimeoutError.new("Waiting for delivery timed out after #{timeout_in_seconds} seconds")
+      # Wait for the delivery report or raise an error if this takes longer than the timeout.
+      # If there is a timeout this does not mean the message is not delivered, rdkafka might still be working on delivering the message.
+      # In this case it is possible to call wait again.
+      #
+      # @param timeout_in_seconds [Integer] Number of seconds to wait before timing out. If this is nil it does not time out.
+      #
+      # @raise [RdkafkaError] When delivering the message failed
+      # @raise [WaitTimeoutError] When the timeout has been reached and the handle is still pending
+      #
+      # @return [DeliveryReport]
+      def wait(timeout_in_seconds=60)
+        timeout = if timeout_in_seconds
+                    Time.now.to_i + timeout_in_seconds
+                  else
+                    nil
+                  end
+        loop do
+          if pending?
+            if timeout && timeout <= Time.now.to_i
+              raise WaitTimeoutError.new("Waiting for delivery timed out after #{timeout_in_seconds} seconds")
+            end
+            sleep 0.1
+            next
+          elsif self[:response] != 0
+            raise RdkafkaError.new(self[:response])
+          else
+            return DeliveryReport.new(self[:partition], self[:offset])
           end
-          sleep 0.1
-          next
-        elsif self[:response] != 0
-          raise RdkafkaError.new(self[:response])
-        else
-          return DeliveryReport.new(self[:partition], self[:offset])
         end
       end
     end
-  end
 
-  class DeliveryReport
-    attr_reader :partition, :offset
+    # Delivery report for a succesfully produced message.
+    class DeliveryReport
+      # The partition this message was produced to.
+      # @return [Integer]
+      attr_reader :partition
+
+      # The offset of the produced message.
+      # @return [Integer]
+      attr_reader :offset
 
-    def initialize(partition, offset)
-      @partition = partition
-      @offset = offset
+      private
+
+      def initialize(partition, offset)
+        @partition = partition
+        @offset = offset
+      end
     end
   end
 end

spec/rdkafka/producer_spec.rb

@@ -102,6 +102,6 @@
     )
     expect {
       handle.wait(0)
-    }.to raise_error Rdkafka::WaitTimeoutError
+    }.to raise_error Rdkafka::Producer::WaitTimeoutError
   end
 end