ref: Get rid of concurrent async

.github/workflows/tests.yml

@@ -11,13 +11,11 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        gemfile: [websocket-driver-0.6.x, websocket-driver-0.7.x]
         ruby: [2.7, "3.0", 3.1, 3.2, 3.3]
     runs-on: ubuntu-latest
     env:
       FERRUM_PROCESS_TIMEOUT: 25
       FERRUM_DEFAULT_TIMEOUT: 15
-      BUNDLE_GEMFILE: .github/gemfiles/${{ matrix.gemfile }}.gemfile
     steps:
       - name: Checkout code
         uses: actions/checkout@v4

CHANGELOG.md

@@ -12,11 +12,15 @@
 - `Ferrum::Page#screeshot` accepts :area option [#410]
 - Resizing page on creation is gone and moved to Cuprite [#427]
 - Min Ruby version is 2.7
-- Refactored internal API of `Browser`, `Page`, `Context`, `Contexts`, `Target` instead of passing browser and making
-cyclic dependency on the browser instance, we pass now a simple client [#431]
+- Refactored internal API of `Ferrum::Browser`, `Ferrum::Page`, `Ferrum::Context`, `Ferrum::Contexts`, `Ferrum::Target`
+instead of passing browser and making cyclic dependency on the browser instance, we pass now a thin client [#431]
+- Bump `websocket-driver` to `~> 0.7` [#432]
+- Got rid of `Concurrent::Async` in `Ferrum::Browser::Subscriber` [#432]
 
 ### Fixed
 
+- Exceptions within `.on()` were swallowed by a thread pool of `Concurrent::Async` [#432]
+
 ### Removed
 
 

ferrum.gemspec

@@ -28,5 +28,5 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency "addressable",      "~> 2.5"
   s.add_runtime_dependency "concurrent-ruby",  "~> 1.1"
   s.add_runtime_dependency "webrick",          "~> 1.7"
-  s.add_runtime_dependency "websocket-driver", ">= 0.6", "< 0.8"
+  s.add_runtime_dependency "websocket-driver", "~> 0.7"
 end

lib/ferrum.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require "ferrum/utils/thread"
 require "ferrum/utils/platform"
 require "ferrum/utils/elapsed_time"
 require "ferrum/utils/attempt"

lib/ferrum/browser.rb

@@ -208,6 +208,7 @@ def restart
     def quit
       return unless @client
 
+      contexts.close_connections
       @client.close
       @process.stop
       @client = @process = @contexts = nil

lib/ferrum/browser/client.rb

@@ -7,8 +7,6 @@
 module Ferrum
   class Browser
     class Client
-      INTERRUPTIONS = %w[Fetch.requestPaused Fetch.authRequired].freeze
-
       extend Forwardable
       delegate %i[timeout timeout=] => :options
 
@@ -19,25 +17,9 @@ def initialize(ws_url, options)
         @options = options
         @pendings = Concurrent::Hash.new
         @ws = WebSocket.new(ws_url, options.ws_max_receive_size, options.logger)
-        @subscriber, @interrupter = Subscriber.build(2)
-
-        @thread = Thread.new do
-          Thread.current.abort_on_exception = true
-          Thread.current.report_on_exception = true if Thread.current.respond_to?(:report_on_exception=)
-
-          loop do
-            message = @ws.messages.pop
-            break unless message
+        @subscriber = Subscriber.new
 
-            if INTERRUPTIONS.include?(message["method"])
-              @interrupter.async.call(message)
-            elsif message.key?("method")
-              @subscriber.async.call(message)
-            else
-              @pendings[message["id"]]&.set(message)
-            end
-          end
-        end
+        start
       end
 
       def command(method, params = {})
@@ -57,23 +39,19 @@ def command(method, params = {})
       end
 
       def on(event, &block)
-        case event
-        when *INTERRUPTIONS
-          @interrupter.on(event, &block)
-        else
-          @subscriber.on(event, &block)
-        end
+        @subscriber.on(event, &block)
       end
 
       def subscribed?(event)
-        [@interrupter, @subscriber].any? { |s| s.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
@@ -85,6 +63,21 @@ def inspect
 
       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

lib/ferrum/browser/subscriber.rb

@@ -3,15 +3,22 @@
 module Ferrum
   class Browser
     class Subscriber
-      include Concurrent::Async
-
-      def self.build(size)
-        (0..size).map { new }
-      end
+      INTERRUPTIONS = %w[Fetch.requestPaused Fetch.authRequired].freeze
 
       def initialize
-        super
+        @regular = Queue.new
+        @priority = Queue.new
         @on = Concurrent::Hash.new { |h, k| h[k] = Concurrent::Array.new }
+
+        start
+      end
+
+      def <<(message)
+        if INTERRUPTIONS.include?(message["method"])
+          @priority.push(message)
+        else
+          @regular.push(message)
+        end
       end
 
       def on(event, &block)
@@ -23,6 +30,33 @@ def subscribed?(event)
         @on.key?(event)
       end
 
+      def close
+        @regular_thread&.kill
+        @priority_thread&.kill
+      end
+
+      private
+
+      def start
+        @regular_thread = Utils::Thread.spawn(abort_on_exception: false) do
+          loop do
+            message = @regular.pop
+            break unless message
+
+            call(message)
+          end
+        end
+
+        @priority_thread = Utils::Thread.spawn(abort_on_exception: false) do
+          loop do
+            message = @priority.pop
+            break unless message
+
+            call(message)
+          end
+        end
+      end
+
       def call(message)
         method, params = message.values_at("method", "params")
         total = @on[method].size

lib/ferrum/browser/web_socket.rb

@@ -27,21 +27,7 @@ def initialize(url, max_receive_size, logger)
         @driver.on(:message, &method(:on_message))
         @driver.on(:close,   &method(:on_close))
 
-        @thread = Thread.new do
-          Thread.current.abort_on_exception = true
-          Thread.current.report_on_exception = true if Thread.current.respond_to?(:report_on_exception=)
-
-          begin
-            loop do
-              data = @sock.readpartial(512)
-              break unless data
-
-              @driver.parse(data)
-            end
-          rescue EOFError, Errno::ECONNRESET, Errno::EPIPE
-            @messages.close
-          end
-        end
+        start
 
         @driver.start
       end
@@ -66,6 +52,7 @@ def on_message(event)
 
       def on_close(_event)
         @messages.close
+        @sock.close
         @thread.kill
       end
 
@@ -79,13 +66,28 @@ def send_message(data)
 
       def write(data)
         @sock.write(data)
-      rescue EOFError, Errno::ECONNRESET, Errno::EPIPE
+      rescue EOFError, Errno::ECONNRESET, Errno::EPIPE, IOError # rubocop:disable Lint/ShadowedException
         @messages.close
       end
 
       def close
         @driver.close
       end
+
+      private
+
+      def start
+        @thread = Utils::Thread.spawn do
+          loop do
+            data = @sock.readpartial(512)
+            break unless data
+
+            @driver.parse(data)
+          end
+        rescue EOFError, Errno::ECONNRESET, Errno::EPIPE, IOError # rubocop:disable Lint/ShadowedException
+          @messages.close
+        end
+      end
     end
   end
 end

lib/ferrum/context.rb

@@ -64,13 +64,21 @@ def add_target(params)
     end
 
     def update_target(target_id, params)
-      @targets[target_id].update(params)
+      @targets[target_id]&.update(params)
     end
 
     def delete_target(target_id)
       @targets.delete(target_id)
     end
 
+    def close_targets_connection
+      @targets.each_value do |target|
+        next unless target.attached?
+
+        target.page.close_connection
+      end
+    end
+
     def dispose
       @contexts.dispose(@id)
     end

lib/ferrum/contexts.rb

@@ -45,11 +45,16 @@ def create(**options)
 
     def dispose(context_id)
       context = @contexts[context_id]
+      context.close_targets_connection
       @client.command("Target.disposeBrowserContext", browserContextId: context.id)
       @contexts.delete(context_id)
       true
     end
 
+    def close_connections
+      @contexts.each_value(&:close_targets_connection)
+    end
+
     def reset
       @default_context = nil
       @contexts.each_key { |id| dispose(id) }

lib/ferrum/page.rb

@@ -120,7 +120,13 @@ def go_to(url = nil)
     def close
       @headers.clear
       client(browser: true).command("Target.closeTarget", targetId: @target_id)
-      @page_client.close
+      close_connection
+
+      true
+    end
+
+    def close_connection
+      @page_client&.close
     end
 
     #

lib/ferrum/page/frames.rb

@@ -134,7 +134,7 @@ def subscribe_frame_stopped_loading
           end
 
           frame = @frames[params["frameId"]]
-          frame.state = :stopped_loading
+          frame&.state = :stopped_loading
 
           @event.set if idling?
         end

lib/ferrum/proxy.rb

@@ -52,7 +52,7 @@ def rotate(host:, port:, user: nil, password: nil)
     end
 
     def stop
-      @file&.unlink
+      @file&.close(true)
       @server.shutdown
     end
 

lib/ferrum/utils/thread.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+module Ferrum
+  module Utils
+    module Thread
+      module_function
+
+      def spawn(abort_on_exception: true)
+        ::Thread.new(abort_on_exception) do
+          ::Thread.current.abort_on_exception = abort_on_exception
+          ::Thread.current.report_on_exception = true if ::Thread.current.respond_to?(:report_on_exception=)
+
+          yield
+        end
+      end
+    end
+  end
+end