Class: CanvasSync::JobBatches::ChainBuilder

Inherits:
Object
  • Object
show all
Defined in:
lib/canvas_sync/job_batches/chain_builder.rb

Constant Summary collapse

VALID_PLACEMENT_PARAMETERS = 
%i[before after with].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(base_type = SerialBatchJob, chain_id: nil) ⇒ ChainBuilder

Returns a new instance of ChainBuilder.



7
8
9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 7

def initialize(base_type = SerialBatchJob, chain_id: nil)
  @chain_id = chain_id || SecureRandom.urlsafe_base64(10)

  if base_type.is_a?(Hash)
    @base_job = base_type
    @base_job[:args] ||= @base_job[:parameters] || []
    @base_job[:kwargs] ||= {}
  else
    @base_job = build_job_hash(base_type)
  end

  self.class.get_chain_parameter(base_job)
end

Instance Attribute Details

#base_job ⇒ Object (readonly)

Returns the value of attribute base_job.



5
6
7
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 5

def base_job
  @base_job
end

Class Method Details

._job_type_definitions ⇒ Object 



195
196
197
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 195

def _job_type_definitions
  @job_type_definitions ||= {}
end

.annotate_batch!(batch, chain) ⇒ Object 



251
252
253
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 251

def annotate_batch!(batch, chain)

end

.build(job, *args, **kwargs, &blk) ⇒ Object

Support builder syntaxt/DSL Chain.build(ConcurrentBatchJob) do

insert(SomeJob, arg1, kwarg: 1)
insert(SerialBatchJob) do
  insert(SomeJob, arg1, kwarg: 1)
end

end



187
188
189
190
191
192
193
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 187

def build(job, *args, **kwargs, &blk)
  new(job).tap do |ch|
    ch.base_job[:args] = args
    ch.base_job[:kwargs] = kwargs
    ch.apply_block(&blk)
  end
end

.enqueue_job(job_def) ⇒ Object

TODO: Add a Chain progress web View Augment Batch tree-view with Chain data

> [DONE] Tree view w/o Chain will only show Parent/Current batches and Job Counts
> If augmented with Chain data, the above will be annotated with Chain-related info and will be able to show Jobs defined in the Chain
  > Chain-jobs will be supplied chain_id and chain_step_id metadata
  > Using server-middleware, if a Chain-job (has chain_id and chain_step_id) creates a Batch, tag the Batch w/ the chain_id and chain_step_id
  > UI will map Batches to Chain-steps using the chain_step_id. UI will add entries for any Chain-steps that were not tied to a Batch
> [DONE] Use a Lua script to find child batch IDs. Support max_depth, items_per_depth, top_depth_slice parameters



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 228

def enqueue_job(job_def)
  job_class = job_def[:job].constantize
  job_args = job_def[:args] || job_def[:parameters] || []
  job_kwargs = job_def[:kwargs] || {}

  # Legacy Support
  if job_def[:options]
    job_args << {} unless job_args[-1].is_a?(Hash)
    job_args[-1].merge!(job_def[:options])
  end

  if job_class.respond_to? :perform_async
    job_class.perform_async(*job_args, **job_kwargs)
  else
    job_class.perform_later(*job_args, **job_kwargs)
  end
end

.get_chain_parameter(job_def, raise_error: true) ⇒ Object 



206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 206

def get_chain_parameter(job_def, raise_error: true)
  unless _job_type_definitions[job_def[:job].to_s].present?
    raise "Job Type #{job_def[:job].to_s} does not accept a sub-chain" if raise_error
    return nil
  end

  key = _job_type_definitions[job_def[:job].to_s][:chain_parameter]
  if key.is_a?(Numeric)
    job_def[:args][key] ||= []
  else
    job_def[:kwargs][key] ||= []
  end
end

.handle_step_complete(status, opts) ⇒ Object 



255
256
257
258
259
260
261
262
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 255

def handle_step_complete(status, opts)
  chain_link = opts[:chain_link]
  chain_id, chain_step_id = chain_link.split('-')

  CanvasSync::JobBatches::Batch.redis.multi do |r|
    r.hset("CHAIN-#{chain_id}-steps", chain_step_id, "complete")
  end
end

.link_to_batch!(chain_link, batch) ⇒ Object 



246
247
248
249
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 246

def link_to_batch!(chain_link, batch)
  # Or make chains a separate entity - Chains show batches, but batches don't show chain?
  # Or "Annotate" a Batch with chain data - could extract chain id from any job entry
end

.register_chain_job(job_class, chain_parameter, **options) ⇒ Object 



199
200
201
202
203
204
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 199

def register_chain_job(job_class, chain_parameter, **options)
  _job_type_definitions[job_class.to_s] = {
    **options,
    chain_parameter: chain_parameter,
  }
end

Instance Method Details

