Merge pull request #8 from ridecharge/pull-middlewares-logic-from-upstream

Pull middlewares logic from upstream

Author: yuryroot

examples/middleware_worker.rb

@@ -0,0 +1,34 @@
+$: << File.expand_path('../lib', File.dirname(__FILE__))
+require 'sneakers'
+require 'sneakers/runner'
+
+class MiddlewareWorker
+  include Sneakers::Worker
+
+  from_queue 'middleware-demo',
+    ack: false
+
+  def work(message)
+    puts "******** MiddlewareWorker -> #{message}"
+  end
+end
+
+class DemoMiddleware
+  def initialize(app, *args)
+    @app = app
+    @args = args
+  end
+
+  def call(deserialized_msg, delivery_info, metadata, handler)
+    puts "******** DemoMiddleware - before; args #{@args}"
+    @app.call(deserialized_msg, delivery_info, metadata, handler)
+    puts "******** DemoMiddleware - after"
+  end
+end
+
+Sneakers.configure
+Sneakers.middleware.use(DemoMiddleware, foo: :bar)
+
+Sneakers.publish("{}", :to_queue => 'middleware-demo')
+r = Sneakers::Runner.new([MiddlewareWorker])
+r.run

lib/sneakers.rb

@@ -18,6 +18,7 @@ module Concerns
 require 'sneakers/concerns/metrics'
 require 'sneakers/handlers/oneshot'
 require 'sneakers/content_type'
+require 'sneakers/middleware/config'
 require 'sneakers/worker'
 require 'sneakers/publisher'
 
@@ -87,6 +88,10 @@ def error_reporters
     CONFIG[:error_reporters]
   end
 
+  def middleware
+    @middleware ||= Sneakers::Middleware::Config
+  end
+
   private
 
   def setup_general_logger!

lib/sneakers/middleware/config.rb

@@ -0,0 +1,23 @@
+module Sneakers
+  module Middleware
+    class Config
+      def self.use(klass, args)
+        middlewares << { class: klass, args: args }
+      end
+
+      def self.delete(klass)
+        middlewares.reject! { |el| el[:class] == klass }
+      end
+
+      def self.to_a
+        middlewares
+      end
+
+      def self.middlewares
+        @middlewares ||= []
+      end
+
+      private_class_method :middlewares
+    end
+  end
+end

lib/sneakers/version.rb

@@ -1,3 +1,3 @@
 module Sneakers
-  VERSION = "0.5.0.curb"
+  VERSION = "10.0.0.curb"
 end

lib/sneakers/worker.rb

@@ -55,11 +55,20 @@ def do_work(delivery_info, metadata, msg, handler)
           Timeout.timeout(@timeout_after, WorkerTimeout) do
             metrics.timing("work.#{self.class.name}.time") do
               deserialized_msg = ContentType.deserialize(msg, @content_type || metadata && metadata[:content_type])
-              if @call_with_params
-                res = work_with_params(deserialized_msg, delivery_info, metadata)
-              else
-                res = work(deserialized_msg)
+
+              app = -> (deserialized_msg, delivery_info, metadata, handler) do
+                if @call_with_params
+                  work_with_params(deserialized_msg, delivery_info, metadata)
+                else
+                  work(deserialized_msg)
+                end
+              end
+
+              middlewares = Sneakers.middleware.to_a
+              block_to_call = middlewares.reverse.reduce(app) do |mem, h|
+                h[:class].new(mem, *h[:args])
               end
+              res = block_to_call.call(deserialized_msg, delivery_info, metadata, handler)
             end
           end
         rescue WorkerTimeout => ex

sneakers.gemspec

@@ -37,5 +37,6 @@ Gem::Specification.new do |gem|
   gem.add_development_dependency 'rake'
   gem.add_development_dependency 'minitest'
   gem.add_development_dependency 'guard'
+  gem.add_development_dependency 'pry-byebug'
 end
 

spec/sneakers/worker_spec.rb

@@ -411,6 +411,61 @@ def with_test_queuefactory(ctx, ack=true, msg=nil, nowork=false)
       w.do_work(header, nil, "msg", handler)
     end
 
+    describe 'middleware' do
+      let(:middleware) do
+        Class.new do
+          def initialize(app, *args)
+            @app = app
+          end
+
+          def call(deserialized_msg, delivery_info, metadata, handler)
+            @app.call(deserialized_msg, delivery_info, metadata, handler)
+          end
+        end
+      end
+
+      let(:worker) do
+        Class.new do
+          include Sneakers::Worker
+          from_queue 'defaults', ack: false
+
+          def work_with_params(msg, delivery_info, metadata)
+            msg
+          end
+        end
+      end
+
+      before do
+        Sneakers.middleware.use(middleware, 'args')
+
+        @delivery_info = Object.new
+        @metadata = Object.new
+        stub(@metadata).[](:content_type) { 'some/fake' }
+        @message = Object.new
+        @handler = Object.new
+      end
+
+      after do
+        Sneakers.middleware.delete(middleware)
+      end
+
+      it 'should process job and call #work_with_params/#work' do
+        w = worker.new(@queue, TestPool.new)
+        mock(w).work_with_params(@message, @delivery_info, @metadata).once
+
+        w.do_work(@delivery_info, @metadata, @message, @handler)
+      end
+
+      it "should call registered middleware" do
+        mock.proxy(middleware).new(instance_of(Proc), 'args').once do |res|
+          mock.proxy(res).call(@message, @delivery_info, @metadata, @handler).once
+        end
+
+        w = worker.new(@queue, TestPool.new)
+        w.do_work(@delivery_info, @metadata, @message, @handler)
+      end
+    end
+
     describe "with ack" do
       before do
         @delivery_info = Object.new

spec/spec_helper.rb

@@ -1,6 +1,8 @@
 require 'bundler/setup'
 require 'simplecov'
 require 'resolv'
+require 'pry-byebug'
+
 SimpleCov.start do
   add_filter "/spec/"
 end