lib/rainbows/reverse_proxy/event_machine.rb

   1 # -*- encoding: binary -*-
   2 # :enddoc:
   3 # TODO: handle large responses without having it all in memory
   4 module Rainbows::ReverseProxy::EventMachine
   5   class Backend < EM::Connection
   6     include Rainbows::ReverseProxy::EvClient # provides receive_data
   7
   8     # +addr+ is a packed sockaddr, so it can be either a UNIX or TCP socket
   9     def initialize(env)
  10       @env = env
  11       @rbuf = ""
  12       @parser = Kcar::Parser.new
  13       @response = @body = nil
  14       @headers = Rack::Utils::HeaderHash.new
  15     end
  16
  17     # prevents us from sending too much at once and OOM-ing on large uploads
  18     def stream_input(input)
  19       if buf = input.read(16384)
  20         send_data buf
  21         EM.next_tick { stream_input(input) }
  22       end
  23     end
  24
  25     def on_write_complete
  26       if @input
  27         buf = @input.read(16384, @junk) and return write(buf)
  28         @input = nil
  29       end
  30     end
  31
  32     def unbind
  33       @env[AsyncCallback].call(@response || Rainbows::ReverseProxy::E502)
  34     end
  35   end
  36
  37   UpstreamSocket = Rainbows::ReverseProxy::UpstreamSocket
  38   def call(env)
  39     input = prepare_input!(env)
  40     io = UpstreamSocket.start(pick_upstream(env))
  41     sock = EM.attach(io, Backend, env)
  42     sock.send_data(build_headers(env, input))
  43     sock.stream_input(input) if input
  44     throw :async
  45   end
  46 end