Class: TxghQueue::Backends::Sqs::Consumer

Inherits:
Object
  • Object
show all
Defined in:
lib/txgh-queue/backends/sqs/consumer.rb

Constant Summary collapse

RECEIVE_PARAMS = 
{ message_attribute_names: %w(history_sequence) }

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queues, logger) ⇒ Consumer

Returns a new instance of Consumer.



12
13
14
15
 
# File 'lib/txgh-queue/backends/sqs/consumer.rb', line 12

def initialize(queues, logger)
  @queues = queues
  @logger = logger
end

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



10
11
12
 
# File 'lib/txgh-queue/backends/sqs/consumer.rb', line 10

def logger
  @logger
end

#queues ⇒ Object (readonly)

Returns the value of attribute queues.



10
11
12
 
# File 'lib/txgh-queue/backends/sqs/consumer.rb', line 10

def queues
  @queues
end

Instance Method Details

#work ⇒ Object 



17
18
19
20
21
22
23
24
25
26
 
# File 'lib/txgh-queue/backends/sqs/consumer.rb', line 17

def work
  queues.each do |queue|
    queue.receive_message(RECEIVE_PARAMS).messages.each do |message|
      logger.info("Received message from #{queue.name}, id: #{message.message_id}")
      Job.new(message, queue, logger).complete
    end
  end
rescue => e
  Txgh.events.publish_error!(e)
end