feat: opt-in HTTP keep-alive via keep_alive_connections

README.md

@@ -54,6 +54,25 @@ fcm = FCM.new(
 
 ```
 
+## HTTP keep-alive
+
+By default each request opens a fresh TCP/TLS connection to FCM. For high-volume
+senders this dominates per-request latency. Pass `keep_alive_connections: true`
+to reuse a thread-local connection (per host) backed by `net-http-persistent`:
+
+```ruby
+fcm = FCM.new(
+  GOOGLE_APPLICATION_CREDENTIALS_PATH,
+  FIREBASE_PROJECT_ID,
+  keep_alive_connections: true,
+  keep_alive_idle_timeout_seconds: 30, # optional, default 30
+  keep_alive_pool_size: 1              # optional, default 1
+)
+```
+
+`Net::HTTP` is not thread-safe, so connections are cached per `(thread, uri)`.
+Connections are dropped on error so half-closed sockets are not reused.
+
 ## Usage
 
 ## HTTP v1 API

fcm.gemspec

@@ -20,5 +20,7 @@ Gem::Specification.new do |s|
   s.require_paths = ["lib"]
 
   s.add_runtime_dependency("faraday", ">= 1.0.0", "< 3.0")
+  s.add_runtime_dependency("faraday-net_http_persistent", "~> 2.0")
   s.add_runtime_dependency("googleauth", "~> 1")
+  s.add_runtime_dependency("net-http-persistent", "~> 4.0")
 end

lib/fcm.rb

@@ -9,6 +9,8 @@ class InvalidCredentialError < StandardError; end
   BASE_URI = "https://fcm.googleapis.com"
   BASE_URI_V1 = "https://fcm.googleapis.com/v1/projects/"
   DEFAULT_TIMEOUT = 30
+  DEFAULT_KEEP_ALIVE_IDLE_TIMEOUT_SECONDS = 30
+  DEFAULT_KEEP_ALIVE_POOL_SIZE = 1
 
   GROUP_NOTIFICATION_BASE_URI = "https://android.googleapis.com"
   INSTANCE_ID_API = "https://iid.googleapis.com"
@@ -18,6 +20,16 @@ def initialize(json_key_path = "", project_name = "", http_options = {})
     @json_key_path = json_key_path
     @project_name = project_name
     @http_options = http_options
+    @keep_alive_connections = http_options.fetch(:keep_alive_connections, false)
+    @keep_alive_idle_timeout_seconds =
+      http_options.fetch(:keep_alive_idle_timeout_seconds, DEFAULT_KEEP_ALIVE_IDLE_TIMEOUT_SECONDS)
+    @keep_alive_pool_size = http_options.fetch(:keep_alive_pool_size, DEFAULT_KEEP_ALIVE_POOL_SIZE)
+
+    # Per-instance key for the thread-local connection cache so multiple FCM
+    # clients in the same process do not share sockets.
+    @thread_connections_key = :"_fcm_connections_#{object_id}"
+
+    require "faraday/net_http_persistent" if @keep_alive_connections
   end
 
   # See https://firebase.google.com/docs/cloud-messaging/send-message
@@ -193,19 +205,72 @@ def send_to_topic_condition(condition, options = {})
   private
 
   def for_uri(uri, extra_headers = {})
-    connection = ::Faraday.new(
+    if @keep_alive_connections
+      with_persistent_connection(uri, extra_headers) { |connection| yield connection }
+    else
+      yield build_one_shot_connection(uri, extra_headers)
+    end
+  end
+
+  def build_one_shot_connection(uri, extra_headers)
+    ::Faraday.new(
       url: uri,
       request: { timeout: @http_options.fetch(:timeout, DEFAULT_TIMEOUT) }
     ) do |faraday|
       faraday.adapter Faraday.default_adapter
-      faraday.headers["Content-Type"] = "application/json"
-      faraday.headers["Authorization"] = "Bearer #{jwt_token}"
-      faraday.headers["access_token_auth"]= "true"
-      extra_headers.each do |key, value|
-        faraday.headers[key] = value
-      end
+      apply_default_headers(faraday, extra_headers)
     end
+  end
+
+  # Reuses a thread-local Faraday connection (one per uri) backed by
+  # net-http-persistent so the TCP/TLS handshake and HTTP/2 stream are
+  # amortised across requests. Bearer tokens and per-call headers are
+  # re-applied each yield because JWTs expire and extra_headers vary.
+  # On error, the cached connection is dropped: the underlying socket may
+  # be half-closed and reusing it would just fail again.
+  def with_persistent_connection(uri, extra_headers)
+    connection = persistent_connection_for(uri)
+    apply_default_headers(connection, extra_headers)
     yield connection
+  rescue StandardError
+    discard_persistent_connection(uri)
+    raise
+  end
+
+  def apply_default_headers(connection, extra_headers)
+    connection.headers["Content-Type"] = "application/json"
+    connection.headers["Authorization"] = "Bearer #{jwt_token}"
+    connection.headers["access_token_auth"] = "true"
+    extra_headers.each { |key, value| connection.headers[key] = value }
+  end
+
+  # Net::HTTP is not thread-safe, so connections are cached per (thread, uri)
+  # rather than shared across threads.
+  def persistent_connection_for(uri)
+    thread_connections[uri] ||= build_persistent_connection(uri)
+  end
+
+  def discard_persistent_connection(uri)
+    connection = thread_connections.delete(uri)
+    connection.close if connection.respond_to?(:close)
+  end
+
+  def thread_connections
+    Thread.current[@thread_connections_key] ||= {}
+  end
+
+  def build_persistent_connection(uri)
+    ::Faraday.new(
+      url: uri,
+      request: { timeout: @http_options.fetch(:timeout, DEFAULT_TIMEOUT) }
+    ) do |faraday|
+      # pool_size defaults to 1: we already cache one Faraday connection per
+      # (thread, uri), and Net::HTTP is not thread-safe — so a single socket
+      # per pool is the safe default. Override only with a specific reason.
+      faraday.adapter :net_http_persistent, pool_size: @keep_alive_pool_size do |http|
+        http.idle_timeout = @keep_alive_idle_timeout_seconds
+      end
+    end
   end
 
   def build_post_body(registration_ids, options = {})

spec/fcm_spec.rb

@@ -514,4 +514,54 @@
       end
     end
   end
+
+  describe 'keep_alive_connections' do
+    let(:client) { FCM.new(json_key_path, project_name, keep_alive_connections: true) }
+    let(:uri) { "#{FCM::BASE_URI_V1}#{project_name}/messages:send" }
+    let(:send_v1_params) { { 'token' => 'token', 'notification' => { 'title' => 'hi' } } }
+
+    before do
+      stub_request(:post, uri).to_return(body: '{}', headers: {}, status: 200)
+    end
+
+    it 'caches a Faraday connection per (thread, uri) and reuses it across calls' do
+      client.send_v1(send_v1_params)
+      first = client.__send__(:thread_connections)[FCM::BASE_URI_V1]
+
+      client.send_v1(send_v1_params)
+      second = client.__send__(:thread_connections)[FCM::BASE_URI_V1]
+
+      expect(first).to be_a(Faraday::Connection)
+      expect(second).to equal(first)
+    end
+
+    it 'discards the cached connection when a request raises' do
+      client.send_v1(send_v1_params)
+      expect(client.__send__(:thread_connections)[FCM::BASE_URI_V1]).to be_a(Faraday::Connection)
+
+      stub_request(:post, uri).to_raise(Faraday::ConnectionFailed.new('boom'))
+
+      expect { client.send_v1(send_v1_params) }.to raise_error(Faraday::ConnectionFailed)
+      expect(client.__send__(:thread_connections)).not_to have_key(FCM::BASE_URI_V1)
+    end
+
+    it 'does not share connections across FCM instances' do
+      other_client = FCM.new(json_key_path, project_name, keep_alive_connections: true)
+      allow(other_client).to receive(:json_key)
+
+      client.send_v1(send_v1_params)
+      other_client.send_v1(send_v1_params)
+
+      expect(client.__send__(:thread_connections)[FCM::BASE_URI_V1])
+        .not_to equal(other_client.__send__(:thread_connections)[FCM::BASE_URI_V1])
+    end
+
+    it 'falls back to one-shot connections when disabled' do
+      one_shot_client = FCM.new(json_key_path, project_name)
+      allow(one_shot_client).to receive(:json_key)
+      one_shot_client.send_v1(send_v1_params)
+
+      expect(one_shot_client.__send__(:thread_connections)).to be_empty
+    end
+  end
 end