lib/rainbows/rev/client.rb

   1 # -*- encoding: binary -*-
   2 require 'rainbows/ev_core'
   3 module Rainbows
   4   module Rev
   5
   6     class Client < ::Rev::IO
   7       include Rainbows::ByteSlice
   8       include Rainbows::EvCore
   9       include Rainbows::HttpResponse
  10       G = Rainbows::G
  11       HH = Rack::Utils::HeaderHash
  12
  13       def initialize(io)
  14         CONN[self] = false
  15         super(io)
  16         post_init
  17         @deferred_bodies = [] # for (fast) regular files only
  18       end
  19
  20       def quit
  21         super
  22         close if @deferred_bodies.empty? && @_write_buffer.empty?
  23       end
  24
  25       # override the ::Rev::IO#write method try to write directly to the
  26       # kernel socket buffers to avoid an extra userspace copy if
  27       # possible.
  28       def write(buf)
  29         if @_write_buffer.empty?
  30           begin
  31             w = @_io.write_nonblock(buf)
  32             if w == Rack::Utils.bytesize(buf)
  33               return on_write_complete
  34             end
  35             # we never care for the return value, but yes, we may return
  36             # a "fake" short write from super(buf) if anybody cares.
  37             buf = byte_slice(buf, w..-1)
  38           rescue Errno::EAGAIN
  39             break # fall through to super(buf)
  40           rescue
  41             return close
  42           end while true
  43         end
  44         super(buf)
  45       end
  46
  47       # queued, optional response bodies, it should only be unpollable "fast"
  48       # devices where read(2) is uninterruptable.  Unfortunately, NFS and ilk
  49       # are also part of this.  We'll also stick DeferredResponse bodies in
  50       # here to prevent connections from being closed on us.
  51       def defer_body(io, out_headers)
  52         @deferred_bodies << io
  53         schedule_write unless out_headers # triggers a write
  54       end
  55
  56       def timeout?
  57         @_write_buffer.empty? && @deferred_bodies.empty? and close.nil?
  58       end
  59
  60       def rev_write_response(response, out)
  61         status, headers, body = response
  62
  63         body.respond_to?(:to_path) or
  64           return write_response(self, response, out)
  65
  66         headers = HH.new(headers)
  67         io = body_to_io(body)
  68         st = io.stat
  69
  70         if st.socket? || st.pipe?
  71           do_chunk = !!(headers['Transfer-Encoding'] =~ %r{\Achunked\z}i)
  72           do_chunk = false if headers.delete('X-Rainbows-Autochunk') == 'no'
  73           # too tricky to support keepalive/pipelining when a response can
  74           # take an indeterminate amount of time here.
  75           if out.nil?
  76             do_chunk = false
  77           else
  78             out[0] = CONN_CLOSE
  79           end
  80
  81           # we only want to attach to the Rev::Loop belonging to the
  82           # main thread in Ruby 1.9
  83           io = DeferredResponse.new(io, self, do_chunk, body).
  84                                     attach(Server::LOOP)
  85         elsif st.file?
  86           headers.delete('Transfer-Encoding')
  87           headers['Content-Length'] ||= st.size.to_s
  88           io = to_sendfile(io)
  89         else # char/block device, directory, whatever... nobody cares
  90           return write_response(self, response, out)
  91         end
  92         defer_body(io, out)
  93         write_header(self, response, out)
  94       end
  95
  96       def app_call
  97         begin
  98           KATO.delete(self)
  99           @env[RACK_INPUT] = @input
 100           @env[REMOTE_ADDR] = @remote_addr
 101           response = APP.call(@env.update(RACK_DEFAULTS))
 102           alive = @hp.keepalive? && G.alive
 103           out = [ alive ? CONN_ALIVE : CONN_CLOSE ] if @hp.headers?
 104
 105           rev_write_response(response, out)
 106           if alive
 107             @env.clear
 108             @hp.reset
 109             @state = :headers
 110             # keepalive requests are always body-less, so @input is unchanged
 111             @hp.headers(@env, @buf) and next
 112             KATO[self] = Time.now
 113           else
 114             quit
 115           end
 116           return
 117         end while true
 118       end
 119
 120       def on_write_complete
 121         if body = @deferred_bodies[0]
 122           # no socket or pipes, body must be a regular file to continue here
 123           return if DeferredResponse === body
 124
 125           begin
 126             rev_sendfile(body)
 127           rescue EOFError # expected at file EOF
 128             @deferred_bodies.shift
 129             body.close
 130             close if :close == @state && @deferred_bodies.empty?
 131           rescue => e
 132             handle_error(e)
 133           end
 134         else
 135           close if :close == @state
 136         end
 137       end
 138
 139       def on_close
 140         while f = @deferred_bodies.shift
 141           DeferredResponse === f or f.close
 142         end
 143         CONN.delete(self)
 144       end
 145
 146     end # module Client
 147   end # module Rev
 148 end # module Rainbows