objectsocket.rb

class ObjectSocket
  BLOCK_SIZE = 4096

  def self.pair
    Socket.pair(Socket::AF_UNIX, Socket::SOCK_STREAM, 0).map { |s| new(s) }
  end

  def initialize(socket)
    @socket = socket
  end

  def nonblock=(val)
    raise 'Cannot change blocking mode on active socket' if @unpacker
    @nonblock = val
  end

  def close
    @socket.close
  end

  def each_object(&block)
    if @nonblock
      @unpacker ||= MessagePack::Unpacker.new
      @unpacker.feed_each(@socket.read_nonblock(BLOCK_SIZE), &block)
    else
      @unpacker ||= MessagePack::Unpacker.new(@socket)
      @unpacker.each(&block)
    end
  end

  def <<(obj)
    @socket.syswrite(obj.to_msgpack)
    self # chainable
  end
end