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) 


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

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
  
  Signal.trap("INT") do
    @terminated = true
  end
end

Instance Attribute Details

#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 



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/yakc/reader.rb', line 16

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