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
|
# File 'lib/massive-import/planner_job.rb', line 10
def perform(args)
import_id = args['import_id']
return unless import_id
import = Import.find_by(id: import_id, status: 'RUNNING')
return unless import
token = SecureRandom.hex
lock_acquired = acquire_planner_lock(import, token)
return unless lock_acquired
if import.total_records == 0
import.update_columns(status: 'COMPLETED')
return
end
recover_timedout_batches(import)
batch_stats = Batch.where(import_id: import.id, attempt: import.attempt).group(:status).count
pending_batches = batch_stats.fetch('PENDING', 0)
running_batches = batch_stats.fetch('RUNNING', 0)
batches_created = 0
if pending_batches * 2 <= import.max_batch_concurrency
batches_created = create_batches(import, token)
return if batches_created < 0
if batches_created == 0 && pending_batches == 0 && running_batches == 0
import.reload
done = handle_attempt_completion(import, token)
return if done
end
end
import.reload
dispatch_workers(import, import.max_batch_concurrency, token)
enqueue_self(import)
ensure
release_planner_lock(import, token) if lock_acquired
end
|