1 # -*- encoding: binary -*-
3 class Rainbows::Rev::Client < ::Rev::IO
4 include Rainbows::EvCore
6 SF = Rainbows::StreamFile
7 CONN = Rainbows::Rev::CONN
8 KATO = Rainbows::Rev::KATO
9 DeferredResponse = Rainbows::Rev::DeferredResponse
10 DeferredChunkResponse = Rainbows::Rev::DeferredChunkResponse
20 enable unless enabled?
25 close if @deferred.nil? && @_write_buffer.empty?
28 # override the ::Rev::IO#write method try to write directly to the
29 # kernel socket buffers to avoid an extra userspace copy if
32 if @_write_buffer.empty?
34 case rv = @_io.kgio_trywrite(buf)
36 return enable_write_watcher
38 break # fall through to super(buf)
40 buf = rv # retry, skb could grow or been drained
43 return handle_error(e)
50 buf = @_io.kgio_tryread(16384)
58 rescue Errno::ECONNRESET
62 # queued, optional response bodies, it should only be unpollable "fast"
63 # devices where read(2) is uninterruptable. Unfortunately, NFS and ilk
64 # are also part of this. We'll also stick DeferredResponse bodies in
65 # here to prevent connections from being closed on us.
71 # allows enabling of write watcher even when read watcher is disabled
73 LOOP # this constant is set in when a worker starts
82 @deferred.nil? && @_write_buffer.empty? and close.nil?
85 # used for streaming sockets and pipes
86 def stream_response(status, headers, io, body)
87 c = stream_response_headers(status, headers) if headers
88 # we only want to attach to the Rev::Loop belonging to the
89 # main thread in Ruby 1.9
90 io = (c ? DeferredChunkResponse : DeferredResponse).new(io, self, body)
91 defer_body(io.attach(LOOP))
94 def rev_write_response(response, alive)
95 status, headers, body = response
96 headers = @hp.headers? ? HH.new(headers) : nil
98 headers[CONNECTION] = alive ? KEEP_ALIVE : CLOSE if headers
99 if body.respond_to?(:to_path)
100 io = body_to_io(body)
104 offset, count = 0, st.size
106 if range = make_range!(@env, status, headers)
107 status, offset, count = range
109 write(response_header(status, headers))
111 return defer_body(SF.new(offset, count, io, body))
112 elsif st.socket? || st.pipe?
113 return stream_response(status, headers, io, body)
115 # char or block device... WTF? fall through to body.each
117 write(response_header(status, headers)) if headers
118 write_body_each(self, body, nil)
123 @env[RACK_INPUT] = @input
124 @env[REMOTE_ADDR] = @_io.kgio_addr
125 response = APP.call(@env.update(RACK_DEFAULTS))
127 rev_write_response(response, alive = @hp.next? && G.alive)
128 return quit unless alive && :close != @state
133 def on_write_complete
135 when DeferredResponse then return
136 when NilClass # fall through
139 return rev_sendfile(@deferred)
140 rescue EOFError # expected at file EOF
147 close if @_write_buffer.empty?
152 KATO[self] = Time.now
164 if msg = Rainbows::Error.response(e)
165 @_io.kgio_trywrite(msg) rescue nil
174 when DeferredResponse, NilClass
179 G.server.logger.error("closing #@deferred: #{e}")