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 || 'request failed'
end

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

def perform_request(env)
  if parallel?(env)
    manager = env[:parallel_manager]
    manager.add do
      perform_single_request(env)
        .callback { env[:response].finish(env) }
    end
  elsif EventMachine.reactor_running?
    # 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 do
        # TODO: no way to communicate the error in async mode
        raise NotImplementedError
      end
  else
    error = nil
    # start EM, block until request is completed
    EventMachine.run do
      perform_single_request(env)
        .callback { EventMachine.stop }
        .errback do |client|
          error = error_message(client)
          EventMachine.stop
        end
    end
    raise_error(error) if error
  end
rescue EventMachine::Connectify::CONNECTError => e
  if e.message.include?('Proxy Authentication Required')
    raise Faraday::ConnectionFailed,
          %(407 "Proxy Authentication Required ")
  end
  raise Faraday::ConnectionFailed, e
rescue StandardError => e
  if defined?(::OpenSSL::SSL::SSLError) && \
     e.is_a?(::OpenSSL::SSL::SSLError)
    raise Faraday::SSLError, e
  end
  raise
end

def perform_single_request(env)
  req = create_request(env)
  req = req.setup_request(env[:method], request_config(env))
  req.callback do |client|
    if env[:request].stream_response?
      warn "Streaming downloads for #{self.class.name} " \
        'are not yet implemented.'
      env[:request].on_data.call(
        client.response,
        client.response.bytesize
      )
    end
    status = client.response_header.status
    reason = client.response_header.http_reason
    save_response(env, status, client.response, nil, reason) do |headers|
      client.response_header.each do |name, value|
        headers[name.to_sym] = value
      end
    end
  end
end

def raise_error(msg)
  error_class = Faraday::ClientError
  if timeout_message?(msg)
    error_class = Faraday::TimeoutError
    msg = 'request timed out'
  elsif msg == Errno::ECONNREFUSED
    error_class = Faraday::ConnectionFailed
    msg = 'connection refused'
  elsif msg == 'connection closed by server'
    error_class = Faraday::ConnectionFailed
  end
  raise error_class, msg
end

def timeout_message?(msg)
  msg == Errno::ETIMEDOUT ||
    (msg.is_a?(String) && msg.include?('timeout error'))
end