finalizer or at_exit

zeromq.rb

require 'cabin'
require 'ffi-rzmq'

class Cabin::Outputs::ZeroMQ
  DEFAULTS = {
    :topology => "pushpull",
    :hwm => 0, # default: no limit
    :linger => -1 # default: wait until all messages are sent
  }

  CONTEXT = ZMQ::Context.new

  attr_reader :socket

  def initialize(addresses, options={})
    options = DEFAULTS.merge(options)

    case options[:topology]
    when "pushpull"
      socket_type = ZMQ::PUSH
    when "pubsub"
      socket_type = ZMQ::PUB
    when Fixnum
      socket_type = options[:topology]
    end # case socket_type

    @socket = CONTEXT.socket(socket_type)
    
    Array(addresses).each do |address|
      error_check(@socket.connect(address), "connecting to #{address}")
    end

    error_check(@socket.setsockopt(ZMQ::LINGER, options[:linger]), "while setting ZMQ::LINGER to #{options[:linger]}")
    error_check(@socket.setsockopt(ZMQ::HWM, options[:hwm]), "while setting ZMQ::HWM to #{options[:hwm]}")

    #define_finalizer
  end

  def <<(event)
    @socket.send_string(event.to_json)
  end

  private
  def error_check(rc, doing)
    unless ZMQ::Util.resultcode_ok?(rc)
      raise "ZeroMQ Error while #{doing}"
    end
  end

  #TODO
  at_exit do
    @socket.close unless @socket.nil?
  end

  # This causes the following message on exit:
  # File exists (epoll.cpp:69)
  # [1]    26175 abort      bundle exec irb
  # def define_finalizer
  #   ObjectSpace.define_finalizer(self, self.class.finalize(@socket))
  # end

  # def self.finalize(socket)
  #   Proc.new { socket.close unless socket.nil? }
  # end
end