Class: EventQ::Amazon::QueueWorker

Inherits:
Object
  • Object
show all
Includes:
WorkerId
Defined in:
lib/eventq_aws/aws_queue_worker.rb,
lib/eventq_aws/jruby/aws_queue_worker.rb

Constant Summary collapse

APPROXIMATE_RECEIVE_COUNT = 
'ApproximateReceiveCount'.freeze
MESSAGE = 
'Message'.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ QueueWorker

Returns a new instance of QueueWorker.



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 11

def initialize
  @threads = []
  @forks = []
  @is_running = false

  @on_retry_exceeded_block = nil
  @on_retry_block = nil
  @on_error_block = nil

  @hash_helper = HashKit::Helper.new
  @serialization_provider_manager = EventQ::SerializationProviders::Manager.new
  @signature_provider_manager = EventQ::SignatureProviders::Manager.new

  @last_gc_flush = Time.now
  @gc_flush_interval = 10

  @queue_poll_wait = 10
end

Instance Attribute Details

#is_running ⇒ Object

Returns the value of attribute is_running.



9
10
11
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 9

def is_running
  @is_running
end

Instance Method Details

#call_on_error_block(error:, message: nil) ⇒ Object 



163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 163

def call_on_error_block(error:, message: nil)
  if @on_error_block
    EventQ.logger.debug { "[#{self.class}] - Executing on_error block." }
    begin
      @on_error_block.call(error, message)
    rescue => e
      EventQ.logger.error("[#{self.class}] - An error occurred executing the on_error block. Error: #{e}")
    end
  else
    EventQ.logger.debug { "[#{self.class}] - No on_error block specified to execute." }
  end
end

#call_on_retry_block(message) ⇒ Object 



189
190
191
192
193
194
195
196
197
198
199
200
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 189

def call_on_retry_block(message)
  if @on_retry_block
    EventQ.logger.debug { "[#{self.class}] - Executing on_retry block." }
    begin
      @on_retry_block.call(message, abort)
    rescue => e
      EventQ.logger.error("[#{self.class}] - An error occurred executing the on_retry block. Error: #{e}")
    end
  else
    EventQ.logger.debug { "[#{self.class}] - No on_retry block specified." }
  end
end

#call_on_retry_exceeded_block(message) ⇒ Object 



176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 176

def call_on_retry_exceeded_block(message)
  if @on_retry_exceeded_block != nil
    EventQ.logger.debug { "[#{self.class}] - Executing on_retry_exceeded block." }
    begin
      @on_retry_exceeded_block.call(message)
    rescue => e
      EventQ.logger.error("[#{self.class}] - An error occurred executing the on_retry_exceeded block. Error: #{e}")
    end
  else
    EventQ.logger.debug { "[#{self.class}] - No on_retry_exceeded block specified." }
  end
end

#deserialize_message(payload) ⇒ Object 



227
228
229
230
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 227

def deserialize_message(payload)
  provider = @serialization_provider_manager.get_provider(EventQ::Configuration.serialization_provider)
  return provider.deserialize(payload)
end

#gc_flush ⇒ Object 



116
117
118
119
120
121
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 116

def gc_flush
  if Time.now - last_gc_flush > @gc_flush_interval
    GC.start
    @last_gc_flush = Time.now
  end
end

#last_gc_flush ⇒ Object 



123
124
125
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 123

def last_gc_flush
  @last_gc_flush
end

#on_error(&block) ⇒ Object 



218
219
220
221
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 218

def on_error(&block)
  @on_error_block = block
  return nil
end

#on_retry(&block) ⇒ Object 



213
214
215
216
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 213

def on_retry(&block)
  @on_retry_block = block
  return nil
end

#on_retry_exceeded(&block) ⇒ Object 



209
210
211
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 209

def on_retry_exceeded(&block)
  @retry_exceeded_block = block
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


223
224
225
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 223

def running?
  return @is_running
end

#serialize_message(msg) ⇒ Object 



232
233
234
235
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 232

def serialize_message(msg)
  provider = @serialization_provider_manager.get_provider(EventQ::Configuration.serialization_provider)
  return provider.serialize(msg)
end

#start(queue, options = {}, &block) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 30

def start(queue, options = {}, &block)

  EventQ.logger.info("[#{self.class}] - Preparing to start listening for messages.")

  configure(queue, options)

  if options[:client] == nil
    raise "[#{self.class}] - :client (QueueClient) must be specified."
  end

  raise "[#{self.class}] - Worker is already running." if running?

  client = options[:client]
  EventQ.logger.debug do
    "[#{self.class} #start] - Listening for messages on queue: #{queue.name}, Queue Url: #{client.get_queue_url(queue)}, Queue arn: #{client.get_queue_arn(queue)}"
  end

  EventQ.logger.info("[#{self.class}] - Listening for messages.")

  @forks = []

  if @fork_count > 1
    Thread.new do
      @fork_count.times do
        pid = fork do
          start_process(options, queue, block)
        end
        @forks.push(pid)
      end
      @forks.each { |pid| Process.wait(pid) }
    end
  else
    start_process(options, queue, block)
  end

  return true
end

#start_process(options, queue, block) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 68

def start_process(options, queue, block)

  %w'INT TERM'.each do |sig|
    Signal.trap(sig) {
      stop
      exit
    }
  end

  @is_running = true
  @threads = []

  #loop through each thread count
  @thread_count.times do
    thr = Thread.new do
      client = options[:client]
      manager = EventQ::Amazon::QueueManager.new({ client: client })

      #begin the queue loop for this thread
      while true do

        #check if the worker is still allowed to run and break out of thread loop if not
        if !@is_running
          break
        end

        has_message_received = thread_process_iteration(client, manager, queue, block)

        gc_flush

        if !has_message_received
          EventQ.logger.debug { "[#{self.class}] - No message received." }
          if @sleep > 0
            EventQ.logger.debug { "[#{self.class}] - Sleeping for #{@sleep} seconds" }
            sleep(@sleep)
          end
        end

      end
    end
    @threads.push(thr)
  end

  if (options.key?(:wait) && options[:wait] == true) || (options.key?(:fork_count) && options[:fork_count] > 1)
    @threads.each { |thr| thr.join }
  end
end

#stop ⇒ Object 



202
203
204
205
206
207
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 202

def stop
  EventQ.logger.info("[#{self.class}] - Stopping.")
  @is_running = false
  @threads.each { |thr| thr.join }
  return true
end

#thread_process_iteration(client, manager, queue, block) ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/eventq_aws/aws_queue_worker.rb', line 127

def thread_process_iteration(client, manager, queue, block)
  #get the queue
  q = manager.get_queue(queue)

  received = false

  begin

    # request a message from the queue
    response = client.sqs.receive_message({
                                              queue_url: q,
                                              max_number_of_messages: 1,
                                              wait_time_seconds: @queue_poll_wait,
                                              attribute_names: [APPROXIMATE_RECEIVE_COUNT]
                                          })

    #check that a message was received
    if response.messages.length > 0
      received = true
      begin
        tag_processing_thread
        process_message(response, client, queue, q, block)
      ensure
        untag_processing_thread
      end

    end

  rescue => e
    EventQ.log(:error, "[#{self.class}] - An unhandled error occurred. Error: #{e} | Backtrace: #{e.backtrace}")
    call_on_error_block(error: e)
  end

  return received
end