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
103
 
# File 'lib/backburner/allq_wrapper.rb', line 93

def initialize(url = 'localhost:8090')
  puts "URL = #{url}"
  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 



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/backburner/allq_wrapper.rb', line 181

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 



198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/backburner/allq_wrapper.rb', line 198

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 



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

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

#clear(tube) ⇒ Object 



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

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

#close ⇒ Object 



166
167
168
169
 
# File 'lib/backburner/allq_wrapper.rb', line 166

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

#delete(job) ⇒ Object 



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

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

#done(job) ⇒ Object 



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

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

#get(tube_name = 'default') ⇒ Object 



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/backburner/allq_wrapper.rb', line 150

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 



171
172
173
174
175
176
177
178
179
 
# File 'lib/backburner/allq_wrapper.rb', line 171

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 



142
143
144
145
146
147
148
 
# File 'lib/backburner/allq_wrapper.rb', line 142

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 



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
246
247
248
249
250
 
# File 'lib/backburner/allq_wrapper.rb', line 218

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 



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

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

#set_children_started(parent_job_id) ⇒ Object 



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

def set_children_started(parent_job_id)
  @client.set_children_started_put(parent_job_id)
end

#stats ⇒ Object 



252
253
254
255
 
# File 'lib/backburner/allq_wrapper.rb', line 252

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

#touch(job) ⇒ Object 



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

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

#tube_names ⇒ Object 



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

def tube_names
  stats_hash = stats
  stats_hash.keys
end

#tubes ⇒ Object 



138
139
140
 
# File 'lib/backburner/allq_wrapper.rb', line 138

def tubes
  tube_names
end