lib/rainbows/ev_core.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 # base module for evented models like Rev and EventMachine
   4 module Rainbows::EvCore
   5   include Rainbows::Const
   6   include Rainbows::Response
   7   NULL_IO = Unicorn::HttpRequest::NULL_IO
   8   HttpParser = Rainbows::HttpParser
   9   autoload :CapInput, 'rainbows/ev_core/cap_input'
  10
  11   # Apps may return this Rack response: AsyncResponse = [ -1, {}, [] ]
  12   ASYNC_CALLBACK = "async.callback".freeze
  13
  14   def write_async_response(response)
  15     status, headers, body = response
  16     if alive = @hp.next?
  17       # we can't do HTTP keepalive without Content-Length or
  18       # "Transfer-Encoding: chunked", and the async.callback stuff
  19       # isn't Rack::Lint-compatible, so we have to enforce it here.
  20       headers = Rack::Utils::HeaderHash.new(headers) unless Hash === headers
  21       alive = headers.include?(Content_Length) ||
  22               !!(%r{\Achunked\z}i =~ headers[Transfer_Encoding])
  23     end
  24     write_response(status, headers, body, alive)
  25   end
  26
  27   ASYNC_CLOSE = "async.close".freeze
  28
  29   def post_init
  30     @hp = HttpParser.new
  31     @env = @hp.env
  32     @buf = @hp.buf
  33     @state = :headers # [ :body [ :trailers ] ] :app_call :close
  34   end
  35
  36   # graceful exit, like SIGQUIT
  37   def quit
  38     @state = :close
  39   end
  40
  41   def want_more
  42   end
  43
  44   def handle_error(e)
  45     msg = Rainbows::Error.response(e) and write(msg)
  46     ensure
  47       quit
  48   end
  49
  50   # returns whether to enable response chunking for autochunk models
  51   def stream_response_headers(status, headers, alive)
  52     headers = Rack::Utils::HeaderHash.new(headers) unless Hash === headers
  53     if headers.include?(Content_Length)
  54       rv = false
  55     else
  56       rv = !!(headers[Transfer_Encoding] =~ %r{\Achunked\z}i)
  57       rv = false if headers.delete('X-Rainbows-Autochunk') == 'no'
  58     end
  59     write_headers(status, headers, alive)
  60     rv
  61   end
  62
  63   def prepare_request_body
  64     # since we don't do streaming input, we have no choice but
  65     # to take over 100-continue handling from the Rack application
  66     if @env[HTTP_EXPECT] =~ /\A100-continue\z/i
  67       write(EXPECT_100_RESPONSE)
  68       @env.delete(HTTP_EXPECT)
  69     end
  70     @input = mkinput
  71     @hp.filter_body(@buf2 = "", @buf)
  72     @input << @buf2
  73     on_read("")
  74   end
  75
  76   # TeeInput doesn't map too well to this right now...
  77   def on_read(data)
  78     case @state
  79     when :headers
  80       @buf << data
  81       @hp.parse or return want_more
  82       @state = :body
  83       if 0 == @hp.content_length
  84         @input = NULL_IO
  85         app_call # common case
  86       else # nil or len > 0
  87         prepare_request_body
  88       end
  89     when :body
  90       if @hp.body_eof?
  91         if @hp.content_length
  92           @input.rewind
  93           app_call
  94         else
  95           @state = :trailers
  96           on_read(data)
  97         end
  98       elsif data.size > 0
  99         @hp.filter_body(@buf2, @buf << data)
 100         @input << @buf2
 101         on_read("")
 102       else
 103         want_more
 104       end
 105     when :trailers
 106       if @hp.trailers(@env, @buf << data)
 107         @input.rewind
 108         app_call
 109       else
 110         want_more
 111       end
 112     end
 113     rescue => e
 114       handle_error(e)
 115   end
 116
 117   def err_413(msg)
 118     write(Rainbows::Const::ERROR_413_RESPONSE)
 119     quit
 120     # zip back up the stack
 121     raise IOError, msg, []
 122   end
 123
 124   TmpIO = Unicorn::TmpIO
 125
 126   def io_for(bytes)
 127     bytes <= CBB ? StringIO.new("") : TmpIO.new
 128   end
 129
 130   def mkinput
 131     max = Rainbows.max_bytes
 132     len = @hp.content_length
 133     if len
 134       if max && (len > max)
 135         err_413("Content-Length too big: #{len} > #{max}")
 136       end
 137       io_for(len)
 138     else
 139       max ? CapInput.new(io_for(max), self, max) : TmpIO.new
 140     end
 141   end
 142
 143   def self.setup
 144     const_set :CBB, Unicorn::TeeInput.client_body_buffer_size
 145   end
 146 end