2 files changed, 32 insertions, 4 deletions
diff --git a/lib/action_cable/channel/base.rb b/lib/action_cable/channel/base.rb
index 9cfeb4b73a..8ee99649f4 100644
--- a/lib/action_cable/channel/base.rb
+++ b/lib/action_cable/channel/base.rb
@@ -35,8 +35,16 @@ module ActionCable
         subscribe
       end
 
-      def receive(data)
-        raise "Not implemented"
+      def receive_data(data)
+        if authorized?
+          if respond_to?(:receive)
+            receive(data)
+          else
+            logger.error "[ActionCable] #{self.class.name} received data (#{data}) but #{self.class.name}#receive callback is not defined"
+          end
+        else
+          unauthorized
+        end
       end
 
       def subscribe
@@ -52,6 +60,15 @@ module ActionCable
       end
 
       protected
+        # Override in subclasses
+        def authorized?
+          true
+        end
+
+        def unauthorized
+          logger.error "[ActionCable] Unauthorized access to #{self.class.name}"
+        end
+
         def connect
           # Override in subclasses
         end
@@ -61,7 +78,11 @@ module ActionCable
         end
 
         def broadcast(data)
-          connection.broadcast({ identifier: @channel_identifier, message: data }.to_json)
+          if authorized?
+            connection.broadcast({ identifier: @channel_identifier, message: data }.to_json)
+          else
+            unauthorized
+          end
         end
 
         def start_periodic_timers
@@ -80,6 +101,10 @@ module ActionCable
           connection.worker_pool
         end
 
+        def logger
+          connection.logger
+        end
+
     end
 
   end
diff --git a/lib/action_cable/server.rb b/lib/action_cable/server.rb
index 2449837105..a0c128129d 100644
--- a/lib/action_cable/server.rb
+++ b/lib/action_cable/server.rb
@@ -24,6 +24,8 @@ module ActionCable
       end
     end
 
+    attr_reader :env
+
     def initialize(env)
       @env = env
     end
@@ -37,6 +39,7 @@ module ActionCable
         @websocket.on(:open) do |event|
           broadcast_ping_timestamp
           @ping_timer = EventMachine.add_periodic_timer(PING_INTERVAL) { broadcast_ping_timestamp }
+          worker_pool.async.invoke(self, :connect) if respond_to?(:connect)
         end
 
         @websocket.on(:message) do |event|
@@ -106,7 +109,7 @@ module ActionCable
 
       def process_message(message)
         if @subscriptions[message['identifier']]
-          @subscriptions[message['identifier']].receive(ActiveSupport::JSON.decode message['data'])
+          @subscriptions[message['identifier']].receive_data(ActiveSupport::JSON.decode message['data'])
         else
           logger.error "Unable to process message: #{message}"
         end