#<<(new_job) ⇒ Object 



40
41
42
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 40

def <<(new_job)
  insert_at(-1, new_job)
end

#[](key) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 26

def [](key)
  if key.is_a?(Class)
    get_sub_chain(key)
  else
    # Legacy Support
    key = :args if key == :parameters

    @base_job[key]
  end
end

#apply_block(&blk) ⇒ Object 



131
132
133
134
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 131

def apply_block(&blk)
  return unless blk.present?
  instance_exec(&blk)
end

#args ⇒ Object 



37
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 37

def args; return self[:args]; end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


104
105
106
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 104

def empty?
  self.class.get_chain_parameter(self).empty?
end

#get_sub_chain(sub_type) ⇒ Object 



108
109
110
111
112
113
114
115
116
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 108

def get_sub_chain(sub_type)
  matching_jobs = find_matching_jobs(sub_type).to_a
  raise "Found multiple \"#{sub_type}\" jobs in the chain" if matching_jobs.count > 1
  return nil if matching_jobs.count == 0

  job = matching_jobs[0][0]
  job = self.class.new(job, chain_id: @chain_id) unless job.is_a?(ChainBuilder)
  job
end

#insert(new_jobs, *args, **kwargs, &blk) ⇒ Object 



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
95
96
97
98
99
100
101
102
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 55

def insert(new_jobs, *args, **kwargs, &blk)
  if new_jobs.is_a?(Class) || new_jobs.is_a?(String)
    job_kwargs = kwargs.except(*VALID_PLACEMENT_PARAMETERS)
    new_jobs = build_job_hash(new_jobs, args: args, kwargs: job_kwargs, &blk)
    kwargs = kwargs.slice(*VALID_PLACEMENT_PARAMETERS)
  else
    invalid_params = kwargs.keys - VALID_PLACEMENT_PARAMETERS
    raise "Invalid placement parameters: #{invalid_params.map(&:to_s).join(', ')}" if invalid_params.present?
    raise "At most one placement parameter may be provided" if kwargs.values.compact.length > 1
    raise "Unexpected number of arguments" if args.length > 0
  end

  new_jobs = [new_jobs] unless new_jobs.is_a?(Array)

  if !kwargs.present?
    insert_at(-1, new_jobs)
  else
    placement = kwargs.keys[0]
    relative_to = kwargs.values[0]

    matching_jobs = find_matching_jobs(relative_to).to_a
    raise "Could not find a \"#{relative_to}\" job in the chain" if matching_jobs.count == 0
    raise "Found multiple \"#{relative_to}\" jobs in the chain" if matching_jobs.count > 1

    relative_job, parent_job, sub_index = matching_jobs[0]
    needed_parent_type = placement == :with ? ConcurrentBatchJob : SerialBatchJob

    chain = self.class.get_chain_parameter(parent_job)

    if parent_job[:job] != needed_parent_type
      old_job = chain[sub_index]
      parent_job = chain[sub_index] = {
        job: needed_parent_type,
        parameters: [],
      }
      sub_index = 0
      chain = self.class.get_chain_parameter(parent_job)
      chain << old_job
    end

    if placement == :with
      chain.insert(-1, *new_jobs)
    else
      sub_index += 1 if placement == :after
      chain.insert(sub_index, *new_jobs)
    end
  end
end

#insert_at(position, new_jobs, *args, **kwargs, &blk) ⇒ Object 



44
45
46
47
48
49
50
51
52
53
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 44

def insert_at(position, new_jobs, *args, **kwargs, &blk)
  chain = self.class.get_chain_parameter(base_job)
  if new_jobs.is_a?(Class) || new_jobs.is_a?(String)
    new_jobs = build_job_hash(new_jobs, args: args, kwargs: kwargs, &blk)
  elsif args.count > 0 || kwargs.count > 0
    raise "Unexpected number of arguments"
  end
  new_jobs = [new_jobs] unless new_jobs.is_a?(Array)
  chain.insert(position, *new_jobs)
end

#kwargs ⇒ Object 



38
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 38

def kwargs; return self[:kwargs]; end

#normalize!(job_def = self.base_job) ⇒ Object 



118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 118

def normalize!(job_def = self.base_job)
  if job_def.is_a?(ChainBuilder)
    job_def.normalize!
  else
    job_def[:job] = job_def[:job].to_s
    job_def[:chain_link] ||= "#{@chain_id}-#{SecureRandom.urlsafe_base64(10)}"
    if (chain = self.class.get_chain_parameter(job_def, raise_error: false)).present?
      chain.map! { |sub_job| normalize!(sub_job) }
    end
    job_def
  end
end

#process! ⇒ Object 



21
22
23
24
 
# File 'lib/canvas_sync/job_batches/chain_builder.rb', line 21

def process!
  normalize!
  self.class.enqueue_job(base_job)
end