Class: YAKC::Reader

Inherits:
Object
  • Object
show all
Defined in:
lib/yakc/reader.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(message_handler:) ⇒ Reader

Returns a new instance of Reader.

Raises:

  • (KeyError) 


6
7
8
9
10
11
12
13
14
15
16
17
 
# File 'lib/yakc/reader.rb', line 6

def initialize( message_handler: )
  @message_handler  = message_handler
  @config           = YAKC.configuration     

  raise KeyError, "YAKC::Reader initialized without a message handler. Please specify one so that your receives messages don't end up on the floor. For more info, go to: https://github.com/gaorlov/yakc#message-handler" unless message_handler
  
  %w(INT TERM).each do |signal|
    Signal.trap(signal) do
      @terminated = true
    end
  end
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



3
4
5
 
# File 'lib/yakc/reader.rb', line 3

def config
  @config
end

#message_handler ⇒ Object (readonly)

Returns the value of attribute message_handler.



3
4
5
 
# File 'lib/yakc/reader.rb', line 3

def message_handler
  @message_handler
end

#terminated ⇒ Object (readonly)

Returns the value of attribute terminated.



3
4
5
 
# File 'lib/yakc/reader.rb', line 3

def terminated
  @terminated
end

Instance Method Details

#read ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/yakc/reader.rb', line 19

def read
  logger.info "YAKC: Starting reading"
  loop do
    consumers.map do |consumer|
      consumer.fetch do |partition, bulk|
        bulk.each do |message|
          message_handler.handle consumer.topic, message
        end
      end
      return if terminated
    end
    return if terminated
  end
rescue => e
  logger.error e
  retry
end