Sha256: ac4ba3f21c51cbe17a7ef2dba66e336423b867cf1f520c569af9980f447cdb87

Contents?: true

Size: 550 Bytes

Versions: 2

Compression:

Stored size: 550 Bytes

Contents

require "java"
require "jruby-kafka/namespace"

java_import 'kafka.consumer.ConsumerIterator'
java_import 'kafka.consumer.KafkaStream'

class Kafka::Consumer
  include Java::JavaLang::Runnable
  java_signature 'void run()'

  @m_stream
  @m_threadNumber
  @m_queue

    def initialize(a_stream, a_threadNumber, a_queue)
      @m_threadNumber = a_threadNumber
      @m_stream = a_stream
      @m_queue = a_queue
    end

    def run
      it = @m_stream.iterator()
      while it.hasNext()
        @m_queue << it.next().message()
      end
    end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version Path
jruby-kafka-0.0.3 lib/jruby-kafka/consumer.rb
jruby-kafka-0.0.1 lib/jruby-kafka/consumer.rb