lib/rainbows/fiber/io.rb

   1 # -*- encoding: binary -*-
   2 module Rainbows
   3   module Fiber
   4
   5     # A partially complete IO wrapper, this exports an IO.select()-able
   6     # #to_io method and gives users the illusion of a synchronous
   7     # interface that yields away from the current Fiber whenever
   8     # the underlying IO object cannot read or write
   9     #
  10     # TODO: subclass off IO and include Kgio::SocketMethods instead
  11     class IO < Struct.new(:to_io, :f)
  12       # :stopdoc:
  13       LOCALHOST = Kgio::LOCALHOST
  14
  15       # needed to write errors with
  16       def write_nonblock(buf)
  17         to_io.write_nonblock(buf)
  18       end
  19
  20       def kgio_addr
  21         to_io.kgio_addr
  22       end
  23
  24       # for wrapping output response bodies
  25       def each(&block)
  26         if buf = readpartial(16384)
  27           yield buf
  28           yield buf while readpartial(16384, buf)
  29         end
  30         rescue EOFError
  31         self
  32       end
  33
  34       def close
  35         fileno = to_io.fileno
  36         RD[fileno] = WR[fileno] = nil
  37         to_io.close unless to_io.closed?
  38       end
  39
  40       def closed?
  41         to_io.closed?
  42       end
  43
  44       def wait_readable
  45         fileno = to_io.fileno
  46         RD[fileno] = self
  47         ::Fiber.yield
  48         RD[fileno] = nil
  49       end
  50
  51       def wait_writable
  52         fileno = to_io.fileno
  53         WR[fileno] = self
  54         ::Fiber.yield
  55         WR[fileno] = nil
  56       end
  57
  58       def write(buf)
  59         begin
  60           case rv = to_io.kgio_trywrite(buf)
  61           when nil
  62             return
  63           when String
  64             buf = rv
  65           when Kgio::WaitWritable
  66             wait_writable
  67           end
  68         end while true
  69       end
  70
  71       # used for reading headers (respecting keepalive_timeout)
  72       def read_timeout
  73         expire = nil
  74         begin
  75           to_io.read_nonblock(16384)
  76         rescue Errno::EAGAIN
  77           return if expire && expire < Time.now
  78           expire ||= Time.now + G.kato
  79           wait_readable
  80           retry
  81         end
  82       end
  83
  84       def readpartial(length, buf = "")
  85         if to_io.respond_to?(:kgio_tryread)
  86           # TODO: use kgio_read!
  87           begin
  88             rv = to_io.kgio_tryread(length, buf)
  89             case rv
  90             when nil
  91               raise EOFError, "end of file reached", []
  92             when Kgio::WaitReadable
  93               wait_readable
  94             else
  95               return rv
  96             end
  97           end while true
  98         else
  99           begin
 100             to_io.read_nonblock(length, buf)
 101           rescue Errno::EAGAIN
 102             wait_readable
 103             retry
 104           end
 105         end
 106       end
 107
 108       def kgio_read(*args)
 109         to_io.kgio_read(*args)
 110       end
 111
 112       def kgio_read!(*args)
 113         to_io.kgio_read!(*args)
 114       end
 115     end
 116   end
 117 end