easy ipc

easy_ipc.rb

require 'set'

module SpecialMessages
  module Stop; end
  module Ack; end
end

class WorkerFactory
  def initialize
    @set = Set.new
  end

  def spawn(&block)
    worker = ProcessWorker.new(&block)
    @set << worker
    worker.start
    worker
  end

  def kill_all
    @set.map(&:kill)
  end

  def stop_all
    @set.map(&:stop)
  end

end


class ProcessWorker

  def initialize(parent_to_child = IO.pipe, child_to_parent = IO.pipe, &block)
    @child_read, @parent_write = parent_to_child
    @parent_read, @child_write = child_to_parent
    @parent_io = [@parent_read, @parent_write]
    @child_io = [@child_read, @child_write]
    @creation_block = block
    @stopped = false
  end
  
  def start
    @child_pid = Process.fork do
      @child = true
      @creation_block.call(self)
    ensure
      own_io(@child).map(&:close)
    end
  ensure
    @child = false
    own_io(!@child).map(&:close)
  end

  def on_received_msg(&block)
    while (msg = receive) && !@stopped
      unless special_message(msg)
        warn("#{@stopped} #{msg}")
        response = block.call(msg)
        send(response)
      end
    end
    #@stopped = false
  end

  def send(msg)
    begin
      Marshal.dump(msg, own_write_io)
    rescue Errno::EPIPE
      return # parent thread already dead
    end
  end

  def receive
    unless own_read_io.eof?
      Marshal.load(own_read_io)
    end
  end

  def stop
    if @child
      @stopped = true
    else
      send(SpecialMessages::Stop)
    end
  end

  def kill
    Process.kill(:KILL, @child_pid)
  end

  private

    def special_message(msg)
      case msg
      when SpecialMessages::Ack
        true
      when SpecialMessages::Stop
        stop
        send(SpecialMessages::Ack)
        true
      else
        false
      end 
    end

    def own_io(child = @child)
      (child && @child_io) || @parent_io
    end
  
    def own_write_io
      own_io[1]
    end

    def own_read_io
      own_io[0]
    end
end


factory = WorkerFactory.new
w = factory.spawn do |worker|
  worker.on_received_msg do |msg|
    case msg
    when Integer
      worker.send("thats an int")
      worker.on_received_msg do |msg|
        "kek #{msg}"
      end
    else
      "something else #{msg}"
    end
  end
end


require 'pry'; binding.pry