1 # -*- encoding: binary -*-
4 module Rainbows::Epoll::Client
6 include Rainbows::EvCore
7 APP = Rainbows.server.app
8 Server = Rainbows::Epoll::Server
9 IN = SleepyPenguin::Epoll::IN | SleepyPenguin::Epoll::ET
10 OUT = SleepyPenguin::Epoll::OUT | SleepyPenguin::Epoll::ET
12 KATO.compare_by_identity if KATO.respond_to?(:compare_by_identity)
13 Rainbows.at_quit { KATO.each_key { |k| k.timeout! }.clear }
14 Rainbows.config!(self, :keepalive_timeout)
17 @@last_expire = Time.now
20 return if ((now = Time.now) - @@last_expire) < 1.0
21 if (ot = KEEPALIVE_TIMEOUT) >= 0
23 KATO.delete_if { |client, time| time < ot and client.timeout! }
30 EP.wait(nil, 1000) { |_, obj| obj.epoll_run }
31 while obj = ReRun.shift
37 Rainbows::Error.listen_loop(e)
38 end while Rainbows.tick || Server.nr > 0
43 @wr_queue = [] # may contain String, ResponsePipe, and StreamFile objects
51 case rv = kgio_tryread(CLIENT_HEADER_BUFFER_SIZE, RBUF)
54 return if @wr_queue[0] || closed?
55 return hijacked if @hp.hijacked?
57 KATO[self] = @@last_expire if :headers == @state
58 return EP.set(self, IN)
61 end until :close == @state
63 rescue Errno::ECONNRESET
68 def app_call input # called by on_read()
69 @env[RACK_INPUT] = input
70 @env[REMOTE_ADDR] = kgio_addr
71 @hp.hijack_setup(@env, self)
72 status, headers, body = APP.call(@env.merge!(RACK_DEFAULTS))
73 return hijacked if @hp.hijacked?
74 ev_write_response(status, headers, body, @hp.next?)
77 def write_response_path(status, headers, body, alive)
82 defer_file(status, headers, body, alive, io, st)
83 elsif st.socket? || st.pipe?
84 chunk = stream_response_headers(status, headers, alive, body)
85 return hijacked if nil == chunk
86 stream_response_body(body, io, chunk)
88 # char or block device... WTF?
89 write_response(status, headers, body, alive)
93 # used for streaming sockets and pipes
94 def stream_response_body(body, io, chunk)
95 pipe = (chunk ? Rainbows::Epoll::ResponseChunkPipe :
96 Rainbows::Epoll::ResponsePipe).new(io, self, body)
97 return @wr_queue << pipe if @wr_queue[0]
98 stream_pipe(pipe) or return
99 @wr_queue[0] or @wr_queue << Z
102 def ev_write_response(status, headers, body, alive)
103 @state = alive ? :headers : :close
104 if body.respond_to?(:to_path)
105 write_response_path(status, headers, body, alive)
107 write_response(status, headers, body, alive)
109 return hijacked if @hp.hijacked?
110 # try to read more if we didn't have to buffer writes
111 next_request if alive && 0 == @wr_queue.size
116 Server.decr # no other place to do this
125 # pipelined request (already in buffer)
127 return if @wr_queue[0] || closed?
128 return hijacked if @hp.hijacked?
129 close if :close == @state
146 def on_deferred_write_complete
147 :close == @state and return close
152 msg = Rainbows::Error.response(e) and kgio_trywrite(msg) rescue nil
157 def write_deferred(obj)
158 Rainbows::StreamFile === obj ? stream_file(obj) : stream_pipe(obj)
161 # writes until our write buffer is empty or we block
162 # returns true if we're done writing everything
164 obj = @wr_queue.shift
166 case rv = String === obj ? kgio_trywrite(obj) : write_deferred(obj)
168 obj = @wr_queue.shift or return on_deferred_write_complete
171 when :wait_writable # Strings and StreamFiles only
172 @wr_queue.unshift(obj)
184 case rv = kgio_trywrite(buf)
190 @wr_queue << buf.dup # >3-word 1.9 strings are copy-on-write
191 return EP.set(self, OUT)
194 @wr_queue << buf.dup # >3-word 1.9 strings are copy-on-write
198 @wr_queue.each { |x| x.respond_to?(:close) and x.close rescue nil }
213 # Rack apps should not hijack here, but they may...
214 def defer_file(status, headers, body, alive, io, st)
215 if r = sendfile_range(status, headers)
216 status, headers, range = r
217 write_headers(status, headers, alive, body) or return hijacked
218 range and defer_file_stream(range[0], range[1], io, body)
220 write_headers(status, headers, alive, body) or return hijacked
221 defer_file_stream(0, st.size, io, body)
225 # returns +nil+ on EOF, :wait_writable if the client blocks
226 def stream_file(sf) # +sf+ is a Rainbows::StreamFile object
227 case n = trysendfile(sf, sf.offset, sf.count)
230 0 == (sf.count -= n) and return sf.close
232 return n # :wait_writable or nil
239 def defer_file_stream(offset, count, io, body)
240 sf = Rainbows::StreamFile.new(offset, count, io, body)
242 stream_file(sf) or return
248 # this alternates between a push and pull model from the pipe -> client
249 # to avoid having too much data in userspace on either end.
250 def stream_pipe(pipe)
251 case buf = pipe.tryread
255 # client is blocked on write, client will pull from pipe later
261 # continue looping...
263 # pipe blocked on read, let the pipe push to the client in the future
268 return pipe.close # nil