Improved reindexing and notifications [skip ci]

Andrew Kane
1 parent fa251bc8
Showing 3 changed files with 54 additions and 24 deletions Show diff stats
lib/searchkick.rb
lib/searchkick/record_indexer.rb
lib/searchkick/reindex_v2_job.rb
@@ -205,17 +205,20 @@ module Searchkick
   end
  
   # message is private
-  def self.callbacks(value = nil, message: "Bulk")
+  def self.callbacks(value = nil, message: nil)
     if block_given?
       previous_value = callbacks_value
       begin
         self.callbacks_value = value
         result = yield
-        if callbacks_value == :bulk
-          event = {
-            name: message,
-            count: indexer.queued_items.size
-          }
+        if callbacks_value == :bulk && indexer.queued_items.any?
+          event = {}
+          if message
+            message.call(event)
+          else
+            event[:name] = "Bulk"
+            event[:count] = indexer.queued_items.size
+          end
           ActiveSupport::Notifications.instrument("request.searchkick", event) do
             indexer.perform
           end
@@ -47,13 +47,13 @@ module Searchkick
         index.reindex_queue.push_records(records)
       when true, :inline
         index_records, other_records = records.partition { |r| index_record?(r) }
-        import_inline(index_records, !full ? other_records : [], method_name: method_name)
+        import_inline(index_records, !full ? other_records : [], method_name: method_name, single: single)
       else
         raise ArgumentError, "Invalid value for mode"
       end
     end
  
-    def reindex_items(klass, items, method_name:)
+    def reindex_items(klass, items, method_name:, single: false)
       routing = items.to_h { |r| [r[:id], r[:routing]] }
       record_ids = routing.keys
  
@@ -68,7 +68,7 @@ module Searchkick
           construct_record(klass, id, routing[id])
         end
  
-      import_inline(records, delete_records, method_name: method_name)
+      import_inline(records, delete_records, method_name: method_name, single: false)
     end
  
     private
@@ -78,23 +78,51 @@ module Searchkick
     end
  
     # import in single request with retries
-    def import_inline(index_records, delete_records, method_name:)
+    def import_inline(index_records, delete_records, method_name:, single:)
       return if index_records.empty? && delete_records.empty?
  
-      action = method_name ? "Update" : "Import"
-      name = (index_records.first || delete_records.first).searchkick_klass.name
-      with_retries do
-        Searchkick.callbacks(:bulk, message: "#{name} #{action}") do
-          if index_records.any?
-            if method_name
-              index.bulk_update(index_records, method_name)
-            else
-              index.bulk_index(index_records)
-            end
+      maybe_bulk(index_records, delete_records, method_name, single) do
+        if index_records.any?
+          if method_name
+            index.bulk_update(index_records, method_name)
+          else
+            index.bulk_index(index_records)
           end
+        end
+
+        if delete_records.any?
+          index.bulk_delete(delete_records)
+        end
+      end
+    end
+
+    def maybe_bulk(index_records, delete_records, method_name, single)
+      if Searchkick.callbacks_value == :bulk
+        yield
+      else
+        # set action and data
+        action =
+          if single && index_records.empty?
+            "Remove"
+          elsif method_name
+            "Update"
+          else
+            single ? "Store" : "Import"
+          end
+        record = index_records.first || delete_records.first
+        name = record.searchkick_klass.name
+        message = lambda do |event|
+          event[:name] = "#{name} #{action}"
+          if single
+            event[:id] = index.search_id(record)
+          else
+            event[:count] = index_records.size + delete_records.size
+          end
+        end
  
-          if delete_records.any?
-            index.bulk_delete(delete_records)
+        with_retries do
+          Searchkick.callbacks(:bulk, message: message) do
+            yield
           end
         end
       end
@@ -7,8 +7,7 @@ module Searchkick
       # may not be needed if calling search_import later
       model = model.unscoped if model.respond_to?(:unscoped)
       items = [{id: id, routing: routing}]
-      # TODO improve notification
-      model.searchkick_index.send(:record_indexer).reindex_items(model, items, method_name: method_name)
+      model.searchkick_index.send(:record_indexer).reindex_items(model, items, method_name: method_name, single: true)
     end
   end
 end
...	...	@@ -205,17 +205,20 @@ module Searchkick
205	205	end
206	206
207	207	# message is private
208		- def self.callbacks(value = nil, message: "Bulk")
	208	+ def self.callbacks(value = nil, message: nil)
