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,84 +3,77 @@
 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 payload_from(notification)
-          deep_compact({
-            message: {
-              token: notification.token,
-              data: notification.data ? stringify(notification.data) : {},
-              android: {
-                notification: {
-                  title: notification.title,
-                  body: notification.body,
-                  notification_count: notification.badge,
-                  sound: notification.sound
-                },
-                collapse_key: notification.thread_id,
-                priority: notification.high_priority == true ? "high" : "normal"
-              }
-            }.deep_merge(notification.google_data ? stringify_data(notification.google_data) : {})
-          })
+        def httpx_session
+          self.class.httpx_sessions ||= {}
+          self.class.httpx_sessions[config] ||= build_httpx_session
         end
 
-        def deep_compact(payload)
-          payload.dig(:message, :android, :notification).try(&:compact!)
-          payload.dig(:message, :android).try(&:compact!)
-          payload[:message].compact!
-          payload
-        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
 
-        # FCM requires data values to be strings.
-        def stringify_data(google_data)
-          google_data.tap do |payload|
-            payload[:data] = stringify(payload[:data]) if payload[:data]
-          end
+        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 stringify(hash)
-          hash.compact.transform_values(&:to_s)
-        end
+        concerning :Payload do
+          def payload_from(notification)
+            deep_compact({
+              message: {
+                token: notification.token,
+                data: notification.data ? stringify(notification.data) : {},
+                android: {
+                  notification: {
+                    title: notification.title,
+                    body: notification.body,
+                    notification_count: notification.badge,
+                    sound: notification.sound
+                  },
+                  collapse_key: notification.thread_id,
+                  priority: notification.high_priority == true ? "high" : "normal"
+                }
+              }.deep_merge(notification.google_data ? stringify_data(notification.google_data) : {})
+            })
+          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
+          def deep_compact(payload)
+            payload.dig(:message, :android, :notification).try(&:compact!)
+            payload.dig(:message, :android).try(&:compact!)
+            payload[:message].compact!
+            payload
+          end
 
-          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)
+          # FCM requires data values to be strings.
+          def stringify_data(google_data)
+            google_data.tap do |payload|
+              payload[:data] = stringify(payload[:data]) if payload[:data]
             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
+          def stringify(hash)
+            hash.compact.transform_values(&:to_s)
           end
         end
 
@@ -92,28 +85,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