Class: ChiliLogger::AWS::SqsBroker

Inherits:
Object
  • Object
show all
Defined in:
lib/brokers/sqs_broker.rb

Overview

class that handles errors when message broker can’t be reached, etc…

Instance Method Summary collapse

Constructor Details

#initialize(custom_config = {}) ⇒ SqsBroker

Returns a new instance of SqsBroker.



9
10
11
12
13
14
15
16
17
18
19
20
21
 
# File 'lib/brokers/sqs_broker.rb', line 9

def initialize(custom_config = {})
  custom_config ||= {}

  @default_secrets = Values::Secrets.new.get_secrets_collection('ChiliLoggerSQSCredentials')
  @default = ChiliLogger::Values::Default.new
  @queue_name = custom_config[:queue_name] || @default_secrets['queue_name']

  @sqs_config = sqs_config(custom_config)
  @sqs = Aws::SQS::Client.new(@sqs_config)

  ENV['AWS_REGION'] = @sqs_config[:region] # Aws::SQS::QueuePoller requires region as an env variable
  @poller = Aws::SQS::QueuePoller.new(queue_url(@queue_name), skip_delete: true)
end

Instance Method Details

#delete_message(msg) ⇒ Object 



49
50
51
 
# File 'lib/brokers/sqs_broker.rb', line 49

def delete_message(msg)
  @poller.delete_messages([msg])
end

#fetch_messages ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/brokers/sqs_broker.rb', line 32

def fetch_messages
  return if ChiliLogger.instance.deactivated

  remaining_msgs = num_of_msgs_in_queue
  messages = []

  @poller.poll do |sqs_msg|
    log = JSON.parse(sqs_msg.body)['body']['log']
    messages << { log: log, fallback_broker_msg: sqs_msg }

    remaining_msgs -= 1
    throw :stop_polling if remaining_msgs <= 0 # prevents infinite loop
  end

  messages
end

#publish(message = @default.sqs_message) ⇒ Object 



23
24
25
26
27
28
29
30
 
# File 'lib/brokers/sqs_broker.rb', line 23

def publish(message = @default.sqs_message)
  return if ChiliLogger.instance.deactivated

  message_body = { body: message }
  queue_message = queue_message(message_body)

  @sqs.send_message(queue_message)
end