1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
| | # -*- encoding: binary -*-
module Rainbows
module Fiber
# A partially complete IO wrapper, this exports an IO.select()-able
# #to_io method and gives users the illusion of a synchronous
# interface that yields away from the current Fiber whenever
# the underlying IO object cannot read or write
class IO < Struct.new(:to_io, :f)
# for wrapping output response bodies
def each(&block)
begin
yield readpartial(16384)
rescue EOFError
break
end while true
self
end
def close
RD.delete(self)
WR.delete(self)
to_io.close unless to_io.closed?
end
def wait_readable
RD[self] = false
::Fiber.yield
RD.delete(self)
end
def wait_writable
WR[self] = false
::Fiber.yield
WR.delete(self)
end
def write(buf)
begin
(w = to_io.write_nonblock(buf)) == buf.size and return
buf = buf[w..-1]
rescue Errno::EAGAIN
wait_writable
retry
end while true
end
# used for reading headers (respecting keepalive_timeout)
def read_timeout
expire = nil
begin
to_io.read_nonblock(16384)
rescue Errno::EAGAIN
return if expire && expire < Time.now
expire ||= Time.now + G.kato
wait_readable
retry
end
end
def readpartial(length, buf = "")
begin
to_io.read_nonblock(length, buf)
rescue Errno::EAGAIN
wait_readable
retry
end
end
end
end
end
|