pipe.rb

pipe.rb

class Pipe
  DestinyNotYetChosenException = Class.new(Exception)
  CannotWriteToReaderException = Class.new(Exception)
  CannotReadFromWriterException = Class.new(Exception)

  def initialize
    @readpipe, @writepipe = IO.pipe
  end

  def destiny_chosen?
    !(@writer.nil? && @reader.nil?)
  end

  def writer!
    if !destiny_chosen?
      @writer = true
      @readpipe.close
    end
  end

  def reader!
    if !destiny_chosen?
      @reader = true
      @writepipe.close
    end
  end

  def writer?
    !!@writer
  end

  def reader?
    !!@reader
  end

  def <<(obj)
    raise CannotWriteToReader if reader?

    # Force writer
    writer!

    Marshal.dump(obj, @writepipe)
  end

  def close
    self << :close
  end

  def load
    raise CannotReadFromWriterException if writer?

    # Force writer
    writer!

    Marshal.load(@readpipe)
  end

  def raw
    raise DestinyNotYetChosenException if !destiny_chosen?

    reader? ? @readpipe : @writepipe
  end

  class << self
    def receive(*pipes)
      @pipes = [*pipes] 
  
      # Force all pipes to be readers
      @pipes.each {|p| p.reader! }
  
      loop do
        ready = IO.select(@pipes.map {|p| p.raw }, nil, nil, nil)
  
        # Should always be ready, but if not, somehow it
        # timed out. Throw error
        if !ready
          raise Exception
        else
          ready_pipes = ready[0].select {|p|
            !p.eof?
          }.map {|raw_pipe|
            @pipes.find {|p| p.raw == raw_pipe }
          }
  
          ready_pipes.each do |pipe|
            obj = pipe.load
  
            if obj == :close
              @pipes = @pipes.reject {|p| p == pipe }
              return if @pipes.empty?
            else
              yield(obj)
            end
          end
        end
      end
    end
  end
end

p1 = Pipe.new
p2 = Pipe.new
printer = Pipe.new
if pid = fork
  Pipe.receive(p1, p2) do |obj|
    case obj
    when String
      printer << "String that says #{obj.inspect}"
    when Symbol
      printer << "Symbol that says #{obj.inspect}"
    when Hash
      printer << "Hash that says #{obj.inspect}"
    end
  end

  printer << "Pipe was closed!"

  printer.close
else
  if fork
    p1 << "Yo Parent!"
    p1 << :yo
    p1 << {:a => 1, :b => 2}

    p1.close
  else
    if fork
      p2 << "Yo Grandparent!"
      p2 << :yo_to
      p2 << {:x => 0, :z => 5}

      p2.close
    else
      Pipe.receive(printer) do |string|
        puts string
      end
    end
  end
end