Sha256: 70d960387ccfb643da95711b9e94a24e0de67c95aa44d263cc0e5805258fcb68
Contents?: true
Size: 1.56 KB
Versions: 12
Compression:
Stored size: 1.56 KB
Contents
module CanvasSync
module Jobs
class ForkGather < CanvasSync::Job
def perform(job_chain, options)
forked_job = self.class.forked_at_job(job_chain)
if forked_job.present?
forked_job.with_lock do
forked_job.fork_count -= 1
forked_job.save!
end
if forked_job.fork_count <= 0
(job_chain[:global_options][:fork_keys] || []).pop&.each do |k|
job_chain[:global_options].delete(k.to_sym)
end
CanvasSync.invoke_next(job_chain)
end
else
CanvasSync.invoke_next(job_chain)
end
end
def self.handle_branch_error(e, job_chain:, skip_invoke: false, **kwargs)
return nil unless job_chain&.[](:global_options)&.[](:fork_path).present?
duped_chain = CanvasSync.duplicate_chain(job_chain)
job_list = duped_chain[:jobs]
while job_list.count > 0
job_class_name = job_list[0][:job]
job_class = job_class_name.constantize
break if job_class <= CanvasSync::Jobs::ForkGather
job_list.shift
end
return nil unless job_list.present?
if skip_invoke
duped_chain
else
CanvasSync.invoke_next(duped_chain)
true
end
end
protected
def self.forked_at_job(job_chain)
fork_item = (job_chain[:global_options][:fork_path] || []).pop
if fork_item.present?
CanvasSync::JobLog.find_by(job_id: fork_item)
else
nil
end
end
end
end
end
Version data entries
12 entries across 12 versions & 1 rubygems