fix: add_target putting wrong target to pendings sometimes

CHANGELOG.md

@@ -7,6 +7,7 @@
   - `#files` information about downloaded files
   - `#wait` wait for file download to be completed
   - `#set_behavior` where and whether to store file
+- `Browser::Client#command` accepts :async parameter [#433]
 
 ### Changed
 - `Ferrum::Page#screeshot` accepts :area option [#410]
@@ -17,10 +18,15 @@ instead of passing browser and making cyclic dependency on the browser instance,
 - Bump `websocket-driver` to `~> 0.7` [#432]
 - Got rid of `Concurrent::Async` in `Ferrum::Browser::Subscriber` [#432]
 - `Ferrum::Page#set_window_bounds` is renamed to `Ferrum::Page#window_bounds=`
+- `Ferrum::Page` get right client from the Target and passes it down everywhere [#433]
+- `Ferrum::Network::InterceptedRequest` accepts `Ferrum::Browser::Client` instead of `Ferrum::Page` [#433]
+- `Ferrum::Browser::Client` -> `Ferrum::Client` [#433]
 
 ### Fixed
 
 - Exceptions within `.on()` were swallowed by a thread pool of `Concurrent::Async` [#432]
+- `Ferrum::Context#add_target` puts wrong target to pendings sometimes [#433]
+- Leaking connection descriptors in tests and after browser quit [#433]
 
 ### Removed
 

lib/ferrum/browser.rb

@@ -4,11 +4,11 @@
 require "forwardable"
 require "ferrum/page"
 require "ferrum/proxy"
+require "ferrum/client"
 require "ferrum/contexts"
 require "ferrum/browser/xvfb"
 require "ferrum/browser/options"
 require "ferrum/browser/process"
-require "ferrum/browser/client"
 require "ferrum/browser/binary"
 require "ferrum/browser/version_info"
 
@@ -209,6 +209,7 @@ def quit
       return unless @client
 
       contexts.close_connections
+
       @client.close
       @process.stop
       @client = @process = @contexts = nil

lib/ferrum/client.rb

@@ -0,0 +1,113 @@
+# frozen_string_literal: true
+
+require "forwardable"
+require "ferrum/client/subscriber"
+require "ferrum/client/web_socket"
+
+module Ferrum
+  class Client
+    extend Forwardable
+    delegate %i[timeout timeout=] => :options
+
+    attr_reader :options
+
+    def initialize(ws_url, options)
+      @command_id = 0
+      @options = options
+      @pendings = Concurrent::Hash.new
+      @ws = WebSocket.new(ws_url, options.ws_max_receive_size, options.logger)
+      @subscriber = Subscriber.new
+
+      start
+    end
+
+    def command(method, async: false, **params)
+      message = build_message(method, params)
+
+      if async
+        @ws.send_message(message)
+        true
+      else
+        pending = Concurrent::IVar.new
+        @pendings[message[:id]] = pending
+        @ws.send_message(message)
+        data = pending.value!(timeout)
+        @pendings.delete(message[:id])
+
+        raise DeadBrowserError if data.nil? && @ws.messages.closed?
+        raise TimeoutError unless data
+
+        error, response = data.values_at("error", "result")
+        raise_browser_error(error) if error
+        response
+      end
+    end
+
+    def on(event, &block)
+      @subscriber.on(event, &block)
+    end
+
+    def subscribed?(event)
+      @subscriber.subscribed?(event)
+    end
+
+    def close
+      @ws.close
+      # Give a thread some time to handle a tail of messages
+      @pendings.clear
+      @thread.kill unless @thread.join(1)
+      @subscriber.close
+    end
+
+    def inspect
+      "#<#{self.class} " \
+        "@command_id=#{@command_id.inspect} " \
+        "@pendings=#{@pendings.inspect} " \
+        "@ws=#{@ws.inspect}>"
+    end
+
+    private
+
+    def start
+      @thread = Utils::Thread.spawn do
+        loop do
+          message = @ws.messages.pop
+          break unless message
+
+          if message.key?("method")
+            @subscriber << message
+          else
+            @pendings[message["id"]]&.set(message)
+          end
+        end
+      end
+    end
+
+    def build_message(method, params)
+      { method: method, params: params }.merge(id: next_command_id)
+    end
+
+    def next_command_id
+      @command_id += 1
+    end
+
+    def raise_browser_error(error)
+      case error["message"]
+      # Node has disappeared while we were trying to get it
+      when "No node with given id found",
+           "Could not find node with given id",
+           "Inspected target navigated or closed"
+        raise NodeNotFoundError, error
+      # Context is lost, page is reloading
+      when "Cannot find context with specified id"
+        raise NoExecutionContextError, error
+      when "No target with given id found"
+        raise NoSuchPageError
+      when /Could not compute content quads/
+        raise CoordinatesNotFoundError
+      else
+        raise BrowserError, error
+      end
+    end
+  end
+end

lib/ferrum/browser/subscriber.rb → lib/ferrum/client/subscriber.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module Ferrum
-  class Browser
+  class Client
     class Subscriber
       INTERRUPTIONS = %w[Fetch.requestPaused Fetch.authRequired].freeze
 

lib/ferrum/browser/web_socket.rb → lib/ferrum/client/web_socket.rb

@@ -5,7 +5,7 @@
 require "websocket/driver"
 
 module Ferrum
-  class Browser
+  class Client
     class WebSocket
       WEBSOCKET_BUG_SLEEP = 0.05
       SKIP_LOGGING_SCREENSHOTS = !ENV["FERRUM_LOGGING_SCREENSHOTS"]

lib/ferrum/context.rb

@@ -54,10 +54,11 @@ def create_target
     end
 
     def add_target(params)
-      target = Target.new(@client, params)
-      @targets.put_if_absent(target.id, target)
-
+      new_target = Target.new(@client, params)
+      target = @targets.put_if_absent(new_target.id, new_target)
+      target ||= new_target # `put_if_absent` returns nil if added a new value or existing if there was one already
       @pendings.put(target, @client.timeout) if @pendings.empty?
+      target
     end
 
     def update_target(target_id, params)

lib/ferrum/network/intercepted_request.rb

@@ -10,9 +10,9 @@ class InterceptedRequest
 
       attr_accessor :request_id, :frame_id, :resource_type, :network_id, :status
 
-      def initialize(page, params)
+      def initialize(client, params)
         @status = nil
-        @page = page
+        @client = client
         @params = params
         @request_id = params["requestId"]
         @frame_id = params["frameId"]
@@ -43,18 +43,18 @@ def respond(**options)
         options = options.merge(body: Base64.strict_encode64(options.fetch(:body, ""))) if has_body
 
         @status = :responded
-        @page.command("Fetch.fulfillRequest", **options)
+        @client.command("Fetch.fulfillRequest", **options)
       end
 
       def continue(**options)
         options = options.merge(requestId: request_id)
         @status = :continued
-        @page.command("Fetch.continueRequest", **options)
+        @client.command("Fetch.continueRequest", **options)
       end
 
       def abort
         @status = :aborted
-        @page.command("Fetch.failRequest", requestId: request_id, errorReason: "BlockedByClient")
+        @client.command("Fetch.failRequest", requestId: request_id, errorReason: "BlockedByClient")
       end
 
       def initial_priority