perform_request

in lib/elastic/transport/transport/base.rb [270:368]

• 74 lines of code
• 35 McCabe index (conditional complexity)

        def perform_request(method, path, params = {}, body = nil, headers = nil, opts = {}, &block)
          raise NoMethodError, 'Implement this method in your transport class' unless block_given?

          start = Time.now
          tries = 0
          reload_on_failure = opts.fetch(:reload_on_failure, @options[:reload_on_failure])
          delay_on_retry = opts.fetch(:delay_on_retry, @options[:delay_on_retry])

          max_retries = max_retries(opts) || max_retries(options)

          params = params.clone
          
          ignore = Array(params.delete(:ignore)).compact.map(&:to_i)

          begin
            sleep(delay_on_retry / 1000.0) if tries > 0
            tries += 1
            connection = get_connection or raise Error.new('Cannot get new connection from pool.')

            if connection.connection.respond_to?(:params) &&
               connection.connection.params.respond_to?(:to_hash)
              params = connection.connection.params.merge(params.to_hash)
            end

            url = connection.full_url(path, params)
            response = block.call(connection, url)
            connection.healthy! if connection.failures.positive?

            
            __raise_transport_error(response) if response.status.to_i >= 300 &&
                                                 @retry_on_status.include?(response.status.to_i)
          rescue Elastic::Transport::Transport::ServerError => e
            raise e unless response && @retry_on_status.include?(response.status)

            log_warn "[#{e.class}] Attempt #{tries} to get response from #{url}"
            if tries <= (max_retries || DEFAULT_MAX_RETRIES)
              retry
            else
              log_fatal "[#{e.class}] Cannot get response from #{url} after #{tries} tries"
              raise e
            end
          rescue *host_unreachable_exceptions => e
            log_error "[#{e.class}] #{e.message} #{connection.host.inspect}"

            connection.dead!

            if reload_on_failure && tries < connections.all.size
              log_warn "[#{e.class}] Reloading connections (attempt #{tries} of #{connections.all.size})"
              reload_connections! and retry
            end

            exception = Elastic::Transport::Transport::Error.new(e.message)

            raise exception unless max_retries

            log_warn "[#{e.class}] Attempt #{tries} connecting to #{connection.host.inspect}"
            if tries <= max_retries
              retry
            else
              log_fatal "[#{e.class}] Cannot connect to #{connection.host.inspect} after #{tries} tries"
              raise exception
            end
          rescue Exception => e
            log_fatal "[#{e.class}] #{e.message} (#{connection.host.inspect if connection})"
            raise e
          end 

          duration = Time.now - start

          if response.status.to_i >= 300
            __log_response(method, path, params, body, url, response, nil, 'N/A', duration)
            __trace(method, path, params, connection_headers(connection), body, url, response, nil, 'N/A', duration) if tracer

            
            log_fatal("[#{response.status}] #{response.body}") unless ignore.include?(response.status.to_i)
            __raise_transport_error response unless ignore.include?(response.status.to_i)
          end

          if response.body &&
            !response.body.empty? &&
            response.headers &&
            response.headers["content-type"] =~ /json/

            
            load_options = {}
            load_options[:mode] = :compat if ::MultiJson.adapter.to_s == "MultiJson::Adapters::Oj"

            json = serializer.load(response.body, load_options)
          end
          took = (json['took'] ? sprintf('%.3fs', json['took'] / 1000.0) : 'n/a') rescue 'n/a'
          __log_response(method, path, params, body, url, response, json, took, duration) unless ignore.include?(response.status.to_i)
          __trace(method, path, params, connection_headers(connection), body, url, response, nil, 'N/A', duration) if tracer
          log_warn(response.headers['warning']) if response.headers&.[]('warning')

          Response.new response.status, json || response.body, response.headers
        ensure
          @last_request_at = Time.now
        end