feat: batch broadcasts

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## master
 
+- Add `batch_broadcasts` option to automatically batch broadcasts for code wrapped in Rails executor. ([@palkan][])
+
 ## 1.4.1 (2023-09-27)
 
 - Fix compatibility with Rails 7.1. ([@palkan][])

docs/getting_started.md

@@ -91,6 +91,12 @@ production:
 
 Or you can use the environment variables (or anything else supported by [anyway_config](https://github.com/palkan/anyway_config)).
 
+### Batching broadcasts automatically
+
+AnyCable supports publishing [broadcast messages in batches](../ruby/broadcast_adapters.md#batching) (to reduce the number of round-trips and ensure delivery order). You can enable automatic batching of broadcasts by setting `ANYCABLE_BROADCAST_BATCHING=true` (or `broadcast_batching: true` in the config file).
+
+Auto-batching uses [Rails executor](https://guides.rubyonrails.org/threading_and_code_execution.html#executor) under the hood, so broadcasts are aggregated within Rails _units of work_, such as HTTP requests, background jobs, etc.
+
 ### Server installation
 
 You can install AnyCable-Go server using one of the [multiple ways](../anycable-go/getting_started.md#installation).

lib/anycable/rails/config.rb

@@ -10,10 +10,12 @@
 # - `persistent_session_enabled` (defaults to false) — whether to store session changes in the connection state
 # - `embedded` (defaults to false) — whether to run RPC server inside a Rails server process
 # - `http_rpc_mount_path` (default to nil) — path to mount HTTP RPC server
+# - `batch_broadcasts` (defaults to false) — whether to batch broadcasts automatically for code wrapped with Rails executor
 AnyCable::Config.attr_config(
   access_logs_disabled: true,
   persistent_session_enabled: false,
   embedded: false,
-  http_rpc_mount_path: nil
+  http_rpc_mount_path: nil,
+  batch_broadcasts: false
 )
 AnyCable::Config.ignore_options :access_logs_disabled, :persistent_session_enabled

lib/anycable/rails/railtie.rb

@@ -59,6 +59,15 @@ class Railtie < ::Rails::Railtie # :nodoc:
             ::Rails.error.report(ex, handled: false, context: {method: method.to_sym, payload: message})
           end
         end
+
+        if AnyCable.config.batch_broadcasts?
+          if AnyCable.broadcast_adapter.respond_to?(:start_batching)
+            app.executor.to_run { AnyCable.broadcast_adapter.start_batching }
+            app.executor.to_complete { AnyCable.broadcast_adapter.finish_batching }
+          else
+            warn "[AnyCable] Auto-batching is enabled for broadcasts but your anycable version doesn't support it. Please, upgrade"
+          end
+        end
       end
 
       initializer "anycable.connection_factory", after: "action_cable.set_configs" do |app|

spec/dummy/app/controllers/broadcasts_controller.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+class BroadcastsController < ApplicationController
+  around_action :maybe_disable_auto_batching
+
+  def create
+    params[:count].to_i.times do |num|
+      ActionCable.server.broadcast "test", {count: num + 1}
+    end
+
+    head :created
+  end
+
+  private
+
+  def maybe_disable_auto_batching(&block)
+    return yield unless params[:disable_auto_batching]
+    AnyCable.broadcast_adapter.batching(false, &block)
+  end
+end

spec/dummy/config/anycable.yml

@@ -4,3 +4,4 @@ test:
   access_logs_disabled: false
   persistent_session_enabled: true
   http_rpc_mount_path: "/_anycable"
+  batch_broadcasts: true

spec/dummy/config/routes.rb

@@ -2,4 +2,5 @@
 
 Rails.application.routes.draw do
   resources :sessions, only: [:create]
+  resources :broadcasts, only: [:create]
 end

spec/integrations/auto_batching_spec.rb

@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+require "action_controller/test_case"
+
+describe "auto-batching", skip: !AnyCable.broadcast_adapter.respond_to?(:start_batching) do
+  include ActionDispatch::Integration::Runner
+  include ActionDispatch::IntegrationTest::Behavior
+
+  # Delegates to `Rails.application`.
+  def app
+    ::Rails.application
+  end
+
+  before { allow(AnyCable.broadcast_adapter).to receive(:raw_broadcast) }
+
+  it "delivers broadcasts in a single batch" do
+    post "/broadcasts", params: {count: 3}
+    expect(AnyCable.broadcast_adapter).to have_received(:raw_broadcast).once
+  end
+
+  context "when auto_batching disabled" do
+    it "delivers broadcast individually" do
+      post "/broadcasts", params: {count: 4, disable_auto_batching: true}
+      expect(AnyCable.broadcast_adapter).to have_received(:raw_broadcast).exactly(4).times
+    end
+  end
+end