Class: Eventboss::UnitOfWork

Inherits:
Object
  • Object
show all
Includes:
Logging, SafeThread
Defined in:
lib/eventboss/unit_of_work.rb

Overview

UnitOfWork handles calls a listener for each message and deletes on success

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from SafeThread

#handle_exception, #safe_thread

Methods included from Logging

#logger

Constructor Details

#initialize(client, queue, listener, message) ⇒ UnitOfWork

Returns a new instance of UnitOfWork.



9
10
11
12
13
14
15
 
# File 'lib/eventboss/unit_of_work.rb', line 9

def initialize(client, queue, listener, message)
  @client = client
  @queue = queue
  @listener = listener
  @message = message
  @logger = logger
end

Instance Attribute Details

#listener ⇒ Object

Returns the value of attribute listener.



7
8
9
 
# File 'lib/eventboss/unit_of_work.rb', line 7

def listener
  @listener
end

#message ⇒ Object

Returns the value of attribute message.



7
8
9
 
# File 'lib/eventboss/unit_of_work.rb', line 7

def message
  @message
end

#queue ⇒ Object

Returns the value of attribute queue.



7
8
9
 
# File 'lib/eventboss/unit_of_work.rb', line 7

def queue
  @queue
end

Instance Method Details

#change_message_visibility(postponed_by) ⇒ Object 



30
31
32
33
34
35
36
 
# File 'lib/eventboss/unit_of_work.rb', line 30

def change_message_visibility(postponed_by)
  @client.change_message_visibility(
    queue_url: @queue.url,
    receipt_handle: @message.receipt_handle,
    visibility_timeout: postponed_by
  )
end

#cleanup ⇒ Object 



38
39
40
41
42
43
 
# File 'lib/eventboss/unit_of_work.rb', line 38

def cleanup
  @client.delete_message(
    queue_url: @queue.url, receipt_handle: @message.receipt_handle
  )
  logger.debug(@message.message_id) { 'Deleting' }
end

#run ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/eventboss/unit_of_work.rb', line 17

def run
  logger.debug(@message.message_id) { 'Started' }
  processor = @listener.new
  processor.receive(JSON.parse(@message.body))
  logger.debug(@message.message_id) { 'Finished' }
rescue StandardError => exception
  handle_exception(exception, processor: processor, message_id: @message.message_id)
else
  cleanup unless processor.postponed_by
ensure
  change_message_visibility(processor.postponed_by) if processor.postponed_by
end