Class: Abid::State

Inherits:
Object
  • Object
show all
Extended by:
Forwardable, MonitorMixin
Defined in:
lib/abid/state.rb

Constant Summary collapse

RUNNING = 
1
SUCCESSED = 
2
FAILED = 
3
STATES = 
constants.map { |c| [const_get(c), c] }.to_h

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(task) ⇒ State

Returns a new instance of State.



60
61
62
63
64
65
 
# File 'lib/abid/state.rb', line 60

def initialize(task)
  @task = task
  @record = nil
  @started = false
  reload
end

Class Method Details

.deserialize(bytes) ⇒ Object 



53
54
55
 
# File 'lib/abid/state.rb', line 53

def deserialize(bytes)
  YAML.load(bytes)
end

.find(task) ⇒ Object 



16
17
18
 
# File 'lib/abid/state.rb', line 16

def find(task)
  synchronize { @cache[task.object_id] ||= new(task) }
end

.list(pattern: nil, started_before: nil, started_after: nil) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/abid/state.rb', line 20

def list(pattern: nil, started_before: nil, started_after: nil)
  dataset = Rake.application.database[:states]

  dataset = dataset.where { start_time < started_before } if started_before
  dataset = dataset.where { start_time > started_after } if started_after
  dataset = dataset.order(:start_time)

  dataset.map do |record|
    next if pattern && record[:name] !~ pattern
    {
      id: record[:id],
      name: record[:name],
      params: deserialize(record[:params]),
      state: STATES[record[:state]],
      start_time: record[:start_time],
      end_time: record[:end_time]
    }
  end.compact
end

.revoke(id) ⇒ Object 



40
41
42
43
44
45
46
47
 
# File 'lib/abid/state.rb', line 40

def revoke(id)
  db = Rake.application.database
  db.transaction do
    running = db[:states].where(id: id, state: RUNNING).count > 0
    fail 'task is now running' if running
    db[:states].where(id: id).delete
  end
end

.serialize(params) ⇒ Object 



49
50
51
 
# File 'lib/abid/state.rb', line 49

def serialize(params)
  YAML.dump(params)
end

Instance Method Details

#assume ⇒ Object 



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/abid/state.rb', line 120

def assume
  fail 'cannot revoke volatile task' if disabled?

  database.transaction do
    reload
    fail 'task is now running' if running?

    new_state = {
      state: SUCCESSED,
      start_time: Time.now,
      end_time: Time.now
    }

    if @record
      dataset.where(id: @record[:id]).update(new_state)
      @record = @record.merge(new_state)
    else
      id = dataset.insert(
        digest: digest,
        name: @task.name,
        params: serialize(@task.params),
        **new_state
      )
      @record = { id: id, **new_state }
    end
  end
end

#database ⇒ Object 



67
68
69
 
# File 'lib/abid/state.rb', line 67

def database
  Rake.application.database
end

#dataset ⇒ Object 



71
72
73
 
# File 'lib/abid/state.rb', line 71

def dataset
  database[:states]
end

#digest ⇒ Object 



189
190
191
 
# File 'lib/abid/state.rb', line 189

def digest
  Digest::MD5.hexdigest(@task.name + "\n" + serialize(@task.params))
end

#disabled? ⇒ Boolean

Returns:

  • (Boolean) 


75
76
77
 
# File 'lib/abid/state.rb', line 75

def disabled?
  @task.volatile? || Rake.application.options.disable_state
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


112
113
114
 
# File 'lib/abid/state.rb', line 112

def failed?
  state == FAILED
end

#finish(error = nil) ⇒ Object 



180
181
182
183
184
185
186
187
 
# File 'lib/abid/state.rb', line 180

def finish(error = nil)
  return if disabled? || preview?
  return unless @record
  return unless @started
  state = error ? FAILED : SUCCESSED
  dataset.where(id: @record[:id]).update(state: state, end_time: Time.now)
  reload
end

#id ⇒ Object 



96
97
98
 
# File 'lib/abid/state.rb', line 96

def id
  @record[:id] if @record
end

#not_found? ⇒ Boolean

Returns:

  • (Boolean) 


116
117
118
 
# File 'lib/abid/state.rb', line 116

def not_found?
  !disabled? && @record.nil?
end

#preview? ⇒ Boolean

Returns:

  • (Boolean) 


79
80
81
 
# File 'lib/abid/state.rb', line 79

def preview?
  Rake.application.options.dryrun || Rake.application.options.preview
end

#reload ⇒ Object 



83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/abid/state.rb', line 83

def reload
  return if disabled?

  if @record
    id = @record[:id]
    @record = dataset.where(id: id).first
  else
    @record = dataset.where(digest: digest).to_a.find do |r|
      [@task.name, @task.params].eql? [r[:name], deserialize(r[:params])]
    end
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


104
105
106
 
# File 'lib/abid/state.rb', line 104

def running?
  state == RUNNING
end

#start ⇒ Object 



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/abid/state.rb', line 148

def start
  return true if disabled? || preview?

  database.transaction do
    reload

    fail AbidErrorTaskAlreadyRunning if running?

    new_state = {
      state: RUNNING,
      start_time: Time.now,
      end_time: nil
    }

    if @record
      dataset.where(id: @record[:id]).update(new_state)
      @record = @record.merge(new_state)
    else
      id = dataset.insert(
        digest: digest,
        name: @task.name,
        params: serialize(@task.params),
        **new_state
      )
      @record = { id: id, **new_state }
    end

    @started = true
    true
  end
end

#state ⇒ Object 



100
101
102
 
# File 'lib/abid/state.rb', line 100

def state
  @record[:state] if @record
end

#successed? ⇒ Boolean

Returns:

  • (Boolean) 


108
109
110
 
# File 'lib/abid/state.rb', line 108

def successed?
  state == SUCCESSED
end