feat: add command callbacks support and backport

Author: palkan

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## master
 
+- Add support and backport for Connection command callbacks. ([@palkan][])
+
 ## 1.3.3 (2022-04-20)
 
 - Added `sid` (unique connection identifier) field to the `welcome` message if present. ([@palkan][])

docs/compatibility.md

@@ -17,11 +17,14 @@ Streaming | ✅
 Broadcasting | ✅
 Periodical timers | 🚫
 Disconnect remote clients | ✅
+Command callbacks | ✅ \*\*\*
 
 \* See [restoring state objects](../architecture.md#restoring-state-objects) for more information on how identifiers work.
 
 \*\* See [channel state](./channels_state.md) for more information on subscription instance variables support.
 
+\*\*\* AnyCable (via `anycable-rails`) also supports [command callbacks](https://github.com/rails/rails/pull/44696) (`before_command`, `after_command`, `around_command`) for older Rails versions (event when not using AnyCable).
+
 ## Runtime checks
 
 AnyCable provides a way to enforce compatibility through runtime checks.

lib/anycable/rails/action_cable_ext/connection.rb

@@ -28,3 +28,30 @@ def request
     @request ||= anycable_request_builder.build_rack_request(@env)
   end
 end)
+
+# Backport command callbacks: https://github.com/rails/rails/pull/44696
+unless ActionCable::Connection::Base.respond_to?(:before_command)
+  ActionCable::Connection::Base.include ActiveSupport::Callbacks
+  ActionCable::Connection::Base.define_callbacks :command
+  ActionCable::Connection::Base.extend(Module.new do
+    def before_command(*methods, &block)
+      set_callback(:command, :before, *methods, &block)
+    end
+
+    def after_command(*methods, &block)
+      set_callback(:command, :after, *methods, &block)
+    end
+
+    def around_command(*methods, &block)
+      set_callback(:command, :around, *methods, &block)
+    end
+  end)
+
+  ActionCable::Connection::Base.prepend(Module.new do
+    def dispatch_websocket_message(*)
+      return super unless websocket.alive?
+
+      run_callbacks(:command) { super }
+    end
+  end)
+end

lib/anycable/rails/connection.rb

@@ -110,22 +110,24 @@ def handle_close
       end
 
       def handle_channel_command(identifier, command, data)
-        # We cannot use subscriptions#execute_command here,
-        # since we MUST return true of false, depending on the status
-        # of execution
-        channel = conn.subscriptions.fetch(identifier)
-        case command
-        when "subscribe"
-          channel.handle_subscribe
-          !channel.rejected?
-        when "unsubscribe"
-          conn.subscriptions.remove_subscription(channel)
-          true
-        when "message"
-          channel.perform_action ActiveSupport::JSON.decode(data)
-          true
-        else
-          false
+        conn.run_callbacks :command do
+          # We cannot use subscriptions#execute_command here,
+          # since we MUST return true of false, depending on the status
+          # of execution
+          channel = conn.subscriptions.fetch(identifier)
+          case command
+          when "subscribe"
+            channel.handle_subscribe
+            !channel.rejected?
+          when "unsubscribe"
+            conn.subscriptions.remove_subscription(channel)
+            true
+          when "message"
+            channel.perform_action ActiveSupport::JSON.decode(data)
+            true
+          else
+            false
+          end
         end
       # Support rescue_from
       # https://github.com/rails/rails/commit/d2571e560c62116f60429c933d0c41a0e249b58b

spec/dummy/app/channels/application_cable/connection.rb

@@ -22,6 +22,8 @@ def log_event(source, data)
 
     state_attr_accessor :token
 
+    around_command :log_command_execution
+
     def connect
       self.current_user = verify_user
       self.url = request.url if current_user
@@ -34,6 +36,12 @@ def disconnect
 
     private
 
+    def log_command_execution
+      start = Time.now
+      yield
+      self.class.log_event("command", time: Time.now - start)
+    end
+
     def verify_user
       return env["warden"]&.user if env["warden"]&.user
 

spec/integrations/rpc/perform_spec.rb

@@ -51,6 +51,16 @@
       end
     end
 
+    context "with arround_command" do
+      let(:log) { ApplicationCable::Connection.events_log }
+
+      it "executes command callbacks" do
+        expect { subject }.to change { log.size }.by(1)
+
+        expect(log.last[:source]).to eq "command"
+      end
+    end
+
     describe "#stop_stream_from" do
       let(:data) { {action: "unfollow_all"} }