session.rb in httpx-1.2.0

- old
+ new

@@ -6,11 +6,10 @@
   #   HTTPX.get(..) #=> delegating to an internal HTTPX::Session object.
   #   HTTPX.plugin(..).get(..) #=> creating an intermediate HTTPX::Session with plugin, then sending the GET request
   class Session
     include Loggable
     include Chainable
-    include Callbacks
 
     EMPTY_HASH = {}.freeze
 
     # initializes the session with a set of +options+, which will be shared by all
     # requests sent from it.
@@ -83,37 +82,11 @@
     def build_request(verb, uri, options = EMPTY_HASH)
       rklass = @options.request_class
       options = @options.merge(options) unless options.is_a?(Options)
       request = rklass.new(verb, uri, options)
       request.persistent = @persistent
-      request.on(:response, &method(:on_response).curry(2)[request])
-      request.on(:promise, &method(:on_promise))
-
-      request.on(:headers) do
-        emit(:request_started, request)
-      end
-      request.on(:body_chunk) do |chunk|
-        emit(:request_body_chunk, request, chunk)
-      end
-      request.on(:done) do
-        emit(:request_completed, request)
-      end
-
-      request.on(:response_started) do |res|
-        if res.is_a?(Response)
-          emit(:response_started, request, res)
-          res.on(:chunk_received) do |chunk|
-            emit(:response_body_chunk, request, res, chunk)
-          end
-        else
-          emit(:request_error, request, res.error)
-        end
-      end
-      request.on(:response) do |res|
-        emit(:response_completed, request, res)
-      end
-
+      set_request_callbacks(request)
       request
     end
 
     private
 
@@ -141,11 +114,11 @@
 
     # returns the HTTPX::Connection through which the +request+ should be sent through.
     def find_connection(request, connections, options)
       uri = request.uri
 
-      connection = pool.find_connection(uri, options) || build_connection(uri, options)
+      connection = pool.find_connection(uri, options) || init_connection(uri, options)
       unless connections.nil? || connections.include?(connection)
         connections << connection
         set_connection_callbacks(connection, connections, options)
       end
       connection
@@ -177,19 +150,10 @@
       end
       connection.only(:altsvc) do |alt_origin, origin, alt_params|
         other_connection = build_altsvc_connection(connection, connections, alt_origin, origin, alt_params, options)
         connections << other_connection if other_connection
       end
-      connection.only(:exhausted) do
-        other_connection = connection.create_idle
-        other_connection.merge(connection)
-        catch(:coalesced) do
-          pool.init_connection(other_connection, options)
-        end
-        set_connection_callbacks(other_connection, connections, options)
-        connections << other_connection
-      end
     end
 
     # returns an HTTPX::Connection for the negotiated Alternative Service (or none).
     def build_altsvc_connection(existing_connection, connections, alt_origin, origin, alt_params, options)
       # do not allow security downgrades on altsvc negotiation
@@ -200,28 +164,23 @@
       # altsvc already exists, somehow it wasn't advertised, probably noop
       return unless altsvc
 
       alt_options = options.merge(ssl: options.ssl.merge(hostname: URI(origin).host))
 
-      connection = pool.find_connection(alt_origin, alt_options) || build_connection(alt_origin, alt_options)
+      connection = pool.find_connection(alt_origin, alt_options) || init_connection(alt_origin, alt_options)
+
       # advertised altsvc is the same origin being used, ignore
       return if connection == existing_connection
 
+      connection.extend(AltSvc::ConnectionMixin) unless connection.is_a?(AltSvc::ConnectionMixin)
+
       set_connection_callbacks(connection, connections, alt_options)
 
       log(level: 1) { "#{origin} alt-svc: #{alt_origin}" }
 
-      # get uninitialized requests
-      # incidentally, all requests will be re-routed to the first
-      # advertised alt-svc, which incidentally follows the spec.
-      existing_connection.purge_pending do |request|
-        request.origin == origin &&
-          request.state == :idle &&
-          !request.headers.key?("alt-used")
-      end
-
       connection.merge(existing_connection)
+      existing_connection.terminate
       connection
     rescue UnsupportedSchemeError
       altsvc["noop"] = true
       nil
     end
@@ -248,34 +207,17 @@
       raise ArgumentError, "wrong number of URIs (given 0, expect 1..+1)" if requests.empty?
 
       requests
     end
 
-    # returns a new HTTPX::Connection object for the given +uri+ and set of +options+.
-    def build_connection(uri, options)
-      type = options.transport || begin
-        case uri.scheme
-        when "http"
-          "tcp"
-        when "https"
-          "ssl"
-        else
-          raise UnsupportedSchemeError, "#{uri}: #{uri.scheme}: unsupported URI scheme"
-        end
-      end
-      init_connection(type, uri, options)
+    def set_request_callbacks(request)
+      request.on(:response, &method(:on_response).curry(2)[request])
+      request.on(:promise, &method(:on_promise))
     end
 
-    def init_connection(type, uri, options)
-      connection = options.connection_class.new(type, uri, options)
-      connection.on(:open) do
-        emit(:connection_opened, connection.origin, connection.io.socket)
-        # only run close callback if it opened
-      end
-      connection.on(:close) do
-        emit(:connection_closed, connection.origin, connection.io.socket) if connection.used?
-      end
+    def init_connection(uri, options)
+      connection = options.connection_class.new(uri, options)
       catch(:coalesced) do
         pool.init_connection(connection, options)
         connection
       end
     end
@@ -367,22 +309,11 @@
 
           include(pl::InstanceMethods) if defined?(pl::InstanceMethods)
           extend(pl::ClassMethods) if defined?(pl::ClassMethods)
 
           opts = @default_options
-          opts.request_class.__send__(:include, pl::RequestMethods) if defined?(pl::RequestMethods)
-          opts.request_class.extend(pl::RequestClassMethods) if defined?(pl::RequestClassMethods)
-          opts.response_class.__send__(:include, pl::ResponseMethods) if defined?(pl::ResponseMethods)
-          opts.response_class.extend(pl::ResponseClassMethods) if defined?(pl::ResponseClassMethods)
-          opts.headers_class.__send__(:include, pl::HeadersMethods) if defined?(pl::HeadersMethods)
-          opts.headers_class.extend(pl::HeadersClassMethods) if defined?(pl::HeadersClassMethods)
-          opts.request_body_class.__send__(:include, pl::RequestBodyMethods) if defined?(pl::RequestBodyMethods)
-          opts.request_body_class.extend(pl::RequestBodyClassMethods) if defined?(pl::RequestBodyClassMethods)
-          opts.response_body_class.__send__(:include, pl::ResponseBodyMethods) if defined?(pl::ResponseBodyMethods)
-          opts.response_body_class.extend(pl::ResponseBodyClassMethods) if defined?(pl::ResponseBodyClassMethods)
-          opts.connection_class.__send__(:include, pl::ConnectionMethods) if defined?(pl::ConnectionMethods)
+          opts.extend_with_plugin_classes(pl)
           if defined?(pl::OptionsMethods)
-            opts.options_class.__send__(:include, pl::OptionsMethods)
 
             (pl::OptionsMethods.instance_methods - Object.instance_methods).each do |meth|
               opts.options_class.method_added(meth)
             end
             @default_options = opts.options_class.new(opts)