Add support for aggregating transforms (see #53)

lib/kiba/streaming_runner.rb

@@ -11,6 +11,11 @@ def transform_stream(stream, t)
           end
           y << returned_row if returned_row
         end
+        if t.respond_to?(:close)
+          t.close do |close_row|
+            y << close_row
+          end
+        end
       end
     end
 

test/support/test_aggregate_transform.rb

@@ -0,0 +1,19 @@
+class AggregateTransform
+  def initialize(aggregate_size:)
+    @aggregate_size = aggregate_size
+  end
+
+  def process(row)
+    @buffer ||= []
+    @buffer << row
+    if @buffer.size == @aggregate_size
+      yield @buffer
+      @buffer = []
+    end
+    nil
+  end
+
+  def close
+    yield @buffer unless @buffer.empty?
+  end
+end

test/support/test_non_closing_transform.rb

@@ -0,0 +1,5 @@
+class NonClosingTransform
+  def process(row)
+    row
+  end
+end

test/test_buffering_transform.rb

@@ -0,0 +1,26 @@
+require_relative 'helper'
+require 'kiba/cli'
+require_relative 'support/test_aggregate_transform'
+require_relative 'support/test_non_closing_transform'
+
+class TestBufferingTransform < Kiba::Test
+  def test_buffering_transform
+    destination_array = []
+    job = Kiba.parse do
+      extend Kiba::DSLExtensions::Config
+      config :kiba, runner: Kiba::StreamingRunner
+
+      source TestEnumerableSource, (1..12)
+      # ensure that a non closing transform won't raise an error
+      transform NonClosingTransform
+      transform AggregateTransform, aggregate_size: 5
+      destination TestArrayDestination, destination_array
+    end
+    Kiba.run(job)
+    assert_equal [
+      [1, 2, 3, 4, 5],
+      [6, 7, 8, 9, 10],
+      [11, 12]
+    ], destination_array
+  end
+end