transaction_protocol.rb

require 'rubygems'
require 'eventmachine'
require 'delegate'

module TransactionProtocol 
  class Writer < EM::Queue
    class << self
      @files = []
      attr_accessor :files
    end

    attr_reader :filename

    def initialize(filename, interval=60)
      raise "File is already associated with a transaction" if self.class.files.include?(filename)

      self.class.files.push(filename)

      @filename = filename
      @timer = EM::PeriodicTimer.new(interval) do 
        self.flush
      end
    end

    def flush
      return if self.empty?

      m = Marshal.new(@filename)
      m.push(self.pop) until self.empty? 
      m.write
    end

    def close
      self.flush
      @timer.cancel
      self.class.files.delete(@filename)
    end

  end

  class Marshal < DelegateClass(Array)
    def initialize(filename)
      @obj = ::Marshal.load(filename)
      raise "Not an array" unless @obj.kind_of?(Array)
      super(@obj)
    end

    def reload
      @obj = ::Marshal.load(filename)
    end

    def write
      ::Marshal.dump(filename)
    end
  end
end