Class: Dwf::Item

Inherits:
Object
  • Object
show all
Defined in:
lib/dwf/item.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Item

Returns a new instance of Item.



11
12
13
 
# File 'lib/dwf/item.rb', line 11

def initialize(options = {})
  assign_attributes(options)
end

Instance Attribute Details

#callback_type ⇒ Object (readonly)

Returns the value of attribute callback_type.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def callback_type
  @callback_type
end

#enqueued_at ⇒ Object (readonly)

Returns the value of attribute enqueued_at.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def enqueued_at
  @enqueued_at
end

#failed_at ⇒ Object (readonly)

Returns the value of attribute failed_at.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def failed_at
  @failed_at
end

#finished_at ⇒ Object (readonly)

Returns the value of attribute finished_at.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def finished_at
  @finished_at
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def id
  @id
end

#incoming ⇒ Object

Returns the value of attribute incoming.



9
10
11
 
# File 'lib/dwf/item.rb', line 9

def incoming
  @incoming
end

#klass ⇒ Object (readonly)

Returns the value of attribute klass.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def klass
  @klass
end

#outgoing ⇒ Object

Returns the value of attribute outgoing.



9
10
11
 
# File 'lib/dwf/item.rb', line 9

def outgoing
  @outgoing
end

#output_payload ⇒ Object (readonly)

Returns the value of attribute output_payload.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def output_payload
  @output_payload
end

#params ⇒ Object (readonly)

Returns the value of attribute params.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def params
  @params
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def queue
  @queue
end

#started_at ⇒ Object (readonly)

Returns the value of attribute started_at.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def started_at
  @started_at
end

#workflow_id ⇒ Object (readonly)

Returns the value of attribute workflow_id.



7
8
9
 
# File 'lib/dwf/item.rb', line 7

def workflow_id
  @workflow_id
end

Class Method Details

.from_hash(hash) ⇒ Object 



15
16
17
 
# File 'lib/dwf/item.rb', line 15

def self.from_hash(hash)
  Module.const_get(hash[:klass]).new(hash)
end

Instance Method Details

#as_json ⇒ Object 



159
160
161
 
# File 'lib/dwf/item.rb', line 159

def as_json
  to_hash.to_json
end

#cb_build_in? ⇒ Boolean

Returns:

  • (Boolean) 


27
28
29
 
# File 'lib/dwf/item.rb', line 27

def cb_build_in?
  callback_type == Dwf::Workflow::BUILD_IN
end

#current_timestamp ⇒ Object 



128
129
130
 
# File 'lib/dwf/item.rb', line 128

def current_timestamp
  Time.now.to_i
end

#enqueue! ⇒ Object 



70
71
72
73
74
75
 
# File 'lib/dwf/item.rb', line 70

def enqueue!
  @enqueued_at = current_timestamp
  @started_at = nil
  @finished_at = nil
  @failed_at = nil
end

#enqueue_outgoing_jobs ⇒ Object 



132
133
134
135
136
137
138
139
 
# File 'lib/dwf/item.rb', line 132

def enqueue_outgoing_jobs
  outgoing.each do |job_name|
    client.check_or_lock(workflow_id, job_name)
    out = client.find_job(workflow_id, job_name)
    out.persist_and_perform_async! if out.ready_to_start?
    client.release_lock(workflow_id, job_name)
  end
end

#enqueued? ⇒ Boolean

Returns:

  • (Boolean) 


100
101
102
 
# File 'lib/dwf/item.rb', line 100

def enqueued?
  !enqueued_at.nil?
end

#fail! ⇒ Object 



96
97
98
 
# File 'lib/dwf/item.rb', line 96

def fail!
  @finished_at = @failed_at = current_timestamp
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


108
109
110
 
# File 'lib/dwf/item.rb', line 108

def failed?
  !failed_at.nil?
end

#finish! ⇒ Object 



92
93
94
 
# File 'lib/dwf/item.rb', line 92

