Merge branch 'master' of github.com:basecamp/action_cable

4 files changed, 26 insertions, 22 deletions

diff --git a/lib/action_cable/broadcaster.rb b/lib/action_cable/broadcaster.rb
index 38c56b8262..7d8cc90970 100644
--- a/lib/action_cable/broadcaster.rb
+++ b/lib/action_cable/broadcaster.rb
@@ -10,8 +10,8 @@ module ActionCable
     end
 
     def broadcast(message)
+      logger.info "[ActionCable] Broadcasting to #{channel}: #{message}"
       redis.publish channel, message.to_json
-      logger.info "[ActionCable] Broadcasting to channel (#{channel}): #{message}"
     end
   end
 end
diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index 6060ccf681..e6ca45ddcc 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -31,6 +31,12 @@ module ActionCable
         perform_connection
       end
 
+      def perform_connection
+        logger.info "#{channel_name} connecting"
+        connect
+        run_subscribe_callbacks
+      end
+
       def perform_action(data)
         if authorized?
           action = extract_action(data)
@@ -52,11 +58,6 @@ module ActionCable
         logger.info "#{channel_name} disconnected"
       end
 
-      def perform_connection
-        logger.info "#{channel_name} connecting"
-        connect
-        run_subscribe_callbacks
-      end
 
       protected
         # Override in subclasses
@@ -78,10 +79,10 @@ module ActionCable
         end
 
 
-        def broadcast(data)
+        def transmit(data, via: nil)
           if authorized?
-            logger.info "#{channel_name} broadcasting #{data.inspect}"
-            connection.broadcast({ identifier: @channel_identifier, message: data }.to_json)
+            logger.info "#{channel_name} transmitting #{data.inspect}".tap { |m| m << " (via #{via})" if via }
+            connection.transmit({ identifier: @channel_identifier, message: data }.to_json)
           else
             unauthorized
           end
diff --git a/lib/action_cable/channel/redis.rb b/lib/action_cable/channel/redis.rb
index b5fc812919..0f77dc0418 100644
--- a/lib/action_cable/channel/redis.rb
+++ b/lib/action_cable/channel/redis.rb
@@ -14,14 +14,14 @@ module ActionCable
         @_redis_channels << [ redis_channel, callback ]
 
         pubsub.subscribe(redis_channel, &callback)
-        logger.info "#{channel_name} subscribed to incoming actions from #{redis_channel}"
+        logger.info "#{channel_name} subscribed to broadcasts from #{redis_channel}"
       end
 
       def unsubscribe_from_all_channels
         if @_redis_channels
           @_redis_channels.each do |redis_channel, callback|
             pubsub.unsubscribe_proc(redis_channel, callback)
-            logger.info "#{channel_name} unsubscribed from incoming actions #{redis_channel}"
+            logger.info "#{channel_name} unsubscribed to broadcasts from #{redis_channel}"
           end
         end
       end
@@ -29,8 +29,7 @@ module ActionCable
       protected
         def default_subscription_callback(channel)
           -> (message) do
-            logger.info "Received a message over the redis channel: #{channel}"
-            broadcast ActiveSupport::JSON.decode(message)
+            transmit ActiveSupport::JSON.decode(message), via: "broadcast from #{channel}"
           end
         end
     end
diff --git a/lib/action_cable/connection/base.rb b/lib/action_cable/connection/base.rb
index c36b7bbc1a..89d0844031 100644
--- a/lib/action_cable/connection/base.rb
+++ b/lib/action_cable/connection/base.rb
@@ -34,8 +34,8 @@ module ActionCable
           @websocket = Faye::WebSocket.new(@env)
 
           @websocket.on(:open) do |event|
-            broadcast_ping_timestamp
-            @ping_timer = EventMachine.add_periodic_timer(PING_INTERVAL) { broadcast_ping_timestamp }
+            transmit_ping_timestamp
+            @ping_timer = EventMachine.add_periodic_timer(PING_INTERVAL) { transmit_ping_timestamp }
             worker_pool.async.invoke(self, :initialize_connection)
           end
 
@@ -87,7 +87,7 @@ module ActionCable
         end
       end
 
-      def broadcast(data)
+      def transmit(data)
         @websocket.send data
       end
 
@@ -109,6 +109,14 @@ module ActionCable
       end
 
       protected
+        def request
+          @request ||= ActionDispatch::Request.new(Rails.application.env_config.merge(env))
+        end
+
+        def cookies
+          request.cookie_jar
+        end
+
         def initialize_connection
           server.add_connection(self)
 
@@ -127,8 +135,8 @@ module ActionCable
           disconnect if respond_to?(:disconnect)
         end
 
-        def broadcast_ping_timestamp
-          broadcast({ identifier: '_ping', message: Time.now.to_i }.to_json)
+        def transmit_ping_timestamp
+          transmit({ identifier: '_ping', message: Time.now.to_i }.to_json)
         end
 
         def subscribe_channel(data)
@@ -173,10 +181,6 @@ module ActionCable
           @websocket && @websocket.ready_state == Faye::WebSocket::API::OPEN
         end
 
-        def request
-          @request ||= ActionDispatch::Request.new(env)
-        end
-
         def websocket?
           @is_websocket ||= Faye::WebSocket.websocket?(@env)
         end