Use pipeline for migration too

Author: dhh

lib/kredis/migration.rb

@@ -10,29 +10,29 @@ def initialize(config = :shared)
   end
 
   def migrate_all(key_pattern)
-    each_key_batch_matching(key_pattern) do |keys|
+    each_key_batch_matching(key_pattern) do |keys, pipeline|
       keys.each do |key|
         ids = key.scan(/\d+/).map(&:to_i)
-        migrate from: key, to: yield(key, *ids)
+        migrate from: key, to: yield(key, *ids), pipeline: pipeline
       end
     end
   end
 
-  def migrate(from:, to:)
+  def migrate(from:, to:, pipeline: nil)
     namespaced_to = Kredis.namespaced_key(to)
 
     if to.present? && from != namespaced_to
       log_migration "Migrating key #{from} to #{namespaced_to}" do
-        @redis.evalsha @copy_sha, keys: [ from, namespaced_to ]
+        (pipeline || @redis).evalsha @copy_sha, keys: [ from, namespaced_to ]
       end
     else
       log_migration "Skipping blank/unaltered migration key #{from} → #{to}"
     end
   end
 
   def delete_all(key_pattern)
-    each_key_batch_matching(key_pattern) do |keys|
-      @redis.del *keys
+    each_key_batch_matching(key_pattern) do |keys, pipeline|
+      pipeline.del *keys
     end
   end
 
@@ -43,7 +43,7 @@ def each_key_batch_matching(key_pattern, &block)
       cursor = "0"
       begin
         cursor, keys = @redis.scan(cursor, match: key_pattern, count: SCAN_BATCH_SIZE)
-        @redis.pipelined { yield keys }
+        @redis.multi { |pipeline| yield keys, pipeline }
       end until cursor == "0"
     end