class Faraday::Adapter::EMHttp

def self.setup_parallel_manager(options = nil)
  Manager.new
end

def call(env)
  super
  perform_request env
  @app.call env
end

def create_request(env)
  EventMachine::HttpRequest.new(env[:url], connection_config(env).merge(@connection_options))
end

def error_message(client)
  client.error or "request failed"
end

def parallel?(env)
  !!env[:parallel_manager]
end

def perform_request(env)
  if parallel?(env)
    manager = env[:parallel_manager]
    manager.add {
      perform_single_request(env).
        callback { env[:response].finish(env) }
    }
  else
    unless EventMachine.reactor_running?
      error = nil
      # start EM, block until request is completed
      EventMachine.run do
        perform_single_request(env).
          callback { EventMachine.stop }.
          errback { |client|
            error = error_message(client)
            EventMachine.stop
          }
      end
      raise_error(error) if error
    else
      # EM is running: instruct upstream that this is an async request
      env[:parallel_manager] = true
      perform_single_request(env).
        callback { env[:response].finish(env) }.
        errback {
          # TODO: no way to communicate the error in async mode
          raise NotImplementedError
        }
    end
  end
rescue EventMachine::Connectify::CONNECTError => err
  if err.message.include?("Proxy Authentication Required")
    raise Error::ConnectionFailed, %{407 "Proxy Authentication Required "}
  else
    raise Error::ConnectionFailed, err
  end
rescue => err
  if defined?(OpenSSL) && OpenSSL::SSL::SSLError === err
    raise Faraday::SSLError, err
  else
    raise
  end
end

def perform_single_request(env)
  req = create_request(env)
  req.setup_request(env[:method], request_config(env)).callback { |client|
    status = client.response_header.status
    reason = client.response_header.http_reason
    save_response(env, status, client.response, nil, reason) do |resp_headers|
      client.response_header.each do |name, value|
        resp_headers[name.to_sym] = value
      end
    end
  }
end

def raise_error(msg)
  errklass = Faraday::Error::ClientError
  if msg == Errno::ETIMEDOUT
    errklass = Faraday::Error::TimeoutError
    msg = "request timed out"
  elsif msg == Errno::ECONNREFUSED
    errklass = Faraday::Error::ConnectionFailed
    msg = "connection refused"
  elsif msg == "connection closed by server"
    errklass = Faraday::Error::ConnectionFailed
  end
  raise errklass, msg
end