lib/rainbows/revactor/proxy.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 # Generic IO wrapper for proxying pipe and socket objects
   4 # this behaves more like Rainbows::Fiber::IO than anything,
   5 # making it highly suitable for proxying data from pipes/sockets
   6 class Rainbows::Revactor::Proxy < Rev::IO
   7   def initialize(io)
   8     @receiver = Actor.current
   9     super(io)
  10     attach(Rev::Loop.default)
  11   end
  12
  13   def close
  14     if @_io
  15       super
  16       @_io = nil
  17     end
  18   end
  19
  20   def each
  21     # when yield-ing, Revactor::TCP#write may raise EOFError
  22     # (instead of Errno::EPIPE), so we need to limit the rescue
  23     # to just readpartial and let EOFErrors during yield bubble up
  24     begin
  25       buf = readpartial(INPUT_SIZE)
  26     rescue EOFError
  27       break
  28     end while yield(buf) || true
  29   end
  30
  31   # this may return more than the specified length, Rainbows! won't care...
  32   def readpartial(length)
  33     @receiver = Actor.current
  34     enable if attached? && ! enabled?
  35
  36     Actor.receive do |filter|
  37       filter.when(T[:rainbows_io_input, self]) do |_, _, data|
  38         return data
  39       end
  40
  41       filter.when(T[:rainbows_io_closed, self]) do
  42         raise EOFError, "connection closed"
  43       end
  44     end
  45   end
  46
  47   def on_close
  48     @receiver << T[:rainbows_io_closed, self]
  49   end
  50
  51   def on_read(data)
  52     @receiver << T[:rainbows_io_input, self, data ]
  53     disable
  54   end
  55 end