lib/wayfarer/base.rb in wayfarer-0.4.3 vs lib/wayfarer/base.rb in wayfarer-0.4.4

@@ -1,50 +1,60 @@
 # frozen_string_literal: true
 
 module Wayfarer
   class Base < ActiveJob::Base
-    include Wayfarer::Middleware::Worker
-    extend Forwardable
+    include Wayfarer::Middleware::Controller
 
-    after_enqueue do |_job|
+    use Wayfarer::Middleware::Stage
+    use Wayfarer::Middleware::Dedup
+    use Wayfarer::Middleware::Normalize
+    use Wayfarer::Middleware::Router
+    use Wayfarer::Middleware::Fetch
+    use Wayfarer::Middleware::Dispatch
+
+    ErrorHandler = lambda do |&block|
+      lambda do |job, error|
+        task = job.arguments.first
+        task.barrier.seen?(task.url)
+        task.gc.run
+        block.call(job, error)
+      end
+    end
+
+    after_enqueue do |job|
+      task = job.arguments.first
       task.counter.increment
     end
 
-    after_perform { |job| GC.new(job).run }
+    after_perform do |job|
+      task = job.arguments.first
+      task.gc.run
+    end
 
-    rescue_from(StandardError) { GC.new(self).run }
+    rescue_from(StandardError) do
+      task = arguments.first
+      task.gc.run
+    end
 
-    def self.retry_on(*argv)
-      super(*argv) do |job, error|
-        job.task.barrier.seen?(job.task.url)
-        GC.new(job).run
-        yield job, error if block_given?
-      end
+    def self.retry_on(*argv, &block)
+      super(*argv, &ErrorHandler.call(&block))
     end
 
-    def self.discard_on(*argv)
-      super(*argv) do |job, error|
-        job.task.barrier.seen?(job.task.url)
-        GC.new(job).run
-        yield job, error if block_given?
-      end
+    def self.discard_on(*argv, &block)
+      super(*argv, &ErrorHandler.call(&block))
     end
 
-    def self.crawl_later(url, batch: SecureRandom.uuid)
-      perform_later(Task.new(url, batch))
+    def self.crawl(url, batch: SecureRandom.uuid)
+      Task.new(url, batch).tap do |task|
+        perform_later(task)
+      end
     end
 
     def retry_job(...)
       super(...) # increments the counter by re-enqueuing the job
+      task = arguments.first
       task.counter.decrement
     end
 
-    def perform(task)
-      task.job = self
-      chain.call(task)
-    end
-
-    def task
-      arguments.first
-    end
+    alias perform call
   end
 end