thread pipe

a.rb

require 'thread'

class Thread
  class Pipe
    class Queue < ::Queue
      attr_accessor :thread_id
    end

    def initialize
      @queues = [Queue.new, Queue.new]
    end

    def thread_id
      Thread.current.object_id
    end

    def reserve_write_queue!
      Thread.exclusive do
        @queues.each do |queue|
          next if queue.thread_id
          queue.thread_id = thread_id
          return queue
        end
      end
    end

    def write_queue
      @queues.detect{|q| q.thread_id == thread_id} || reserve_write_queue!
    end

    def read_queue
      @queues.detect{|q| q != write_queue}
    end

    def write(object)
      write_queue.push(object)
    end

    def read
      read_queue.pop
    end
  end
end




if $0 = __FILE__

  STDOUT.sync = true

  pipe = Thread::Pipe.new

  a = Thread.new do
    Thread.current.abort_on_exception = true
    pipe.write 'hai from a!'
    pipe.write 'hai from a 2!'
    puts pipe.read
  end

  b = Thread.new do
    Thread.current.abort_on_exception = true
    pipe.write 'hai from b!'
    puts pipe.read
    puts pipe.read
  end

  a.join
  b.join

end