lib/rainbows/stream_response_epoll.rb

   1 # -*- encoding: binary -*-
   2 require "sleepy_penguin"
   3 require "raindrops"
   4
   5 # Like Unicorn itself, this concurrency model is only intended for use
   6 # behind nginx and completely unsupported otherwise.  Even further from
   7 # Unicorn, this isn't even a good idea with normal LAN clients, only nginx!
   8 #
   9 # It does NOT require a thread-safe Rack application at any point, but
  10 # allows streaming data asynchronously via nginx (using the
  11 # "X-Accel-Buffering: no" header to disable buffering).
  12 #
  13 # Unlike Rainbows::Base, this does NOT support persistent
  14 # connections or pipelining.  All \Rainbows! specific configuration
  15 # options are ignored (except Rainbows::Configurator#use).
  16 #
  17 # === RubyGem Requirements
  18 #
  19 # * raindrops 0.6.0 or later
  20 # * sleepy_penguin 3.0.1 or later
  21 module Rainbows::StreamResponseEpoll
  22   # :stopdoc:
  23   autoload :Client, "rainbows/stream_response_epoll/client"
  24
  25   def http_response_write(socket, status, headers, body)
  26     hijack = ep_client = false
  27
  28     if headers
  29       # don't set extra headers here, this is only intended for
  30       # consuming by nginx.
  31       code = status.to_i
  32       msg = Rack::Utils::HTTP_STATUS_CODES[code]
  33       buf = "HTTP/1.0 #{msg ? %Q(#{code} #{msg}) : status}\r\n"
  34       headers.each do |key, value|
  35         case key
  36         when "rack.hijack"
  37           hijack = value
  38           body = nil # ensure we do not close body
  39         else
  40           if /\n/ =~ value
  41             # avoiding blank, key-only cookies with /\n+/
  42             value.split(/\n+/).each { |v| buf << "#{key}: #{v}\r\n" }
  43           else
  44             buf << "#{key}: #{value}\r\n"
  45           end
  46         end
  47       end
  48       buf << "X-Accel-Buffering: no\r\n\r\n".freeze
  49
  50       case rv = socket.kgio_trywrite(buf)
  51       when nil then break
  52       when String # retry, socket buffer may grow
  53         buf = rv
  54       when :wait_writable
  55         ep_client = Client.new(socket, buf)
  56         if hijack
  57           ep_client.hijack(hijack)
  58         else
  59           body.each { |chunk| ep_client.write(chunk) }
  60           ep_client.close
  61         end
  62         # body is nil on hijack, in which case ep_client is never closed by us
  63         return
  64       end while true
  65     end
  66
  67     if hijack
  68       hijack.call(socket)
  69       return
  70     end
  71
  72     body.each do |chunk|
  73       if ep_client
  74         ep_client.write(chunk)
  75       else
  76         case rv = socket.kgio_trywrite(chunk)
  77         when nil then break
  78         when String # retry, socket buffer may grow
  79           chunk = rv
  80         when :wait_writable
  81           ep_client = Client.new(socket, chunk)
  82           break
  83         end while true
  84       end
  85     end
  86   ensure
  87     return if hijack
  88     body.respond_to?(:close) and body.close
  89     if ep_client
  90       ep_client.close
  91     else
  92       socket.shutdown
  93       socket.close
  94     end
  95   end
  96
  97   # once a client is accepted, it is processed in its entirety here
  98   # in 3 easy steps: read request, call app, write app response
  99   def process_client(client)
 100     status, headers, body = @app.call(env = @request.read(client))
 101
 102     if 100 == status.to_i
 103       client.write("HTTP/1.1 100 Continue\r\n\r\n".freeze)
 104       env.delete('HTTP_EXPECT'.freeze)
 105       status, headers, body = @app.call(env)
 106     end
 107     @request.headers? or headers = nil
 108     return if @request.hijacked?
 109     http_response_write(client, status, headers, body)
 110   rescue => e
 111     handle_error(client, e)
 112   end
 113
 114   # :startdoc:
 115 end