push bulk with dynamic queues (sidekiq#6552)

Allow payloads with mixed queues.

Author: ericproulx

lib/sidekiq/client.rb

@@ -269,14 +269,16 @@ def atomic_push(conn, payloads)
           [at, Sidekiq.dump_json(hash)]
         })
       else
-        queue = payloads.first["queue"]
         now = Time.now.to_f
-        to_push = payloads.map { |entry|
-          entry["enqueued_at"] = now
-          Sidekiq.dump_json(entry)
-        }
-        conn.sadd("queues", [queue])
-        conn.lpush("queue:#{queue}", to_push)
+        grouped_queues = payloads.group_by { |job| job["queue"] }
+        conn.sadd("queues", grouped_queues.keys)
+        grouped_queues.each do |queue, grouped_payloads|
+          to_push = grouped_payloads.map { |entry|
+            entry["enqueued_at"] = now
+            Sidekiq.dump_json(entry)
+          }
+          conn.lpush("queue:#{queue}", to_push)
+        end
       end
     end
   end

test/client_test.rb

@@ -58,6 +58,13 @@ def call(worker_class, job, queue, redis)
   end
 end
 
+class MiddlewareDynamicQueue
+  def call(worker_class, job, queue, redis)
+    job["queue"] = job["args"].first.odd? ? "odd_queue" : "even_queue"
+    yield
+  end
+end
+
 class DJob < BaseJob
 end
 
@@ -479,6 +486,18 @@ def call(worker_klass, msg, q, r)
       refute result[0]
       assert_match(/[0-9a-f]{12}/, result[1])
     end
+
+    it "push bulk sends worker in correct queue" do
+      @client.middleware do |chain|
+        chain.add MiddlewareDynamicQueue
+      end
+
+      @client.push_bulk("class" => MyJob, "args" => 3.times.map { [_1] })
+      even_queue = Sidekiq::Queue.new("even_queue")
+      odd_queue = Sidekiq::Queue.new("odd_queue")
+      assert_equal 2, even_queue.size
+      assert_equal 1, odd_queue.size
+    end
   end
 
   describe "inheritance" do