Deliver Messages Async

README.md

@@ -37,7 +37,7 @@ If you are using AWS SDK version 3, please also add this line:
 gem 'aws-sdk-sqs'
 ```
 
-The extra gem `aws-sdk-sqs` is required in order to keep Shoryuken compatible with AWS SDK version 2 and 3. 
+The extra gem `aws-sdk-sqs` is required in order to keep Shoryuken compatible with AWS SDK version 2 and 3.
 
 And then execute:
 

lib/shoryuken.rb

@@ -88,4 +88,7 @@ def self.shoryuken_options
   )
 end
 
-require 'shoryuken/extensions/active_job_adapter' if Shoryuken.active_job?
+if Shoryuken.active_job?
+  require 'shoryuken/extensions/active_job_adapter'
+  require 'shoryuken/extensions/active_job_concurrent_send_adapter'
+end

lib/shoryuken/default_worker_registry.rb

@@ -1,7 +1,7 @@
 module Shoryuken
   class DefaultWorkerRegistry < WorkerRegistry
     def initialize
-      @workers = {}
+      @workers = Concurrent::Hash.new
     end
 
     def batch_receive_messages?(queue)

lib/shoryuken/extensions/active_job_concurrent_send_adapter.rb

@@ -0,0 +1,26 @@
+ module ActiveJob
+   module QueueAdapters
+     class ShoryukenConcurrentSendAdapter < ShoryukenAdapter
+
+       attr_accessor :error_handler
+       attr_accessor :success_handler
+
+       def initialize
+         @error_handler = ->(error, job, options) { Shoryuken.logger.warn("Failed to enqueue job: #{job.inspect} due to error: #{error}") }
+         @success_handler = ->(job, options) { nil }
+       end
+
+       def enqueue(job, options = {})
+         send_concurrently(job, options) { |job, options| super(job, options) }
+       end
+
+       private
+
+       def send_concurrently(job, options)
+         Concurrent::Promises.future(job, options) { |job, options| yield(job, options) }
+           .then(job, options) { |job, options| success_handler.call(job, options) }
+           .rescue(job, options) { |err, (job, options)| error_handler.call(err, job, options) }
+       end
+     end
+   end
+ end

spec/shared_examples_for_active_job.rb

@@ -0,0 +1,61 @@
+RSpec.shared_examples "active_job_adapters" do
+  let(:job) { double 'Job', id: '123', queue_name: 'queue' }
+  let(:fifo) { false }
+  let(:queue) { double 'Queue', fifo?: fifo }
+
+  before do
+    allow(Shoryuken::Client).to receive(:queues).with(job.queue_name).and_return(queue)
+    allow(job).to receive(:serialize).and_return(
+      'job_class' => 'Worker',
+      'job_id'     => job.id,
+      'queue_name' => job.queue_name,
+      'arguments'  => nil,
+      'locale'     => nil
+    )
+  end
+
+  describe '#enqueue' do
+    specify do
+      expect(queue).to receive(:send_message) do |hash|
+        expect(hash[:message_deduplication_id]).to_not be
+      end
+      expect(Shoryuken).to receive(:register_worker).with(job.queue_name, described_class::JobWrapper)
+
+      subject.enqueue(job)
+    end
+
+    context 'when fifo' do
+      let(:fifo) { true }
+
+      it 'does not include job_id in the deduplication_id' do
+        expect(queue).to receive(:send_message) do |hash|
+          message_deduplication_id = Digest::SHA256.hexdigest(JSON.dump(job.serialize.except('job_id')))
+
+          expect(hash[:message_deduplication_id]).to eq(message_deduplication_id)
+        end
+        expect(Shoryuken).to receive(:register_worker).with(job.queue_name, described_class::JobWrapper)
+
+        subject.enqueue(job)
+      end
+    end
+  end
+
+  describe '#enqueue_at' do
+    specify do
+      delay = 1
+
+      expect(queue).to receive(:send_message) do |hash|
+        expect(hash[:message_deduplication_id]).to_not be
+        expect(hash[:delay_seconds]).to eq(delay)
+      end
+
+      expect(Shoryuken).to receive(:register_worker).with(job.queue_name, described_class::JobWrapper)
+
+      # need to figure out what to require Time.current and N.minutes to remove the stub
+      allow(subject).to receive(:calculate_delay).and_return(delay)
+
+      subject.enqueue_at(job, nil)
+    end
+  end
+
+end

spec/shoryuken/extensions/active_job_adapter_spec.rb

@@ -1,64 +1,7 @@
 require 'spec_helper'
-require 'active_job'
 require 'shoryuken/extensions/active_job_adapter'
+require 'shared_examples_for_active_job'
 
 RSpec.describe ActiveJob::QueueAdapters::ShoryukenAdapter do
-  let(:job) { double 'Job', id: '123', queue_name: 'queue' }
-  let(:fifo) { false }
-  let(:queue) { double 'Queue', fifo?: fifo }
-
-  before do
-    allow(Shoryuken::Client).to receive(:queues).with(job.queue_name).and_return(queue)
-    allow(job).to receive(:serialize).and_return(
-      'job_class' => 'Worker',
-      'job_id'     => job.id,
-      'queue_name' => job.queue_name,
-      'arguments'  => nil,
-      'locale'     => nil
-    )
-  end
-
-  describe '#enqueue' do
-    specify do
-      expect(queue).to receive(:send_message) do |hash|
-        expect(hash[:message_deduplication_id]).to_not be
-      end
-      expect(Shoryuken).to receive(:register_worker).with(job.queue_name, described_class::JobWrapper)
-
-      subject.enqueue(job)
-    end
-
-    context 'when fifo' do
-      let(:fifo) { true }
-
-      it 'does not include job_id in the deduplication_id' do
-        expect(queue).to receive(:send_message) do |hash|
-          message_deduplication_id = Digest::SHA256.hexdigest(JSON.dump(job.serialize.except('job_id')))
-
-          expect(hash[:message_deduplication_id]).to eq(message_deduplication_id)
-        end
-        expect(Shoryuken).to receive(:register_worker).with(job.queue_name, described_class::JobWrapper)
-
-        subject.enqueue(job)
-      end
-    end
-  end
-
-  describe '#enqueue_at' do
-    specify do
-      delay = 1
-
-      expect(queue).to receive(:send_message) do |hash|
-        expect(hash[:message_deduplication_id]).to_not be
-        expect(hash[:delay_seconds]).to eq(delay)
-      end
-
-      expect(Shoryuken).to receive(:register_worker).with(job.queue_name, described_class::JobWrapper)
-
-      # need to figure out what to require Time.current and N.minutes to remove the stub
-      allow(subject).to receive(:calculate_delay).and_return(delay)
-
-      subject.enqueue_at(job, nil)
-    end
-  end
+  include_examples "active_job_adapters"
 end

spec/shoryuken/extensions/active_job_concurrent_send_adapter_spec.rb

@@ -0,0 +1,7 @@
+require 'spec_helper'
+require 'shared_examples_for_active_job'
+require 'shoryuken/extensions/active_job_concurrent_send_adapter'
+
+RSpec.describe ActiveJob::QueueAdapters::ShoryukenConcurrentSendAdapter do
+  include_examples "active_job_adapters"
+end