lib/httpx/session.rb in httpx-0.23.4 vs lib/httpx/session.rb in httpx-0.24.0

@@ -2,10 +2,11 @@
 
 module HTTPX
   class Session
     include Loggable
     include Chainable
+    include Callbacks
 
     EMPTY_HASH = {}.freeze
 
     def initialize(options = EMPTY_HASH, &blk)
       @options = self.class.default_options.merge(options)
@@ -43,10 +44,35 @@
       rklass = @options.request_class
       options = @options.merge(options) unless options.is_a?(Options)
       request = rklass.new(verb, uri, options.merge(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
+
       request
     end
 
     private
 
@@ -172,11 +198,20 @@
           "ssl"
         else
           raise UnsupportedSchemeError, "#{uri}: #{uri.scheme}: unsupported URI scheme"
         end
       end
+      init_connection(type, uri, options)
+    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
+        connection.on(:close) { emit(:connection_closed, connection.origin, connection.io.socket) }
+      end
       catch(:coalesced) do
         pool.init_connection(connection, options)
         connection
       end
     end
@@ -250,9 +285,10 @@
 
       def inherited(klass)
         super
         klass.instance_variable_set(:@default_options, @default_options)
         klass.instance_variable_set(:@plugins, @plugins.dup)
+        klass.instance_variable_set(:@callbacks, @callbacks.dup)
       end
 
       def plugin(pl, options = nil, &block)
         # raise Error, "Cannot add a plugin to a frozen config" if frozen?
         pl = Plugins.load_plugin(pl) if pl.is_a?(Symbol)