FCM: HTTP persistent connections with HTTPX

As a result of this change, FCM connections now also make use of a connection pool.
Accordingly, a configuration option, similar to the one used for APNs, has been introduced.

Author: intrip

Gemfile.lock

@@ -7,7 +7,6 @@ PATH
       googleauth (~> 1.14)
       httpx (~> 1.6)
       jwt (>= 2)
-      net-http (~> 0.6)
       railties (>= 8.0)
 
 GEM

action_push_native.gemspec

@@ -38,5 +38,4 @@ Gem::Specification.new do |spec|
   spec.add_dependency "httpx", "~> 1.6"
   spec.add_dependency "jwt", ">= 2"
   spec.add_dependency "googleauth", "~> 1.14"
-  spec.add_dependency "net-http", "~> 0.6"
 end

lib/action_push_native.rb

@@ -3,7 +3,6 @@
 require "zeitwerk"
 require "action_push_native/engine"
 require "action_push_native/errors"
-require "net/http"
 require "httpx"
 require "googleauth"
 require "jwt"

lib/action_push_native/service/apns.rb

@@ -3,13 +3,15 @@
 module ActionPushNative
   module Service
     class Apns
-      def initialize(config)
-        @config = config
-      end
+      include NetworkErrorHandling
 
       # Per-application HTTPX session
       cattr_accessor :httpx_sessions
 
+      def initialize(config)
+        @config = config
+      end
+
       def push(notification)
         notification.apple_data = ApnoticLegacyConverter.convert(notification.apple_data) if notification.apple_data.present?
 
@@ -70,23 +72,6 @@ def handle_error(response)
           end
         end
 
-        def handle_network_error(error)
-          case error
-          when Errno::ETIMEDOUT, HTTPX::TimeoutError
-            raise ActionPushNative::TimeoutError, error.message
-          when Errno::ECONNRESET, Errno::ECONNABORTED, Errno::ECONNREFUSED, Errno::EHOSTUNREACH,
-               SocketError, IOError, EOFError, Errno::EPIPE, Errno::EINVAL, HTTPX::ConnectionError,
-               HTTPX::TLSError, HTTPX::Connection::HTTP2::Error
-            raise ActionPushNative::ConnectionError, error.message
-          when OpenSSL::SSL::SSLError
-            if error.message.include?("SSL_connect")
-              raise ActionPushNative::ConnectionError, error.message
-            else
-              raise
-            end
-          end
-        end
-
         def handle_apns_error(response)
           status = response.status
           reason = JSON.parse(response.body.to_s)["reason"] unless response.body.empty?

lib/action_push_native/service/fcm.rb

@@ -3,22 +3,41 @@
 module ActionPushNative
   module Service
     class Fcm
-      # FCM suggests at least a 10s timeout for requests, we set 15 to add some buffer.
-      # https://firebase.google.com/docs/cloud-messaging/scale-fcm#timeouts
-      DEFAULT_TIMEOUT = 15.seconds
+      include NetworkErrorHandling
+
+      # Per-application HTTPX session
+      cattr_accessor :httpx_sessions
 
       def initialize(config)
         @config = config
       end
 
       def push(notification)
-        response = post_request payload_from(notification)
-        handle_error(response) unless response.code == "200"
+        response = httpx_session.post("v1/projects/#{config.fetch(:project_id)}/messages:send", json: payload_from(notification), headers: { authorization: "Bearer #{access_token}" })
+        handle_error(response) if response.error
       end
 
       private
         attr_reader :config
 
