-
Notifications
You must be signed in to change notification settings - Fork 417
/
scheduled_task.rb
94 lines (79 loc) · 1.97 KB
/
scheduled_task.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
67
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
require 'observer'
require 'concurrent/obligation'
require 'concurrent/runnable'
module Concurrent
class ScheduledTask
include Obligation
include Observable
include Runnable
attr_reader :schedule_time
def initialize(schedule_time, opts = {}, &block)
now = Time.now
if ! block_given?
raise ArgumentError.new('no block given')
elsif schedule_time.is_a?(Time)
if schedule_time <= now
raise ArgumentError.new('schedule time must be in the future')
else
@schedule_time = schedule_time.dup
end
elsif schedule_time.to_f <= 0.0
raise ArgumentError.new('seconds must be greater than zero')
else
@schedule_time = now + schedule_time.to_f
end
@state = :pending
@task = block
@schedule_time.freeze
set_deref_options(opts)
end
def cancelled?
return @state == :cancelled
end
def in_progress?
return @state == :in_progress
end
def cancel
return false if mutex.locked?
return mutex.synchronize do
if @state == :pending
@state = :cancelled
event.set
true
else
false
end
end
end
def add_observer(observer, func = :update)
return false unless @state == :pending || @state == :in_progress
super
end
protected
def on_task
while (diff = @schedule_time.to_f - Time.now.to_f) > 0
sleep( diff > 60 ? 60 : diff )
end
if @state == :pending
mutex.synchronize do
@state = :in_progress
begin
@value = @task.call
@state = :fulfilled
rescue => ex
@reason = ex
@state = :rejected
ensure
changed
end
end
end
if self.changed?
notify_observers(Time.now, self.value, @reason)
delete_observers
end
event.set
self.stop
end
end
end