lib/rainbows/process_client.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 module Rainbows::ProcessClient
   4   include Rainbows::Response
   5   include Rainbows::Const
   6
   7   NULL_IO = Unicorn::HttpRequest::NULL_IO
   8   IC = Unicorn::HttpRequest.input_class
   9   Rainbows.config!(self, :client_header_buffer_size, :keepalive_timeout)
  10
  11   def read_expire
  12     Rainbows.now + KEEPALIVE_TIMEOUT
  13   end
  14
  15   # used for reading headers (respecting keepalive_timeout)
  16   def timed_read(buf)
  17     expire = nil
  18     begin
  19       case rv = kgio_tryread(CLIENT_HEADER_BUFFER_SIZE, buf)
  20       when :wait_readable
  21         return if expire && expire < Rainbows.now
  22         expire ||= read_expire
  23         kgio_wait_readable(KEEPALIVE_TIMEOUT)
  24       else
  25         return rv
  26       end
  27     end while true
  28   end
  29
  30   def process_loop
  31     @hp = hp = Rainbows::HttpParser.new
  32     kgio_read!(CLIENT_HEADER_BUFFER_SIZE, buf = hp.buf) or return
  33
  34     begin # loop
  35       until env = hp.parse
  36         timed_read(buf2 ||= "") or return
  37         buf << buf2
  38       end
  39
  40       set_input(env, hp)
  41       env['REMOTE_ADDR'] = kgio_addr
  42       hp.hijack_setup(to_io)
  43       status, headers, body = APP.call(env.merge!(RACK_DEFAULTS))
  44
  45       if 100 == status.to_i
  46         write("HTTP/1.1 100 Continue\r\n\r\n".freeze)
  47         env.delete('HTTP_EXPECT'.freeze)
  48         status, headers, body = APP.call(env)
  49       end
  50       return if hp.hijacked?
  51       write_response(status, headers, body, alive = hp.next?) or return
  52     end while alive
  53   # if we get any error, try to write something back to the client
  54   # assuming we haven't closed the socket, but don't get hung up
  55   # if the socket is already closed or broken.  We'll always ensure
  56   # the socket is closed at the end of this function
  57   rescue => e
  58     handle_error(e)
  59   ensure
  60     close unless closed? || hp.hijacked?
  61   end
  62
  63   def handle_error(e)
  64     Rainbows::Error.write(self, e)
  65   end
  66
  67   def set_input(env, hp)
  68     env['rack.input'] = 0 == hp.content_length ? NULL_IO : IC.new(self, hp)
  69   end
  70
  71   def process_pipeline(env, hp)
  72     begin
  73       set_input(env, hp)
  74       env['REMOTE_ADDR'] = kgio_addr
  75       hp.hijack_setup(to_io)
  76       status, headers, body = APP.call(env.merge!(RACK_DEFAULTS))
  77       if 100 == status.to_i
  78         write("HTTP/1.1 100 Continue\r\n\r\n".freeze)
  79         env.delete('HTTP_EXPECT'.freeze)
  80         status, headers, body = APP.call(env)
  81       end
  82       return if hp.hijacked?
  83       write_response(status, headers, body, alive = hp.next?) or return
  84     end while alive && pipeline_ready(hp)
  85     alive or close
  86     rescue => e
  87       handle_error(e)
  88   end
  89
  90   # override this in subclass/module
  91   def pipeline_ready(hp)
  92   end
  93 end