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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
| | # -*- encoding: binary -*-
require 'rev'
require 'rainbows/fiber'
require 'rainbows/fiber/io'
module Rainbows::Fiber
module Rev
G = Rainbows::G
# keep-alive timeout class
class Kato < ::Rev::TimerWatcher
def initialize
@watch = []
super(1, true)
end
def <<(fiber)
@watch << fiber
enable unless enabled?
end
def on_timer
@watch.uniq!
while f = @watch.shift
f.resume if f.alive?
end
disable
end
end
class Heartbeat < ::Rev::TimerWatcher
def on_timer
exit if (! G.tick && G.cur <= 0)
end
end
class Sleeper < ::Rev::TimerWatcher
def initialize(seconds)
@f = ::Fiber.current
super(seconds, false)
attach(::Rev::Loop.default)
::Fiber.yield
end
def on_timer
@f.resume
end
end
class Server < ::Rev::IOWatcher
include Unicorn
include Rainbows
include Rainbows::Const
FIO = Rainbows::Fiber::IO
def to_io
@io
end
def initialize(io)
@io = io
super(self, :r)
end
def close
detach if attached?
@io.close
end
def on_readable
return if G.cur >= MAX
c = Rainbows.accept(@io) and ::Fiber.new { process(c) }.resume
end
def process(io)
G.cur += 1
client = FIO.new(io, ::Fiber.current)
buf = client.read_timeout or return
hp = HttpParser.new
env = {}
alive = true
remote_addr = TCPSocket === io ? io.peeraddr.last : LOCALHOST
begin # loop
buf << (client.read_timeout or return) until hp.headers(env, buf)
env[CLIENT_IO] = client
env[RACK_INPUT] = 0 == hp.content_length ?
HttpRequest::NULL_IO : TeeInput.new(client, env, hp, buf)
env[REMOTE_ADDR] = remote_addr
response = APP.call(env.update(RACK_DEFAULTS))
if 100 == response.first.to_i
client.write(EXPECT_100_RESPONSE)
env.delete(HTTP_EXPECT)
response = APP.call(env)
end
alive = hp.keepalive? && G.alive
out = [ alive ? CONN_ALIVE : CONN_CLOSE ] if hp.headers?
HttpResponse.write(client, response, out)
end while alive and hp.reset.nil? and env.clear
rescue => e
Error.write(io, e)
ensure
G.cur -= 1
client.close
end
end
end
class IO # see rainbows/fiber/io for original definition
class Watcher < ::Rev::IOWatcher
def initialize(fio, flag)
@fiber = fio.f
super(fio, flag)
attach(::Rev::Loop.default)
end
def on_readable
@fiber.resume
end
alias on_writable on_readable
end
undef_method :wait_readable
undef_method :wait_writable
undef_method :close
def initialize(*args)
super
@r = @w = false
end
def close
@w.detach if @w
@r.detach if @r
@r = @w = false
to_io.close unless to_io.closed?
end
def wait_writable
@w ||= Watcher.new(self, :w)
@w.enable unless @w.enabled?
::Fiber.yield
@w.disable
end
def wait_readable
@r ||= Watcher.new(self, :r)
@r.enable unless @r.enabled?
KATO << f
::Fiber.yield
@r.disable
end
end
end
|