def finish!
  @finished_at = current_timestamp
end

#finished? ⇒ Boolean

Returns:

  • (Boolean) 


104
105
106
 
# File 'lib/dwf/item.rb', line 104

def finished?
  !finished_at.nil?
end

#mark_as_finished ⇒ Object 



82
83
84
85
 
# File 'lib/dwf/item.rb', line 82

def mark_as_finished
  finish!
  persist!
end

#mark_as_started ⇒ Object 



77
78
79
80
 
# File 'lib/dwf/item.rb', line 77

def mark_as_started
  start!
  persist!
end

#name ⇒ Object 



41
42
43
 
# File 'lib/dwf/item.rb', line 41

def name
  @name ||= "#{klass}|#{id}"
end

#no_dependencies? ⇒ Boolean

Returns:

  • (Boolean) 


49
50
51
 
# File 'lib/dwf/item.rb', line 49

def no_dependencies?
  incoming.empty?
end

#output(data) ⇒ Object 



45
46
47
 
# File 'lib/dwf/item.rb', line 45

def output(data)
  @output_payload = data
end

#parents_succeeded? ⇒ Boolean

Returns:

  • (Boolean) 


53
54
55
56
57
 
# File 'lib/dwf/item.rb', line 53

def parents_succeeded?
  incoming.all? do |name|
    client.find_job(workflow_id, name).succeeded?
  end
end

#payloads ⇒ Object 



59
60
61
62
63
64
65
66
67
68
 
# File 'lib/dwf/item.rb', line 59

def payloads
  incoming.map do |job_name|
    job = client.find_job(workflow_id, job_name)
    {
      id: job.name,
      class: job.klass.to_s,
      output: job.output_payload
    }
  end
end

#perform ⇒ Object 



25
 
# File 'lib/dwf/item.rb', line 25

def perform; end

#perform_async ⇒ Object 



36
37
38
39
 
# File 'lib/dwf/item.rb', line 36

def perform_async
  Dwf::Worker.set(queue: queue || client.config.namespace)
             .perform_async(workflow_id, name)
end

#persist! ⇒ Object 



163
164
165
 
# File 'lib/dwf/item.rb', line 163

def persist!
  client.persist_job(self)
end

#persist_and_perform_async! ⇒ Object 



19
20
21
22
23
 
# File 'lib/dwf/item.rb', line 19

def persist_and_perform_async!
  enqueue!
  persist!
  perform_async
end

#ready_to_start? ⇒ Boolean

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/dwf/item.rb', line 124

def ready_to_start?
  !running? && !enqueued? && !finished? && !failed? && parents_succeeded?
end

#reload ⇒ Object 



31
32
33
34
 
# File 'lib/dwf/item.rb', line 31

def reload
  item = client.find_job(workflow_id, name)
  assign_attributes(item.to_hash)
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


120
121
122
 
# File 'lib/dwf/item.rb', line 120

def running?
  started? && !finished?
end

#start! ⇒ Object 



87
88
89
90
 
# File 'lib/dwf/item.rb', line 87

def start!
  @started_at = current_timestamp
  @failed_at = nil
end

#started? ⇒ Boolean

Returns:

  • (Boolean) 


116
117
118
 
# File 'lib/dwf/item.rb', line 116

def started?
  !started_at.nil?
end

#succeeded? ⇒ Boolean

Returns:

  • (Boolean) 


112
113
114
 
# File 'lib/dwf/item.rb', line 112

def succeeded?
  finished? && !failed?
end

#to_hash ⇒ Object 



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/dwf/item.rb', line 141

def to_hash
  {
    id: id,
    klass: klass.to_s,
    queue: queue,
    incoming: incoming,
    outgoing: outgoing,
    finished_at: finished_at,
    enqueued_at: enqueued_at,
    started_at: started_at,
    failed_at: failed_at,
    params: params,
    workflow_id: workflow_id,
    callback_type: callback_type,
    output_payload: output_payload
  }
end