WIP

examples/default_worker.rb

@@ -1,13 +1,9 @@
 class DefaultWorker
   include Shoryuken::Worker
 
-  shoryuken_options queue: 'test.fifo', auto_delete: true
+  shoryuken_options queue: 'default', auto_delete: true
 
   def perform(sqs_msg, body)
     Shoryuken.logger.info("Received message: '#{body}'")
-
-    raise body
   end
 end
-
-10.times { |i| DefaultWorker.perform_async("#{rand(1000)}-#{i}}") }

lib/shoryuken/cli.rb

@@ -7,7 +7,6 @@
 require 'shoryuken'
 require 'concurrent-edge'
 
-
 module Shoryuken
   # See: https://github.com/mperham/sidekiq/blob/33f5d6b2b6c0dfaab11e5d39688cab7ebadc83ae/lib/sidekiq/cli.rb#L20
   class Shutdown < Interrupt; end

lib/shoryuken/environment_loader.rb

@@ -56,7 +56,6 @@ def initialize_aws
       unless Shoryuken.options[:aws].to_h.empty?
         Shoryuken.logger.warn { '[DEPRECATION] aws in shoryuken.yml is deprecated. Please use configure_server and configure_client in your initializer' }
       end
-
       Shoryuken::AwsConfig.setup(Shoryuken.options[:aws])
     end
 

lib/shoryuken/fetcher.rb

@@ -4,11 +4,6 @@ class Fetcher < Concurrent::Actor::RestartingContext
 
     FETCH_LIMIT = 10
 
-    def on_message(msg)
-      method, *args = msg
-      send(method, *args)
-    end
-
     def initialize(manager)
       @manager = manager
     end

lib/shoryuken/manager.rb

@@ -5,11 +5,6 @@ module Shoryuken
   class Manager < Concurrent::Actor::RestartingContext
     include Util
 
-    def on_message(msg)
-      method, *args = msg
-      send(method, *args)
-    end
-
     def initialize
       @count = Shoryuken.options[:concurrency] || 25
 
@@ -140,13 +135,6 @@ def dispatch
       end
     end
 
-
-    def on_event(event)
-      if event == :reset
-      end
-      logger.info(event.inspect)
-    end
-
     private
 
     def dispatch_later

lib/shoryuken/middleware/server/auto_extend_visibility.rb

@@ -5,14 +5,14 @@ class AutoExtendVisibility
         EXTEND_UPFRONT_SECONDS = 5
 
         def call(worker, queue, sqs_msg, body)
-          # timer = auto_visibility_timer(worker, queue, sqs_msg, body)
+          timer = auto_visibility_timer(worker, queue, sqs_msg, body)
           begin
             yield
           ensure
-            # if timer
-            #   timer.cancel
-            #   @visibility_extender.terminate
-            # end
+            if timer
+              timer.shutdown
+              @visibility_extender.ask!(:terminate)
+            end
           end
         end
 
@@ -24,7 +24,9 @@ class MessageVisibilityExtender < Concurrent::Actor::RestartingContext
           def auto_extend(worker, queue, sqs_msg, body)
             queue_visibility_timeout = Shoryuken::Client.queues(queue).visibility_timeout
 
-            every(queue_visibility_timeout - EXTEND_UPFRONT_SECONDS) do
+            interval = queue_visibility_timeout - EXTEND_UPFRONT_SECONDS
+
+            Concurrent::TimerTask.execute(execution_interval: interval, timeout_interval: interval) do
               begin
                 logger.debug do
                   "Extending message #{worker_name(worker.class, sqs_msg, body)}/#{queue}/#{sqs_msg.message_id} " \
@@ -45,7 +47,7 @@ def auto_extend(worker, queue, sqs_msg, body)
 
         def auto_visibility_timer(worker, queue, sqs_msg, body)
           return unless worker.class.auto_visibility_timeout?
-          @visibility_extender = MessageVisibilityExtender.new_link
+          @visibility_extender = MessageVisibilityExtender.spawn! name: :message_visibility_extender, link: true
           @visibility_extender.auto_extend(worker, queue, sqs_msg, body)
         end
       end

lib/shoryuken/processor.rb

@@ -8,11 +8,6 @@ def initialize(manager)
       @manager = manager
     end
 
-    def on_message(msg)
-      method, *args = msg
-      send(method, *args)
-    end
-
     def process(queue, sqs_msg)
       worker = Shoryuken.worker_registry.fetch_worker(queue, sqs_msg)
 

lib/shoryuken/util.rb

@@ -1,5 +1,10 @@
 module Shoryuken
   module Util
+    def on_message(msg)
+      method, *args = msg
+      send(method, *args)
+    end
+
     def watchdog(last_words)
       yield
     rescue => ex