Class: GoodJob::ElasticsearchProcessQueue

Inherits:
HasUtils::GoodJob::BaseWorker
  • Object
show all
Includes:
HasUtils::GoodJob::UniqueJob
Defined in:
app/workers/good_job/elasticsearch_process_queue.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.on_success(batch, _context) ⇒ Object 



41
42
43
44
45
46
47
48
 
# File 'app/workers/good_job/elasticsearch_process_queue.rb', line 41

def self.on_success(batch, _context)
  index_class_name = batch.properties[:index_class_name]
  record_ids_count = batch.properties[:record_ids_count]
  key = batch.properties[:key]
  starting_time = batch.properties[:starting_time]
  ::Stat.save_to_stats(Stat.new, "::GoodJob::ElasticsearchProcessQueue", "elasticsearch", index_class_name, { record_ids_count: record_ids_count }, starting_time, ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)) if ::HasHelpers::Feature.active?(:elasticsearch_reindex_stat)
  ::HasHelpers::Index.after_index(key)
end

Instance Method Details

#perform(index_class_name, index_name) ⇒ Object 



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
 
# File 'app/workers/good_job/elasticsearch_process_queue.rb', line 11

def perform(index_class_name, index_name)
  starting_time = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
  index_klass = index_class_name.constantize
  limit = 100
  is_keyword_search = index_name.include?("keyword")
  reindex_queue = is_keyword_search ? index_klass.keyword_reindex_queue : index_klass.reindex_queue
  records = reindex_queue.reserve(limit: limit)

  if records.any?
    record_ids = records.map { |record| record[:record_id] }
    key = records.map { |record| record[:key] }.uniq
    routing_map = records.each_with_object({}) { |record, h| h[record[:record_id]] = record[:routing_key] }

    batch = new_batch(on_success: true)
    batch.properties[:index_class_name] = index_class_name
    batch.properties[:record_ids_count] = record_ids.size
    batch.properties[:key] = key
    batch.properties[:starting_time] = starting_time

    batch.add do
      GoodJob::ElasticsearchProcessBatch.perform_later(index_klass.name, index_name, record_ids, routing_map) if ::Searchkick.client.indices.exists_alias?(name: index_name)
    end
    batch.enqueue

    if record_ids.size == limit
      GoodJob::ElasticsearchProcessQueue.perform_later(index_class_name, index_name)
    end
  end
end