Class: Backburner::AllQWrapper

Inherits:
Object
  • Object
show all
Defined in:
lib/backburner/allq_wrapper.rb

Constant Summary collapse

DEFAULT_TIMEOUT = 
17_800

Instance Method Summary collapse

Constructor Details

#initialize(url = 'localhost:8090') ⇒ AllQWrapper

Returns a new instance of AllQWrapper.



93
94
95
96
97
98
99
100
101
102
 
# File 'lib/backburner/allq_wrapper.rb', line 93

def initialize(url = 'localhost:8090')
  allq_conf = Allq::Configuration.new do |config|
    config.host = url
  end

  raw_client = Allq::ApiClient.new(allq_conf)
  @client = Allq::ActionsApi.new(raw_client)
  @admin = Allq::AdminApi.new(raw_client)
  @recent_times = []
end

Instance Method Details

#build_new_job(body, options) ⇒ Object 



176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/backburner/allq_wrapper.rb', line 176

def build_new_job(body, options)
  adjusted_priority = map_priority(options[:pri] || 5)

  ttl = options[:ttl] || options[:ttr] || DEFAULT_TIMEOUT
  tube_name = options[:tube_name] || 'default'
  delay = options[:delay] || 0
  parent_id = options[:parent_id]

  Allq::NewJob.new(tube: tube_name,
                   body: Base64.strict_encode64(body),
                   ttl: ttl,
                   delay: delay,
                   priority: adjusted_priority,
                   shard_key: options[:shard_key],
                   parent_id: parent_id)
end

#build_new_parent_job(body, options) ⇒ Object 



193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/backburner/allq_wrapper.rb', line 193

def build_new_parent_job(body, options)
  adjusted_priority = map_priority(options[:pri] || 5)
  ttl = options[:ttl] || options[:ttr] || DEFAULT_TIMEOUT
  tube_name = options[:tube_name] || 'default'
  delay = options[:delay] || 0
  limit = options[:limit]
  timeout = options[:timeout] || 3_600
  run_on_timeout = options[:run_on_timeout] || false

  Allq::NewParentJob.new(tube: tube_name,
                         body: Base64.strict_encode64(body),
                         ttl: ttl,
                         delay: delay,
                         priority: adjusted_priority,
                         timeout: timeout,
                         run_on_timeout: run_on_timeout,
                         shard_key: options[:shard_key],
                         limit: limit)
end

#bury(job) ⇒ Object 



124
125
126
 
# File 'lib/backburner/allq_wrapper.rb', line 124

def bury(job)
  @client.bury_put(job.id)
end

#clear(tube) ⇒ Object 



104
105
106
 
# File 'lib/backburner/allq_wrapper.rb', line 104

def clear(tube)
   @client.tube_delete(tube)
end

#close ⇒ Object 



161
162
163
164
 
# File 'lib/backburner/allq_wrapper.rb', line 161

def close
rescue StandardError => e
  puts(e)
end

#delete(job) ⇒ Object 



116
117
118
 
# File 'lib/backburner/allq_wrapper.rb', line 116

def delete(job)
  @client.job_delete(job.id)
end

#done(job) ⇒ Object 



112
113
114
 
# File 'lib/backburner/allq_wrapper.rb', line 112

def done(job)
  @client.job_delete(job.id)
end

#get(tube_name = 'default') ⇒ Object 



145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/backburner/allq_wrapper.rb', line 145

def get(tube_name = 'default')
  job = nil
  job = @client.job_get(tube_name)

  # Inplace decode
  job.body = Base64.decode64(job.body) if job&.body

  Backburner::AllQJob.new(self, job)
rescue StandardError => e
  if e.message == "Couldn't resolve host name"
    puts('COUDNT RESOLVE HOST NAME------ SHOULD REBOOT')
  else
    puts(e)
  end
end

#map_priority(app_priority) ⇒ Object 



166
167
168
169
170
171
172
173
174
 
# File 'lib/backburner/allq_wrapper.rb', line 166

def map_priority(app_priority)
  app_priority = app_priority.to_i

  # IF already using allq-like priority, stick with it
  return app_priority if app_priority < 11 && app_priority > 0

  # return app_priority unless larger than 10
  app_priority > 10 ? 5 : app_priority
end

#peek_buried(tube_name = 'default') ⇒ Object 



137
138
139
140
141
142
143
 
# File 'lib/backburner/allq_wrapper.rb', line 137

def peek_buried(tube_name = 'default')
  job = @client.peek_get(tube_name, buried: true)
  return nil if job.body.nil?

  job.body = Base64.decode64(job.body) if job
  Backburner::AllQJob.new(self, job)
end

#put(body, options) ⇒ Object 



213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/backburner/allq_wrapper.rb', line 213

def put(body, options)
  # New school put
  retry_count = 0
  is_parent = options[:is_parent] || false
  result = nil

  begin
    Timeout.timeout(10) do
      if is_parent
        new_job = build_new_parent_job(body, options)
        result = @client.parent_job_post(new_job)
      else
        new_job = build_new_job(body, options)
        result = @client.job_post(new_job)
      end
      raise 'PUT returned nil' if result.nil? || result.to_s == ''
    end
  rescue Timeout::Error
    puts('ALLQ PUT timeout, retrying...')
    sleep(5)
    retry_count += 1
    retry if retry_count < 4
    raise "Failed to put on allq, we are investigating the problem, please try again -> #{body}"
  rescue StandardError => e
    puts('Failed to ALLQ PUT, retrying...')
    puts(e)
    retry_count += 1
    sleep(5)
    retry if retry_count < 4
    raise "Failed to put on allq, we are investigating the problem, please try again: #{body}"
  end
  result
end

#release(job, delay = 0) ⇒ Object 



120
121
122
 
# File 'lib/backburner/allq_wrapper.rb', line 120

def release(job, delay = 0)
  @client.release_put(job.id, delay: delay)
end

#stats ⇒ Object 



247
248
249
250
 
# File 'lib/backburner/allq_wrapper.rb', line 247

def stats(tube)
  final_stats = stats
  final_stats[tube]
end

#touch(job) ⇒ Object 



108
109
110
 
# File 'lib/backburner/allq_wrapper.rb', line 108

def touch(job)
  @client.touch_put(job.id)
end

#tube_names ⇒ Object 



128
129
130
131
 
# File 'lib/backburner/allq_wrapper.rb', line 128

def tube_names
  stats_hash = stats
  stats_hash.keys
end

#tubes ⇒ Object 



133
134
135
 
# File 'lib/backburner/allq_wrapper.rb', line 133

def tubes
  tube_names
end