From c7105496abb9c88043808878adb33057f0722cd4 Mon Sep 17 00:00:00 2001
From: Vladimir Dementyev <palkan@hey.com>
Date: Mon, 1 Apr 2024 14:58:58 -0700
Subject: [PATCH] feat: stream_from(..., whisper: true)

---
 CHANGELOG.md                                  |  6 +++
 docs/extensions.md                            | 39 +++++++++++++++++++
 .../rails/action_cable_ext/channel.rb         | 10 ++++-
 lib/anycable/rails/compatibility.rb           |  2 +-
 spec/dummy/app/channels/test_channel.rb       |  2 +-
 spec/integrations/rpc/subscriptions_spec.rb   |  1 +
 6 files changed, 57 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 77e1b9d..c5d279b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@
 
 ## master
 
+- Allow specifying the _whispering_ stream via `#stream_from(..., whisper: true)`. ([@palkan][])
+
+  You can use specify the stream to use for _whispering_ (client-initiated broadcasts) by adding `whisper: true` to the `#stream_from` (or `#stream_for`) method call.
+
+  NOTE: This feature is only supported when using AnyCable server and ignored otherwise.
+
 ## 1.5.0.rc.1
 
 - Support passing objects to `ActionCable.server.broadcast`. ([@palkan][])
diff --git a/docs/extensions.md b/docs/extensions.md
index 2985c05..ad95454 100644
--- a/docs/extensions.md
+++ b/docs/extensions.md
@@ -66,6 +66,43 @@ Auto-batching uses [Rails executor](https://guides.rubyonrails.org/threading_and
 
 This feature is only supported when using AnyCable.
 
+## Whispering
+
+AnyCable supports _whispering_, or client-initiated broadcasts. A typical use-case for whispering is sending typing notifications in messaging apps or sharing cursor positions. Here is an example client-side code leveraging whispers (using [AnyCable JS][anycable-client]):
+
+```js
+let channel = cable.subscribeTo("ChatChannel", {id: 42});
+
+channel.on("message", (msg) => {
+  if (msg.event === "typing") {
+    console.log(`user ${msg.name} is typing`);
+  }
+})
+
+// publishing whispers
+const { user } = getCurrentUser();
+
+channel.whisper({event: "typing", name})
+```
+
+You MUST explicitly enable whispers in your channel class as follows:
+
+```ruby
+class ChatChannel < ApplicationCable::Channel
+  def subscribed
+    room = Chat::Room.find(params[:id])
+
+    stream_for room, whisper: true
+  end
+end
+```
+
+Adding `whisper: true` to the stream subscription enables **sending** broadcasts for this client; all subscribed client receive whispers (as regular broadcasts).
+
+**IMPORTANT:** There can be only one whisper stream per channel subscription (since from the protocol perspective clients don't know about streams).
+
+**NOTE:** This feature requires AnyCable server and it's ignored otherwise.
+
 ## Helpers
 
 AnyCable provides a few helpers you can use in your views:
@@ -102,3 +139,5 @@ module ApplicationCable
   end
 end
 ```
+
+[anycable-client]: https://github.com/anycable/anycable-client
diff --git a/lib/anycable/rails/action_cable_ext/channel.rb b/lib/anycable/rails/action_cable_ext/channel.rb
index b415337..068acf2 100644
--- a/lib/anycable/rails/action_cable_ext/channel.rb
+++ b/lib/anycable/rails/action_cable_ext/channel.rb
@@ -22,12 +22,20 @@ def stop_periodic_timers
     super unless anycabled?
   end
 
-  def stream_from(broadcasting, _callback = nil, **)
+  def stream_from(broadcasting, _callback = nil, **opts)
+    whispering = opts.delete(:whisper)
     return super unless anycabled?
 
     broadcasting = String(broadcasting)
 
     connection.anycable_socket.subscribe identifier, broadcasting
+    if whispering
+      connection.anycable_socket.whisper identifier, broadcasting
+    end
+  end
+
+  def stream_for(model, callback = nil, **opts, &block)
+    stream_from(broadcasting_for(model), callback || block, **opts)
   end
 
   def stop_stream_from(broadcasting)
diff --git a/lib/anycable/rails/compatibility.rb b/lib/anycable/rails/compatibility.rb
index 0351f9c..4f367c3 100644
--- a/lib/anycable/rails/compatibility.rb
+++ b/lib/anycable/rails/compatibility.rb
@@ -11,7 +11,7 @@ module Compatibility # :nodoc:
     ]
 
     ActionCable::Channel::Base.prepend(Module.new do
-      def stream_from(broadcasting, callback = nil, coder: nil)
+      def stream_from(broadcasting, callback = nil, coder: nil, **)
         if coder.present? && coder != ActiveSupport::JSON
           raise AnyCable::CompatibilityError, "Custom coders are not supported by AnyCable"
         end
diff --git a/spec/dummy/app/channels/test_channel.rb b/spec/dummy/app/channels/test_channel.rb
index ed83a23..d0830f6 100644
--- a/spec/dummy/app/channels/test_channel.rb
+++ b/spec/dummy/app/channels/test_channel.rb
@@ -7,7 +7,7 @@ def subscribed
     if current_user.secret != "123"
       reject
     else
-      stream_from "test"
+      stream_from "test", whisper: true
     end
   end
 
diff --git a/spec/integrations/rpc/subscriptions_spec.rb b/spec/integrations/rpc/subscriptions_spec.rb
index 4b590c2..6803a74 100644
--- a/spec/integrations/rpc/subscriptions_spec.rb
+++ b/spec/integrations/rpc/subscriptions_spec.rb
@@ -30,6 +30,7 @@
         expect(subject.streams).to eq ["test"]
         expect(subject.stop_streams).to eq false
         expect(subject.transmissions.first).to include("confirm_subscription")
+        expect(subject.istate["$w"]).to eq "test"
       end
     end