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,42 @@
+# ActiveJob docs: http://edgeguides.rubyonrails.org/active_job_basics.html
+# Example adapters ref: https://github.com/rails/rails/tree/master/activejob/lib/active_job/queue_adapters
+module ActiveJob
+  module QueueAdapters
+    # == Shoryuken concurrent adapter for Active Job
+    #
+    # This adapter sends messages asynchronously (ie non-blocking) and allows
+    # the caller to set up handlers for both success and failure
+    #
+    # To use this adapter, set up as:
+    #
+    # adapter = ActiveJob::QueueAdapters::ShoryukenConcurrentSendAdapter.new
+    # adapter.success_handler = ->(response, job, options) { StatsD.increment(job.class.name + ".success") }
+    # adapter.error_handler = ->(err, job, options) { StatsD.increment(job.class.name + ".failure") }
+    #
+    # config.active_job.queue_adapter = adapter
+    class ShoryukenConcurrentSendAdapter < ShoryukenAdapter
+      attr_accessor :error_handler
+      attr_accessor :success_handler
+
+      def initialize
+        self.error_handler = lambda { |error, job, _options|
+          Shoryuken.logger.warn("Failed to enqueue job: #{job.inspect} due to error: #{error}")
+        }
+        self.success_handler = ->(_send_message_response, _job, _options) { nil }
+      end
+
+      def enqueue(job, options = {})
+        send_concurrently(job, options) { |f_job, f_options| super(f_job, f_options) }
+      end
+
+      private
+
+      def send_concurrently(job, options)
+        Concurrent::Promises
+          .future(job, options) { |f_job, f_options| [ yield(f_job, f_options), f_job, f_options ] }
+          .then { |send_message_response, f_job, f_options| success_handler.call(send_message_response, f_job, f_options) }
+          .rescue(job, options) { |err, f_job, f_options| error_handler.call(err, f_job, f_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,29 @@
+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"
+
+  context "#success_hander" do
+    it "is called when a job succeeds" do
+      options = {}
+      response = true
+      allow(queue).to receive(:send_message).and_return(response)
+      expect(subject.success_handler).to receive(:call).with(response, job, options)
+
+      subject.enqueue(job, options)
+    end
+  end
+
+  context "#error_handler" do
+    it "is called when sending a job fails" do
+      options = {}
+      response = Aws::SQS::Errors::InternalError.new("error", "error")
+      allow(queue).to receive(:send_message).and_raise(response)
+      expect(subject.error_handler).to receive(:call).with(response, job, options).and_call_original
+
+      subject.enqueue(job, options)
+    end
+  end
+end