lib/shifty/worker.rb in shifty-0.3.0 vs lib/shifty/worker.rb in shifty-0.4.0

@@ -1,20 +1,20 @@
-require 'ostruct'
+require "ostruct"
+require "shifty/taggable"
 
 module Shifty
   class Worker
-    attr_reader :supply
+    attr_reader :supply, :tags
 
-    def initialize(p={}, &block)
-      @supply   = p[:supply]
-      @task     = block || p[:task]
-      @context  = p[:context] || OpenStruct.new
-      # don't define default task here
-      # because we want to allow for
-      # an initialized worker to have
-      # a task injected, including
-      # method-based tasks.
+    include Shifty::Taggable
+
+    def initialize(p = {}, &block)
+      @supply       = p[:supply]
+      @task         = block || p[:task]
+      @context      = p[:context] || OpenStruct.new
+      self.criteria = p[:criteria]
+      self.tags     = p[:tags]
     end
 
     def shift
       ensure_ready_to_work!
       workflow.resume
@@ -22,15 +22,15 @@
 
     def ready_to_work?
       @task && (supply || !task_accepts_a_value?)
     end
 
-    def supplies(subscribing_party)
-      subscribing_party.supply = self
-      subscribing_party
+    def supplies(subscribing_worker)
+      subscribing_worker.supply = self
+      subscribing_worker
     end
-    alias_method :"|", :supplies
+    alias_method :|, :supplies
 
     def supply=(supplier)
       raise WorkerError.new("Worker is a source, and cannot accept a supply") unless suppliable?
       @supply = supplier
     end
@@ -48,33 +48,36 @@
         raise "This worker's task expects to receive a value from a supplier, but has no supply."
       end
     end
 
     def workflow
-      @my_little_machine ||= Fiber.new do
+      @my_little_machine ||= Fiber.new {
         loop do
-          value = supply && supply.shift
-          Fiber.yield @task.call(value, supply, @context)
+          value = supply&.shift
+          if criteria_passes?
+            Fiber.yield @task.call(value, supply, @context)
+          else
+            Fiber.yield value
+          end
         end
-      end
+      }
     end
 
     def default_task
-      Proc.new { |value| value }
+      proc { |value| value }
     end
 
     def task_accepts_a_value?
       @task.arity > 0
     end
 
     def task_method_exists?
-      self.methods.include? :task
+      methods.include? :task
     end
 
     def task_method_accepts_a_value?
-      self.method(:task).arity > 0
+      method(:task).arity > 0
     end
-
   end
 
   class WorkerError < StandardError; end
 end