Replace conditional fiber yield in step with pluggable pipe strategy

lib/mementus/pipeline/step.rb

@@ -2,6 +2,13 @@
 
 module Mementus
   module Pipeline
+    class Pipe
+      # Basic passthrough.
+      def call(element)
+        Fiber.yield(element)
+      end
+    end
+
     # Represents a step in a pipeline chain.
     #
     # New steps are constructed from a `source` enumerable (usually the previous
@@ -21,14 +28,10 @@ module Pipeline
     # @param source [Enumerable]
     # @param pipe [#call]
     class Step
-      def initialize(source, pipe=nil)
+      def initialize(source, pipe=Pipe.new)
         @context = Fiber.new do
           source.each do |element|
-            if pipe
-              pipe.call(element)
-            else
-              Fiber.yield(element)
-            end
+            pipe.call(element)
           end
         end
       end

spec/pipeline/step_spec.rb

@@ -48,10 +48,24 @@
       expect(step.all).to eq([:a, :b, :c])
     end
 
-    it 'processes output values based on the given pipe' do
-      pipe = -> (value) { Fiber.yield(value.to_s.upcase) }
-      step = Mementus::Pipeline::Step.new([:a, :b, :c], pipe)
+    it 'transforms output values based on the given pipe' do
+      transform = -> (value) { Fiber.yield(value.to_s.upcase) }
+      step = Mementus::Pipeline::Step.new([:a, :b, :c], transform)
       expect(step.all).to eq(['A', 'B', 'C'])
     end
+
+    it 'filters output values based on the given pipe' do
+      filter = -> (value) { Fiber.yield(value) if value == :a }
+      step = Mementus::Pipeline::Step.new([:a, :b, :c], filter)
+      expect(step.all).to eq([:a])
+    end
+
+    it 'transforms and filters output values based on the given pipes' do
+      filter = -> (value) { Fiber.yield(value) if value == :a }
+      transform = -> (value) { Fiber.yield(value.to_s.upcase) }
+      prev = Mementus::Pipeline::Step.new([:a, :b, :c], filter)
+      step = Mementus::Pipeline::Step.new(prev, transform)
+      expect(step.all).to eq(['A'])
+    end
   end
 end