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
| | # -*- encoding: binary -*-
# :enddoc:
class Rainbows::StreamResponseEpoll::Client
OUT = SleepyPenguin::Epoll::OUT
N = Raindrops.new(1)
EP = SleepyPenguin::Epoll.new
timeout = Rainbows.server.timeout
thr = Thread.new do
begin
EP.wait(nil, timeout) { |_,client| client.epoll_run }
rescue Errno::EINTR
rescue => e
Rainbows::Error.listen_loop(e)
end while Rainbows.alive || N[0] > 0
end
Rainbows.at_quit { thr.join(timeout) }
attr_reader :to_io
def initialize(io, unwritten)
@finish = false
@to_io = io
@wr_queue = [ unwritten.dup ]
EP.set(self, OUT)
end
def write(str)
@wr_queue << str.dup
end
def close
@finish = true
end
def hijack(hijack)
@finish = hijack
end
def epoll_run
return if @to_io.closed?
buf = @wr_queue.shift or return on_write_complete
case rv = @to_io.kgio_trywrite(buf)
when nil
buf = @wr_queue.shift or return on_write_complete
when String # retry, socket buffer may grow
buf = rv
when :wait_writable
return @wr_queue.unshift(buf)
end while true
rescue => err
@to_io.close
N.decr(0, 1)
end
def on_write_complete
if true == @finish
@to_io.shutdown
@to_io.close
N.decr(0, 1)
elsif @finish.respond_to?(:call) # hijacked
EP.delete(self)
N.decr(0, 1)
@finish.call(@to_io)
end
end
end
|