Seperate ProcessBatchJob again for cleaner code

Andrew Kane
1 parent d92c36f5
Showing 4 changed files with 25 additions and 44 deletions Show diff stats
lib/searchkick/bulk_reindex_job.rb
lib/searchkick/index.rb
lib/searchkick/process_batch_job.rb
lib/searchkick/process_queue_job.rb
@@ -2,7 +2,7 @@ module Searchkick
   class BulkReindexJob < ActiveJob::Base
     queue_as :searchkick
-    def perform(class_name:, record_ids: nil, index_name: nil, method_name: nil, batch_id: nil, min_id: nil, max_id: nil, delete_missing: false)
+    def perform(class_name:, record_ids: nil, index_name: nil, method_name: nil, batch_id: nil, min_id: nil, max_id: nil)
       klass = class_name.constantize
       index = index_name ? Searchkick::Index.new(index_name) : klass.searchkick_index
       record_ids ||= min_id..max_id
@@ -10,9 +10,7 @@ module Searchkick
         Searchkick.load_records(klass, record_ids),
         method_name: method_name,
         batch: true,
-        batch_id: batch_id,
-        delete_missing: delete_missing,
-        record_ids: record_ids
+        batch_id: batch_id
       )
     end
   end
@@ -247,14 +247,14 @@ module Searchkick
       end
     end
-    def import_scope(scope, resume: false, method_name: nil, async: false, batch: false, batch_id: nil, full: false, delete_missing: false, record_ids: nil)
+    def import_scope(scope, resume: false, method_name: nil, async: false, batch: false, batch_id: nil, full: false)
       batch_size = @options[:batch_size] || 1000
       # use scope for import
       scope = scope.search_import if scope.respond_to?(:search_import)
       if batch
-        import_or_update scope.to_a, method_name, async, delete_missing, record_ids, scope.model_name.name.constantize
+        import_or_update scope.to_a, method_name, async
         redis.srem(batches_key, batch_id) if batch_id && redis
       elsif full && async
         if scope.respond_to?(:primary_key)
@@ -422,7 +422,7 @@ module Searchkick
       end
     end
-    def import_or_update(records, method_name, async, delete_missing = false, record_ids = nil, klass = nil)
+    def import_or_update(records, method_name, async)
       if records.any?
         if async
           Searchkick::BulkReindexJob.perform_later(
@@ -433,22 +433,9 @@ module Searchkick
           )
         else
           records = records.select(&:should_index?)
-
-          delete_records =
-            if delete_missing
-              # determine which records to delete
-              (record_ids - records.map { |r| r.id.to_s }).map { |id| m = klass.new; m.id = id; m }
-            else
-              []
-            end
-
-          with_retries do
-            # bulk reindex
-            possibly_bulk(delete_records.any?) do
-              if records.any?
-                method_name ? bulk_update(records, method_name) : import(records)
-              end
-              bulk_delete(delete_records) if delete_records.any?
+          if records.any?
+            with_retries do
+              method_name ? bulk_update(records, method_name) : import(records)
             end
           end
         end
@@ -473,18 +460,6 @@ module Searchkick
       Searchkick.redis
     end
-    # use bulk if no callbacks value set and deleted records
-    # if no deleted records, we can show friendlier notifications
-    def possibly_bulk(deleted_records)
-      if Searchkick.callbacks_value || !deleted_records
-        yield
-      else
-        Searchkick.callbacks(:bulk) do
-          yield
-        end
-      end
-    end
-
     def batches_key
       "searchkick:reindex:#{name}:batches"
     end
@@ -3,12 +3,21 @@ module Searchkick
     queue_as :searchkick
     def perform(class_name:, record_ids:)
-      # job deprecated
-      Searchkick::BulkReindexJob.perform_now(
-        class_name: class_name,
-        record_ids: record_ids,
-        delete_missing: true
-      )
+      klass = class_name.constantize
+      scope = Searchkick.load_records(klass, record_ids)
+      scope = scope.search_import if scope.respond_to?(:search_import)
+      records = scope.select(&:should_index?)
+
+      # determine which records to delete
+      delete_ids = record_ids - records.map { |r| r.id.to_s }
+      delete_records = delete_ids.map { |id| m = klass.new; m.id = id; m }
+
+      # bulk reindex
+      index = klass.searchkick_index
+      Searchkick.callbacks(:bulk) do
+        index.bulk_index(records)
+        index.bulk_delete(delete_records)
+      end
     end
   end
 end
@@ -8,10 +8,9 @@ module Searchkick
       limit = model.searchkick_index.options[:batch_size] || 1000
       record_ids = Searchkick::ReindexQueue.new(model.searchkick_index.name).reserve(limit: limit)
       if record_ids.any?
-        Searchkick::BulkReindexJob.perform_later(
+        Searchkick::ProcessBatchJob.perform_later(
           class_name: model.name,
-          record_ids: record_ids,
-          delete_missing: true
+          record_ids: record_ids
         )
         # TODO when moving to reliable queuing, mark as complete
	@@ -3,12 +3,21 @@ module Searchkick		@@ -3,12 +3,21 @@ module Searchkick
3	queue_as :searchkick	3	queue_as :searchkick
4		4
5	def perform(class_name:, record_ids:)	5	def perform(class_name:, record_ids:)
6	- # job deprecated
7	- Searchkick::BulkReindexJob.perform_now(
8	- class_name: class_name,
9	- record_ids: record_ids,
10	- delete_missing: true
11	- )	6	+ klass = class_name.constantize
		7	+ scope = Searchkick.load_records(klass, record_ids)
		8	+ scope = scope.search_import if scope.respond_to?(:search_import)
		9	+ records = scope.select(&:should_index?)
		10	+
		11	+ # determine which records to delete
		12	+ delete_ids = record_ids - records.map { \|r\| r.id.to_s }
		13	+ delete_records = delete_ids.map { \|id\| m = klass.new; m.id = id; m }
		14	+
		15	+ # bulk reindex
		16	+ index = klass.searchkick_index
		17	+ Searchkick.callbacks(:bulk) do
		18	+ index.bulk_index(records)
		19	+ index.bulk_delete(delete_records)
		20	+ end
12	end	21	end
13	end	22	end
14	end	23	end