Processing CSV files more fast with Fibers + EM.defer

gistfile1.rb

require 'bundler/setup'
require 'fiber'
require 'csv'
require 'pp'
require 'eventmachine'

def a_slow_api_call(row)
  sleep 0.3

  row.collect(&:to_i).inject(&:+)
end

def another_api_call(row)
  sleep 0.1

  row.collect(&:to_i).inject(&:+) * 2
end

def consume(row)
  return if row.nil?
  
  long_process = Proc.new do
    # call an API...
    result_1 = a_slow_api_call(row)

    # call other API...
    result_2 = another_api_call(row)

    data = [ row[0], row[1], result_1, result_2 ]
    data
  end

  process_result = Proc.new do |data|
    # do some stuff with data
    puts "#{row[0].rjust(7, ' ')} - #{data[1]} - #{data[2]} - #{data[3]}"
  end

  EM.defer(long_process, process_result)
end

`rm -f xalala.csv ; touch xalala.csv`
`for i in $(seq 1 10000) ; do echo "$i;$(date +"%s")" >> xalala.csv ; done`

processor = Fiber.new do
  csv = CSV.open('xalala.csv', col_sep: ';')

  csv.each { |row| Fiber.yield row }

  csv.close
end

EM.run do
  while processor.alive?
    consume(processor.resume)
  end
  
  EventMachine.add_periodic_timer(1) do
    if EventMachine.defers_finished?
      EventMachine.stop
      true
    else
      false
    end
  end
end