+        def httpx_session
+          self.class.httpx_sessions ||= {}
+          self.class.httpx_sessions[config] ||= build_httpx_session
+        end
+
+        # FCM suggests at least a 10s timeout for requests, we set 15 to add some buffer.
+        # https://firebase.google.com/docs/cloud-messaging/scale-fcm#timeouts
+        DEFAULT_REQUEST_TIMEOUT = 15.seconds
+        DEFAULT_POOL_SIZE       = 5
+
+        def build_httpx_session
+          HTTPX.
+            plugin(:persistent, close_on_fork: true).
+            with(timeout: { request_timeout: config[:request_timeout] || DEFAULT_REQUEST_TIMEOUT }).
+            with(pool_options: { max_connections: config[:connection_pool_size] || DEFAULT_POOL_SIZE }).
+            with(origin: "https://fcm.googleapis.com")
+        end
+
         def payload_from(notification)
           deep_compact({
             message: {
@@ -56,34 +75,6 @@ def stringify(hash)
           hash.compact.transform_values(&:to_s)
         end
 
-        def post_request(payload)
-          uri = URI("https://fcm.googleapis.com/v1/projects/#{config.fetch(:project_id)}/messages:send")
-          request = Net::HTTP::Post.new(uri)
-          request["Authorization"] = "Bearer #{access_token}"
-          request["Content-Type"]  = "application/json"
-          request.body             = payload.to_json
-
-          rescue_and_reraise_network_errors do
-            Net::HTTP.start(uri.host, uri.port, use_ssl: true, read_timeout: config[:request_timeout] || DEFAULT_TIMEOUT) do |http|
-              http.request(request)
-            end
-          end
-        end
-
-        def rescue_and_reraise_network_errors
-          yield
-        rescue Net::ReadTimeout, Net::OpenTimeout => e
-          raise ActionPushNative::TimeoutError, e.message
-        rescue Errno::ECONNRESET, SocketError => e
-          raise ActionPushNative::ConnectionError, e.message
-        rescue OpenSSL::SSL::SSLError => e
-          if e.message.include?("SSL_connect")
-            raise ActionPushNative::ConnectionError, e.message
-          else
-            raise
-          end
-        end
-
         def access_token
           authorizer = Google::Auth::ServiceAccountCredentials.make_creds \
             json_key_io: StringIO.new(config.fetch(:encryption_key)),
@@ -92,28 +83,36 @@ def access_token
         end
 
         def handle_error(response)
-          code = response.code
+          if response.is_a?(HTTPX::ErrorResponse)
+            handle_network_error(response.error)
+          else
+            handle_fcm_error(response)
+          end
+        end
+
+        def handle_fcm_error(response)
+          status = response.status
           reason = \
             begin
-              JSON.parse(response.body).dig("error", "message")
+              JSON.parse(response.body.to_s).dig("error", "message")
             rescue JSON::ParserError
-              response.body
+              response.body.to_s
             end
 
-          Rails.logger.error("FCM response error #{code}: #{reason}")
+          Rails.logger.error("FCM response error #{status}: #{reason}")
 
           case
           when reason =~ /message is too big/i
             raise ActionPushNative::PayloadTooLargeError, reason
-          when code == "400"
+          when status == 400
             raise ActionPushNative::BadRequestError, reason
-          when code == "404"
+          when status == 404
             raise ActionPushNative::TokenError, reason
-          when code.in?([ "401", "403" ])
+          when status.in?([ 401, 403 ])
             raise ActionPushNative::ForbiddenError, reason
-          when code == "429"
+          when status == 429
             raise ActionPushNative::TooManyRequestsError, reason
-          when code == "503"
+          when status == 503
             raise ActionPushNative::ServiceUnavailableError, reason
           else
             raise ActionPushNative::InternalServerError, reason

lib/action_push_native/service/network_error_handling.rb

@@ -0,0 +1,20 @@
+module ActionPushNative::Service::NetworkErrorHandling
+  private
+
+  def handle_network_error(error)
+    case error
+    when Errno::ETIMEDOUT, HTTPX::TimeoutError
+      raise ActionPushNative::TimeoutError, error.message
+    when Errno::ECONNRESET, Errno::ECONNABORTED, Errno::ECONNREFUSED, Errno::EHOSTUNREACH,
+      SocketError, IOError, EOFError, Errno::EPIPE, Errno::EINVAL, HTTPX::ConnectionError,
+      HTTPX::TLSError, HTTPX::Connection::HTTP2::Error
+      raise ActionPushNative::ConnectionError, error.message
+    when OpenSSL::SSL::SSLError
+      if error.message.include?("SSL_connect")
+        raise ActionPushNative::ConnectionError, error.message
+      else
+        raise
+      end
+    end
+  end
+end

lib/generators/action_push_native/install/templates/config/push.yml.tt

@@ -32,5 +32,9 @@ shared:
     # Firebase project_id
     project_id: your_project_id
 
+    # Set this to the number of threads used to process notifications (default: 5).
+    # When the pool size is too small a HTTPX::PoolTimeoutError error will be raised.
+    # connection_pool_size: 5
+
     # Change the request timeout (default: 15).
     # request_timeout: 30

test/dummy/config/push.yml

@@ -29,6 +29,10 @@ shared:
     # See https://firebase.google.com/docs/cloud-messaging/auth-server
     encryption_key: <%= Rails.application.credentials.dig(:action_push_native, :fcm, :encryption_key) %>
 
+    # Set this to the number of threads used to process notifications (default: 5).
+    # When the pool size is too small a HTTPX::PoolTimeoutError error will be raised.
+    # connection_pool_size: 5
+
     # Firebase project_id
     project_id: your_project_id
 

test/jobs/action_push_native/notification_job_test.rb

@@ -36,7 +36,8 @@ class NotificationJobTest < ActiveSupport::TestCase
 
     test "Socket errors are retried" do
       device = action_push_native_devices(:pixel9)
-      Net::HTTP.any_instance.stubs(:request).raises(SocketError)
+      stub_request(:post, "https://fcm.googleapis.com/v1/projects/your_project_id/messages:send").
+        to_raise(SocketError.new)
       ActionPushNative::Service::Fcm.any_instance.stubs(:access_token).returns("fake_access_token")
 
       assert_enqueued_jobs 1, only: ActionPushNative::NotificationJob do

test/lib/action_push_native/service/fcm_test.rb

@@ -49,7 +49,14 @@ class FcmTest < ActiveSupport::TestCase
           @fcm.push(@notification)
         end
 
-        Net::HTTP.stubs(:start).raises(OpenSSL::SSL::SSLError.new("SSL_connect returned=1 errno=0 state=error"))
+        stub_request(:post, "https://fcm.googleapis.com/v1/projects/your_project_id/messages:send").
+          to_return(status: 500, body: "Not a JSON")
+        assert_raises ActionPushNative::InternalServerError do
+          @fcm.push(@notification)
+        end
+
+        stub_request(:post, "https://fcm.googleapis.com/v1/projects/your_project_id/messages:send").
+          to_raise(OpenSSL::SSL::SSLError.new("SSL_connect returned=1 errno=0 state=error"))
         assert_raises ActionPushNative::ConnectionError do
           @fcm.push(@notification)
         end