Class: Sidekiq::Postpone
- Inherits:
-
Object
- Object
- Sidekiq::Postpone
- Defined in:
- lib/sidekiq/postpone/version.rb,
lib/sidekiq/postpone/core_ext.rb,
lib/sidekiq/postpone.rb
Defined Under Namespace
Modules: CoreExt
Constant Summary collapse
- VERSION =
"0.3.1"
Class Method Summary collapse
Instance Method Summary collapse
- #all_jobs ⇒ Object
- #clear! ⇒ Object
- #empty? ⇒ Boolean
- #flush! ⇒ Object
-
#initialize(*client_args) ⇒ Postpone
constructor
A new instance of Postpone.
- #jids ⇒ Object
- #join!(other) ⇒ Object
- #push(payloads) ⇒ Object
- #wrap(join_parent: true, flush: true) ⇒ Object
Constructor Details
#initialize(*client_args) ⇒ Postpone
Returns a new instance of Postpone.
8 9 10 11 12 |
# File 'lib/sidekiq/postpone.rb', line 8 def initialize(*client_args) @client_args = client_args setup_queues setup_schedule end |
Class Method Details
.wrap(*client_args, **kwargs, &block) ⇒ Object
32 33 34 |
# File 'lib/sidekiq/postpone.rb', line 32 def self.wrap(*client_args, **kwargs, &block) new(*client_args).wrap(**kwargs, &block) end |
Instance Method Details
#all_jobs ⇒ Object
81 82 83 |
# File 'lib/sidekiq/postpone.rb', line 81 def all_jobs [*@queues.values.flatten(1), *@schedule] end |
#clear! ⇒ Object
45 46 47 48 |
# File 'lib/sidekiq/postpone.rb', line 45 def clear! @queues.clear @schedule.clear end |
#empty? ⇒ Boolean
77 78 79 |
# File 'lib/sidekiq/postpone.rb', line 77 def empty? @queues.empty? && @schedule.empty? end |
#flush! ⇒ Object
50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
# File 'lib/sidekiq/postpone.rb', line 50 def flush! current_postpone = Thread.current[:sidekiq_postpone] return if empty? Thread.current[:sidekiq_postpone] = nil # activate real raw_push client = Sidekiq::Client.new(*@client_args) @queues.each_value { |item| client.raw_push(item) } client.raw_push(@schedule) unless @schedule.empty? clear! ensure Thread.current[:sidekiq_postpone] = current_postpone end |
#jids ⇒ Object
85 86 87 |
# File 'lib/sidekiq/postpone.rb', line 85 def jids all_jobs.map! { |j| j['jid'] } end |
#join!(other) ⇒ Object
66 67 68 69 70 71 72 73 74 75 |
# File 'lib/sidekiq/postpone.rb', line 66 def join!(other) return if empty? @queues.each do |name, payloads| other.queues[name].concat(payloads) end other.schedule.concat(@schedule) clear! end |
#push(payloads) ⇒ Object
36 37 38 39 40 41 42 43 |
# File 'lib/sidekiq/postpone.rb', line 36 def push(payloads) if payloads.first['at'] @schedule.concat(payloads) else q = payloads.first['queue'] @queues[q].concat(payloads) end end |
#wrap(join_parent: true, flush: true) ⇒ Object
14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
# File 'lib/sidekiq/postpone.rb', line 14 def wrap(join_parent: true, flush: true) enter! begin yield self rescue clear! raise end.tap do if join_parent && (parent = Thread.current[:sidekiq_postpone_stack][-2]) join!(parent) elsif flush @flush_on_leave = true end end ensure leave! end |