+++ /dev/null
-module Faraday
- class Adapter
- # EventMachine adapter is useful for either asynchronous requests
- # when in EM reactor loop or for making parallel requests in
- # synchronous code.
- class EMHttp < Faraday::Adapter
- module Options
- def connection_config(env)
- options = {}
- configure_proxy(options, env)
- configure_timeout(options, env)
- configure_socket(options, env)
- configure_ssl(options, env)
- options
- end
-
- def request_config(env)
- options = {
- :body => read_body(env),
- :head => env[:request_headers],
- # :keepalive => true,
- # :file => 'path/to/file', # stream data off disk
- }
- configure_compression(options, env)
- options
- end
-
- def read_body(env)
- body = env[:body]
- body.respond_to?(:read) ? body.read : body
- end
-
- def configure_proxy(options, env)
- if proxy = request_options(env)[:proxy]
- options[:proxy] = {
- :host => proxy[:uri].host,
- :port => proxy[:uri].port,
- :authorization => [proxy[:user], proxy[:password]]
- }
- end
- end
-
- def configure_socket(options, env)
- if bind = request_options(env)[:bind]
- options[:bind] = {
- :host => bind[:host],
- :port => bind[:port]
- }
- end
- end
-
- def configure_ssl(options, env)
- if env[:url].scheme == 'https' && env[:ssl]
- options[:ssl] = {
- :cert_chain_file => env[:ssl][:ca_file],
- :verify_peer => env[:ssl].fetch(:verify, true)
- }
- end
- end
-
- def configure_timeout(options, env)
- timeout, open_timeout = request_options(env).values_at(:timeout, :open_timeout)
- options[:connect_timeout] = options[:inactivity_timeout] = timeout
- options[:connect_timeout] = open_timeout if open_timeout
- end
-
- def configure_compression(options, env)
- if env[:method] == :get and not options[:head].key? 'accept-encoding'
- options[:head]['accept-encoding'] = 'gzip, compressed'
- end
- end
-
- def request_options(env)
- env[:request]
- end
- end
-
- include Options
-
- dependency 'em-http'
-
- self.supports_parallel = true
-
- def self.setup_parallel_manager(options = nil)
- Manager.new
- end
-
- def call(env)
- super
- perform_request env
- @app.call env
- 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
-
- # TODO: reuse the connection to support pipelining
- def perform_single_request(env)
- req = EventMachine::HttpRequest.new(env[:url], connection_config(env))
- req.setup_request(env[:method], request_config(env)).callback { |client|
- save_response(env, client.response_header.status, client.response) do |resp_headers|
- client.response_header.each do |name, value|
- resp_headers[name.to_sym] = value
- end
- end
- }
- end
-
- def error_message(client)
- client.error or "request failed"
- 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
-
- def parallel?(env)
- !!env[:parallel_manager]
- end
-
- # The parallel manager is designed to start an EventMachine loop
- # and block until all registered requests have been completed.
- class Manager
- def initialize
- reset
- end
-
- def reset
- @registered_procs = []
- @num_registered = 0
- @num_succeeded = 0
- @errors = []
- @running = false
- end
-
- def running?() @running end
-
- def add
- if running?
- perform_request { yield }
- else
- @registered_procs << Proc.new
- end
- @num_registered += 1
- end
-
- def run
- if @num_registered > 0
- @running = true
- EventMachine.run do
- @registered_procs.each do |proc|
- perform_request(&proc)
- end
- end
- if @errors.size > 0
- raise Faraday::Error::ClientError, @errors.first || "connection failed"
- end
- end
- ensure
- reset
- end
-
- def perform_request
- client = yield
- client.callback { @num_succeeded += 1; check_finished }
- client.errback { @errors << client.error; check_finished }
- end
-
- def check_finished
- if @num_succeeded + @errors.size == @num_registered
- EventMachine.stop
- end
- end
- end
- end
- end
-end
-
-begin
- require 'openssl'
-rescue LoadError
- warn "Warning: no such file to load -- openssl. Make sure it is installed if you want HTTPS support"
-else
- require 'faraday/adapter/em_http_ssl_patch'
-end if Faraday::Adapter::EMHttp.loaded?
projects / dsa-puppet.git / blobdiff