diff --git a/lib/anycable/rails/action_cable_ext/channel.rb b/lib/anycable/rails/action_cable_ext/channel.rb index 068acf2..ba3d302 100644 --- a/lib/anycable/rails/action_cable_ext/channel.rb +++ b/lib/anycable/rails/action_cable_ext/channel.rb @@ -24,6 +24,11 @@ def stop_periodic_timers def stream_from(broadcasting, _callback = nil, **opts) whispering = opts.delete(:whisper) + if whispering + self.class.state_attr_accessor(:whisper_stream) unless respond_to?(:whisper_stream) + self.whisper_stream = broadcasting + end + return super unless anycabled? broadcasting = String(broadcasting) @@ -50,6 +55,12 @@ def stop_all_streams connection.anycable_socket.unsubscribe_from_all identifier end + def whisper(payload) + return unless whisper_stream + + broadcast_to whisper_stream, payload + end + private def anycabled? diff --git a/lib/anycable/rails/action_cable_ext/whisper_stream.rb b/lib/anycable/rails/action_cable_ext/whisper_stream.rb new file mode 100644 index 0000000..aece7da --- /dev/null +++ b/lib/anycable/rails/action_cable_ext/whisper_stream.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +require "action_cable" + +ActionCable::Connection::Subscriptions.prepend(Module.new do + def execute_command(data) + return whisper(data) if data["command"] == "whisper" + + super + end + + def whisper(data) + find(data).whisper(ActiveSupport::JSON.decode(data["data"])) + end +end) diff --git a/lib/anycable/rails/ext.rb b/lib/anycable/rails/ext.rb index d9e6399..eca04c7 100644 --- a/lib/anycable/rails/ext.rb +++ b/lib/anycable/rails/ext.rb @@ -5,6 +5,7 @@ module Rails module Ext autoload :JWT, "anycable/rails/ext/jwt" autoload :SignedStreams, "anycable/rails/ext/signed_streams" + autoload :WhisperStreams, "anycable/rails/ext/whisper_stream" end end end