explode_with_merge_and_buffer.rb

require 'kiba'
require 'awesome_print'

class ExampleSource
  def each
    (1..10).each do |i|
      yield ({
        index: i
      })
    end
  end
end

class BufferedPrinter
  def initialize(buffer: 3)
    @buffer = buffer
    @rows = []
  end

  def write(row)
    @rows << row
    flush if @rows.size >= @buffer
  end

  def close
    flush
  end

  def flush
    @rows.each {|r| ap r}
    @rows = []
  end
end

class ArrayExploder
  def process(row)
    if row.is_a? Array
      row.each{|r| yield r}
      return nil
    end
    row
  end
end


job = Kiba.parse do
  source ExampleSource

  transform do |row|
    [1, 2, 3, 4, 5].map do |i|
      row.merge(sub_index: i)
    end
  end
  transform ArrayExploder

  transform do |row|
    %w(guy buddy pal).map do |guy|
      row.merge(guy: guy)
    end
  end
  transform ArrayExploder


  destination BufferedPrinter
end

Kiba.run job