diff --git a/app/models/solid_cable/message.rb b/app/models/solid_cable/message.rb index 398af1b..4a16d70 100644 --- a/app/models/solid_cable/message.rb +++ b/app/models/solid_cable/message.rb @@ -15,6 +15,13 @@ def broadcast(channel, payload) channel_hash: channel_hash_for(channel) }) end + def broadcast_batch(broadcasts) + created_at = Time.current + insert_all broadcasts.map { |channel, payload| + { created_at:, channel:, payload:, channel_hash: channel_hash_for(channel) } + } + end + # Need to unpack this as a signed integer since Postgresql and SQLite # don't support unsigned integers def channel_hash_for(channel) diff --git a/lib/action_cable/subscription_adapter/solid_cable.rb b/lib/action_cable/subscription_adapter/solid_cable.rb index e1b9f43..df14d0b 100644 --- a/lib/action_cable/subscription_adapter/solid_cable.rb +++ b/lib/action_cable/subscription_adapter/solid_cable.rb @@ -20,10 +20,11 @@ def initialize(*) end @listener = nil + @broadcaster = nil end def broadcast(channel, payload) - ::SolidCable::Message.broadcast(channel, payload) + broadcaster.broadcast(channel, payload) ::SolidCable::TrimJob.perform_now if ::SolidCable.autotrim? end @@ -36,7 +37,10 @@ def unsubscribe(channel, callback) listener.remove_subscriber(channel, callback) end - delegate :shutdown, to: :listener + def shutdown + @broadcaster&.shutdown + @listener&.shutdown + end private def listener @@ -45,6 +49,12 @@ def listener end end + def broadcaster + @broadcaster || @mutex.synchronize do + @broadcaster ||= ::SolidCable::BatchedBroadcaster.new + end + end + def pubsub_executor @pubsub_executor ||= if respond_to?(:executor, true) diff --git a/lib/generators/solid_cable/install/templates/config/cable.yml b/lib/generators/solid_cable/install/templates/config/cable.yml index b9adc5a..8db79af 100644 --- a/lib/generators/solid_cable/install/templates/config/cable.yml +++ b/lib/generators/solid_cable/install/templates/config/cable.yml @@ -15,3 +15,5 @@ production: writing: cable polling_interval: 0.1.seconds message_retention: 1.day + writer_batch_size: 4 + writer_batch_delay: 0.001.seconds diff --git a/lib/solid_cable.rb b/lib/solid_cable.rb index 2694896..03c434f 100644 --- a/lib/solid_cable.rb +++ b/lib/solid_cable.rb @@ -3,13 +3,15 @@ require "solid_cable/version" require "solid_cable/engine" require "solid_cable/configuration" +require "solid_cable/batched_broadcaster" require "action_cable/subscription_adapter/solid_cable" module SolidCable class << self delegate :connects_to, :silence_polling?, :polling_interval, :message_retention, :autotrim?, :trim_batch_size, :use_skip_locked, - :trim_chance, :reconnect_attempts, to: :configuration + :trim_chance, :reconnect_attempts, :writer_batch_size, :writer_batch_delay, + to: :configuration def configuration @configuration ||= Configuration.new(**Rails.application.config_for("cable")) diff --git a/lib/solid_cable/batched_broadcaster.rb b/lib/solid_cable/batched_broadcaster.rb new file mode 100644 index 0000000..53f2ec8 --- /dev/null +++ b/lib/solid_cable/batched_broadcaster.rb @@ -0,0 +1,84 @@ +# frozen_string_literal: true + +module SolidCable + class BatchedBroadcaster + Stopped = Class.new(StandardError) + Message = Struct.new(:channel, :payload, keyword_init: true) + + def initialize(batch_size: SolidCable.writer_batch_size, batch_delay: SolidCable.writer_batch_delay) + @batch_size = batch_size + @batch_delay = batch_delay + @queue = Queue.new + + @thread = Thread.new do + Thread.current.name = "solid_cable_writer" + Thread.current.abort_on_exception = true + listen_for_initial_messages + end + end + + def broadcast(channel, payload) + message = Message.new(channel:, payload:) + + queue.enq message + rescue ClosedQueueError + raise Stopped, "Solid Cable writer has stopped" + end + + def shutdown + queue.close + thread.join + end + + private + attr_reader :batch_size, :batch_delay, :queue, :thread, :queue_size + + def listen_for_initial_messages + loop do + message = queue.pop + + break if message.nil? + + collect_batch(message) + end + end + + def collect_batch(first_message) + batch = [ first_message ] + deadline = monotonic_time + batch_delay + + drain_queue_into(batch) + wait_for_messages_until(batch, deadline) + + flush(batch) + end + + def drain_queue_into(batch) + while batch.size < batch_size && (message = queue.pop(timeout: 0)) + batch << message + end + end + + def wait_for_messages_until(batch, deadline) + while batch.size < batch_size && (remaining = deadline - monotonic_time).positive? + message = queue.pop(timeout: remaining) + break if message.nil? + + batch << message + end + end + + def flush(batch) + Rails.application.executor.wrap do + SolidCable::Message. + broadcast_batch(batch.map { |message| [ message.channel, message.payload ] }) + end + rescue StandardError => error + Rails.error.report(error) + end + + def monotonic_time + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + end +end diff --git a/lib/solid_cable/configuration.rb b/lib/solid_cable/configuration.rb index 236f640..f29922a 100644 --- a/lib/solid_cable/configuration.rb +++ b/lib/solid_cable/configuration.rb @@ -6,7 +6,7 @@ def initialize(**options) attr_writer :connects_to, :silence_polling, :polling_interval, :message_retention, :autotrim, :trim_batch_size, :use_skip_locked, - :trim_chance, :reconnect_attempts + :trim_chance, :reconnect_attempts, :writer_batch_size, :writer_batch_delay def connects_to @connects_to ||= options.connects_to.to_h.deep_transform_values(&:to_sym) @@ -66,6 +66,15 @@ def reconnect_attempts end end + def writer_batch_size + @writer_batch_size ||= [ (options.writer_batch_size || 4).to_i, 1 ].max + end + + def writer_batch_delay + @writer_batch_delay ||= + [ parse_duration(options.writer_batch_delay, default: 0.001.seconds), 0 ].max + end + private attr_reader :options diff --git a/test/lib/action_cable/subscription_adapter/solid_cable_test.rb b/test/lib/action_cable/subscription_adapter/solid_cable_test.rb index 4e1e0c3..87e7899 100644 --- a/test/lib/action_cable/subscription_adapter/solid_cable_test.rb +++ b/test/lib/action_cable/subscription_adapter/solid_cable_test.rb @@ -151,6 +151,7 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase test "does not send old messages" do @tx_adapter.broadcast("channel", "channel1") @tx_adapter.broadcast("channel", "channel2") + wait_for_messages("channel1", "channel2") subscribe_as_queue("channel") do |queue| assert_empty queue @@ -159,6 +160,7 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase @tx_adapter.broadcast("channel", "channel4") @tx_adapter.broadcast("other", "other1") @tx_adapter.broadcast("other", "other2") + wait_for_messages("channel3", "channel4", "other1", "other2") subscribe_as_queue("other") do |other_queue| assert_empty other_queue @@ -169,6 +171,7 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase @tx_adapter.broadcast("channel", "channel5") @tx_adapter.broadcast("channel", "channel6") + wait_for_messages("channel5", "channel6") subscribe_as_queue("channel") do |queue| assert_empty queue @@ -258,4 +261,10 @@ def with_active_record_logger(logger) def next_message_in_queue(queue) Timeout.timeout(5, nil, "Failed to get next item in queue") { queue.pop } end + + def wait_for_messages(*payloads) + Timeout.timeout(5, nil, "Failed to persist broadcasts") do + sleep 0.001 until SolidCable::Message.where(payload: payloads).count == payloads.size + end + end end diff --git a/test/lib/solid_cable/batched_broadcaster_test.rb b/test/lib/solid_cable/batched_broadcaster_test.rb new file mode 100644 index 0000000..03edaef --- /dev/null +++ b/test/lib/solid_cable/batched_broadcaster_test.rb @@ -0,0 +1,50 @@ +# frozen_string_literal: true + +require "test_helper" + +class SolidCable::BatchedBroadcasterTest < ActiveSupport::TestCase + teardown do + @broadcaster&.shutdown + end + + test "batches concurrent broadcasts" do + @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 2, batch_delay: 1) + batches = Queue.new + + SolidCable::Message.stub(:broadcast_batch, ->(batch) { batches << batch }) do + threads = [ + Thread.new { @broadcaster.broadcast("one", "first") }, + Thread.new { @broadcaster.broadcast("two", "second") } + ] + threads.each(&:join) + @broadcaster.shutdown + end + + assert_equal 1, batches.size + assert_equal [ [ "one", "first" ], [ "two", "second" ] ], batches.pop.sort + end + + test "rejects broadcasts after shutdown" do + @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 2, batch_delay: 0) + @broadcaster.shutdown + + assert_raises(SolidCable::BatchedBroadcaster::Stopped) do + @broadcaster.broadcast("channel", "payload") + end + end + + test "reports write errors" do + @broadcaster = SolidCable::BatchedBroadcaster.new(batch_size: 1, batch_delay: 0) + write_error = RuntimeError.new("write failed") + reported_errors = Queue.new + + Rails.error.stub(:report, ->(error, **) { reported_errors << error }) do + SolidCable::Message.stub(:broadcast_batch, ->(*) { raise write_error }) do + @broadcaster.broadcast("channel", "payload") + @broadcaster.shutdown + end + end + + assert_same write_error, reported_errors.pop + end +end