209	209	if block_given?
210	210	previous_value = callbacks_value
211	211	begin
212	212	self.callbacks_value = value
213	213	result = yield
214		- if callbacks_value == :bulk
215		- event = {
216		- name: message,
217		- count: indexer.queued_items.size
218		- }
	214	+ if callbacks_value == :bulk && indexer.queued_items.any?
	215	+ event = {}
	216	+ if message
	217	+ message.call(event)
	218	+ else
	219	+ event[:name] = "Bulk"
	220	+ event[:count] = indexer.queued_items.size
	221	+ end
219	222	ActiveSupport::Notifications.instrument("request.searchkick", event) do
220	223	indexer.perform
221	224	end
...	...
...	...	@@ -47,13 +47,13 @@ module Searchkick
47	47	index.reindex_queue.push_records(records)
48	48	when true, :inline
49	49	index_records, other_records = records.partition { \|r\| index_record?(r) }
50		- import_inline(index_records, !full ? other_records : [], method_name: method_name)
	50	+ import_inline(index_records, !full ? other_records : [], method_name: method_name, single: single)
51	51	else
52	52	raise ArgumentError, "Invalid value for mode"
53	53	end
54	54	end
55	55
56		- def reindex_items(klass, items, method_name:)
	56	+ def reindex_items(klass, items, method_name:, single: false)
57	57	routing = items.to_h { \|r\| [r[:id], r[:routing]] }
58	58	record_ids = routing.keys
59	59
...	...	@@ -68,7 +68,7 @@ module Searchkick
68	68	construct_record(klass, id, routing[id])
69	69	end
70	70
71		- import_inline(records, delete_records, method_name: method_name)
	71	+ import_inline(records, delete_records, method_name: method_name, single: false)
72	72	end
73	73
74	74	private
...	...	@@ -78,23 +78,51 @@ module Searchkick
78	78	end
79	79
80	80	# import in single request with retries
81		- def import_inline(index_records, delete_records, method_name:)
	81	+ def import_inline(index_records, delete_records, method_name:, single:)
82	82	return if index_records.empty? && delete_records.empty?
83	83
84		- action = method_name ? "Update" : "Import"
85		- name = (index_records.first \|\| delete_records.first).searchkick_klass.name
86		- with_retries do
87		- Searchkick.callbacks(:bulk, message: "#{name} #{action}") do
88		- if index_records.any?
89		- if method_name
90		- index.bulk_update(index_records, method_name)
91		- else
92		- index.bulk_index(index_records)
93		- end
	84	+ maybe_bulk(index_records, delete_records, method_name, single) do
	85	+ if index_records.any?
	86	+ if method_name
	87	+ index.bulk_update(index_records, method_name)
	88	+ else
	89	+ index.bulk_index(index_records)
94	90	end
	91	+ end
	92	+
	93	+ if delete_records.any?
	94	+ index.bulk_delete(delete_records)
	95	+ end
	96	+ end
	97	+ end
	98	+
	99	+ def maybe_bulk(index_records, delete_records, method_name, single)
	100	+ if Searchkick.callbacks_value == :bulk
	101	+ yield
	102	+ else
	103	+ # set action and data
	104	+ action =
	105	+ if single && index_records.empty?
	106	+ "Remove"
	107	+ elsif method_name
	108	+ "Update"
	109	+ else
	110	+ single ? "Store" : "Import"
	111	+ end
	112	+ record = index_records.first \|\| delete_records.first
	113	+ name = record.searchkick_klass.name
	114	+ message = lambda do \|event\|
	115	+ event[:name] = "#{name} #{action}"
	116	+ if single
	117	+ event[:id] = index.search_id(record)
	118	+ else
	119	+ event[:count] = index_records.size + delete_records.size
	120	+ end
	121	+ end
95	122
96		- if delete_records.any?
97		- index.bulk_delete(delete_records)
	123	+ with_retries do
	124	+ Searchkick.callbacks(:bulk, message: message) do
	125	+ yield
98	126	end
99	127	end
100	128	end
...	...
...	...	@@ -7,8 +7,7 @@ module Searchkick
7	7	# may not be needed if calling search_import later
8	8	model = model.unscoped if model.respond_to?(:unscoped)
9	9	items = [{id: id, routing: routing}]
10		- # TODO improve notification
11		- model.searchkick_index.send(:record_indexer).reindex_items(model, items, method_name: method_name)
	10	+ model.searchkick_index.send(:record_indexer).reindex_items(model, items, method_name: method_name, single: true)
12	11	end
13	12	end
14	13	end
...	...