Implement a first, working but ugly version of perform_all_later

Supporting concurrency-limited, scheduled and immediate jobs.

Author: rosa

app/models/solid_queue/job.rb

@@ -13,24 +13,54 @@ class SolidQueue::Job < SolidQueue::Record
   DEFAULT_QUEUE_NAME = "default"
 
   class << self
-    def enqueue_active_job(active_job, scheduled_at: Time.current)
-      enqueue \
-        queue_name: active_job.queue_name,
-        active_job_id: active_job.job_id,
-        priority: active_job.priority,
-        scheduled_at: scheduled_at,
-        class_name: active_job.class.name,
-        arguments: active_job.serialize,
-        concurrency_key: active_job.try(:concurrency_key)
-    end
+    def enqueue_active_jobs(active_jobs)
+      scheduled_jobs, immediate_jobs = active_jobs.partition(&:scheduled_at)
+      with_concurrency_limits, without_concurrency_limits = immediate_jobs.partition(&:concurrency_limited?)
+
+      with_concurrency_limits.each do |active_job|
+        enqueue_active_job(active_job)
+      end
 
-    def enqueue(**kwargs)
-      create!(**kwargs.compact.with_defaults(defaults)).tap do
-        SolidQueue.logger.debug "[SolidQueue] Enqueued job #{kwargs}"
+      transaction do
+        job_rows = scheduled_jobs.map { |job| attributes_from_active_job(job) }
+        self.insert_all(job_rows)
+        inserted_jobs = where(active_job_id: scheduled_jobs.map(&:job_id))
+        execution_rows = inserted_jobs.map { |job| job.attributes.slice("queue_name", "priority", "scheduled_at").merge(job_id: job.id) }
+        SolidQueue::ScheduledExecution.insert_all(execution_rows)
+      end
+
+      transaction do
+        job_rows = without_concurrency_limits.map { |job| attributes_from_active_job(job) }
+        self.insert_all(job_rows)
+        inserted_jobs = where(active_job_id: without_concurrency_limits.map(&:job_id))
+        execution_rows = inserted_jobs.map { |job| job.attributes.slice("queue_name", "priority").merge(job_id: job.id) }
+        SolidQueue::ReadyExecution.insert_all(execution_rows)
       end
     end
 
+    def enqueue_active_job(active_job, scheduled_at: Time.current)
+      enqueue **attributes_from_active_job(active_job).reverse_merge(scheduled_at: scheduled_at)
+    end
+
     private
+      def enqueue(**kwargs)
+        create!(**kwargs).tap do
+          SolidQueue.logger.debug "[SolidQueue] Enqueued job #{kwargs}"
+        end
+      end
+
+      def attributes_from_active_job(active_job)
+        {
+          queue_name: active_job.queue_name,
+          active_job_id: active_job.job_id,
+          priority: active_job.priority,
+          scheduled_at: active_job.scheduled_at,
+          class_name: active_job.class.name,
+          arguments: active_job.serialize,
+          concurrency_key: active_job.concurrency_key
+        }.compact.with_defaults(defaults)
+      end
+
       def defaults
         { queue_name: DEFAULT_QUEUE_NAME, priority: DEFAULT_PRIORITY }
       end

app/models/solid_queue/job/executable.rb

@@ -17,7 +17,7 @@ module Executable
         after_create :prepare_for_execution
 
         scope :finished, -> { where.not(finished_at: nil) }
-        scope :failed, -> { includes(:failed_execution).where.not(failed_execution: {id: nil}) }
+        scope :failed, -> { includes(:failed_execution).where.not(failed_execution: { id: nil }) }
       end
 
       %w[ ready claimed failed scheduled ].each do |status|

lib/active_job/concurrency_controls.rb

@@ -36,6 +36,10 @@ def concurrency_key
       end
     end
 
+    def concurrency_limited?
+      concurrency_key.present?
+    end
+
     private
       def concurrency_group
         compute_concurrency_parameter(self.class.concurrency_group)

lib/active_job/queue_adapters/solid_queue_adapter.rb

@@ -19,6 +19,10 @@ def enqueue_at(active_job, timestamp) # :nodoc:
           active_job.provider_job_id = job.id
         end
       end
+
+      def enqueue_all(active_jobs) # :nodoc:
+        SolidQueue::Job.enqueue_active_jobs(active_jobs)
+      end
     end
   end
 end

test/models/solid_queue/job_test.rb

@@ -102,6 +102,24 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     end
   end
 
+  test "enqueue multiple jobs" do
+    active_jobs = [
+      AddToBufferJob.new(2),
+      AddToBufferJob.new(6).set(wait: 2.minutes),
+      NonOverlappingJob.new(@result),
+      StoreResultJob.new(42),
+      AddToBufferJob.new(4),
+      NonOverlappingGroupedJob1.new(@result),
+      AddToBufferJob.new(6).set(wait: 3.minutes),
+      NonOverlappingJob.new(@result),
+      NonOverlappingGroupedJob2.new(@result)
+    ]
+
+    assert_multi(ready: 5, scheduled: 2, blocked: 2) do
+      ActiveJob.perform_all_later(active_jobs)
+    end
+  end
+
   test "block jobs when concurrency limits are reached" do
     assert_ready do
       NonOverlappingJob.perform_later(@result, name: "A")
@@ -148,4 +166,14 @@ def assert_blocked(&block)
         assert_difference -> { SolidQueue::Job.count } => +1, -> { SolidQueue::BlockedExecution.count } => +1, &block
       end
     end
+
+    def assert_multi(ready: 0, scheduled: 0, blocked: 0, &block)
+      assert_difference -> { SolidQueue::Job.count }, +(ready + scheduled + blocked) do
+        assert_difference -> { SolidQueue::ReadyExecution.count }, +ready do
+          assert_difference -> { SolidQueue::ScheduledExecution.count }, +scheduled do
+            assert_difference -> { SolidQueue::BlockedExecution.count }, +blocked, &block
+          end
+        end
+      end
+    end
 end