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::EvCore
   8       G = Rainbows::G
   9
  10       def initialize(io)
  11         CONN[self] = false
  12         super(io)
  13         post_init
  14         @deferred_bodies = [] # for (fast) regular files only
  15       end
  16
  17       # queued, optional response bodies, it should only be unpollable "fast"
  18       # devices where read(2) is uninterruptable.  Unfortunately, NFS and ilk
  19       # are also part of this.  We'll also stick DeferredResponse bodies in
  20       # here to prevent connections from being closed on us.
  21       def defer_body(io, out_headers)
  22         @deferred_bodies << io
  23         schedule_write unless out_headers # triggers a write
  24       end
  25
  26       def timeout?
  27         @_write_buffer.empty? && @deferred_bodies.empty? and close.nil?
  28       end
  29
  30       def app_call
  31         begin
  32           KATO.delete(self)
  33           @env[RACK_INPUT] = @input
  34           @env[REMOTE_ADDR] = @remote_addr
  35           response = APP.call(@env.update(RACK_DEFAULTS))
  36           alive = @hp.keepalive? && G.alive
  37           out = [ alive ? CONN_ALIVE : CONN_CLOSE ] if @hp.headers?
  38
  39           DeferredResponse.write(self, response, out)
  40           if alive
  41             @env.clear
  42             @hp.reset
  43             @state = :headers
  44             # keepalive requests are always body-less, so @input is unchanged
  45             @hp.headers(@env, @buf) and next
  46             KATO[self] = Time.now
  47           else
  48             quit
  49           end
  50           return
  51         end while true
  52       end
  53
  54       def on_write_complete
  55         if body = @deferred_bodies[0]
  56           return if DeferredResponse === body
  57           begin
  58             begin
  59               write(body.sysread(CHUNK_SIZE))
  60             rescue EOFError # expected at file EOF
  61               @deferred_bodies.shift
  62               body.close
  63               close if :close == @state && @deferred_bodies.empty?
  64             end
  65           rescue => e
  66             handle_error(e)
  67           end
  68         else
  69           close if :close == @state
  70         end
  71       end
  72
  73       def on_close
  74         CONN.delete(self)
  75       end
  76
  77     end # module Client
  78   end # module Rev
  79 end # module Rainbows