Threaded MARC Reader

marc-threadedreader.rb

require 'concurrent'
require 'thread'
require 'marc'

class MARC::ThreadedReader < MARC::Reader
  include Enumerable
  include Concurrent::Async
  
  def initialize(file, options={})
    super
    init_mutex # needed by Concurrent::Async
    @cache = SizedQueue.new(12)
  end
  
  def each_raw_record
    @handle.each_line(MARC::END_OF_RECORD) {|rr| yield rr}
  end
  
  def process_raw_record(rr)
    MARC::Reader.decode(rr, :forgiving=>true)
  end
  
  def fill_queue
    each_raw_record do |rr|
      @cache << Concurrent::Promise.new {rr}.then{|x| process_raw_record(x)}.execute
    end
    @cache << :end_of_records
  end
      
  def each
    self.async.fill_queue
    loop do
      r = @cache.pop
      break if r == :end_of_records
      yield r.value
    end
  end
end