lib/rdkafka/consumer.rb
# Poll for the next message on one of the subscribed topics
#
# @param timeout_ms [Integer] Timeout of this poll
#
# @raise [RdkafkaError] When polling fails
#
# @return [Message, nil] A message or nil if there was no new message within the timeout
Poll for the next message on one of the subscribed topics
Poll for the next message on one of the subscribed topics
def poll(timeout_ms)
      message_ptr = Rdkafka::Bindings.rd_kafka_consumer_poll(@native_kafka, timeout_ms)
      if message_ptr.null?
        nil
      else
        native_message = Rdkafka::Bindings::Message.new(message_ptr)
        if native_message[:err] != 0
          raise Rdkafka::RdkafkaError.new(native_message[:err])
        end
        Rdkafka::Consumer::Message.new(native_message)
      end
    ensure
      if !message_ptr.nil? && !message_ptr.null?
        Rdkafka::Bindings.rd_kafka_message_destroy(message_ptr)
      end
    end
def poll ( timeout_ms ) message_ptr = Rdkafka :: Bindings . rd_kafka_consumer_poll ( @native_kafka , timeout_ms ) if message_ptr . null? nil else native_message = Rdkafka :: Bindings :: Message . new ( message_ptr ) if native_message [ :err ] != 0 raise Rdkafka :: RdkafkaError . new ( native_message [ :err ] ) end Rdkafka :: Consumer :: Message . new ( native_message ) end ensure if ! message_ptr . nil? && ! message_ptr . null? Rdkafka :: Bindings . rd_kafka_message_destroy ( message_ptr ) end end
Poll for the next message on one of the subscribed topics
Poll for the next message on one of the subscribed topics
# Create struct wrapper # Raise error if needed # Create a message to pass out # Clean up rdkafka message if there is one
A message or nil if there was no new message within the timeout
A message or nil if there was no new message within the timeout
Message, nil
[RdkafkaError] When polling fails
[ RdkafkaError ] When polling fails
timeout_ms
Timeout of this poll
Timeout of this poll