6 files changed, 340 insertions, 81 deletions

diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb
index d41bf3064b..9a3a3581e6 100644
--- a/actioncable/test/channel/base_test.rb
+++ b/actioncable/test/channel/base_test.rb
@@ -1,6 +1,6 @@
-require 'test_helper'
-require 'stubs/test_connection'
-require 'stubs/room'
+require "test_helper"
+require "stubs/test_connection"
+require "stubs/room"
 
 class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
   class ActionCable::Channel::Base
@@ -58,7 +58,7 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
     end
 
     def get_latest
-      transmit data: 'latest'
+      transmit data: "latest"
     end
 
     def receive
@@ -74,14 +74,16 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
   setup do
     @user = User.new "lifo"
     @connection = TestConnection.new(@user)
-    @channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
+    @channel = ChatChannel.new @connection, "{id: 1}", id: 1
   end
 
-  test "should subscribe to a channel on initialize" do
+  test "should subscribe to a channel" do
+    @channel.subscribe_to_channel
     assert_equal 1, @channel.room.id
   end
 
   test "on subscribe callbacks" do
+    @channel.subscribe_to_channel
     assert @channel.subscribed
   end
 
@@ -90,6 +92,8 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
   end
 
   test "unsubscribing from a channel" do
+    @channel.subscribe_to_channel
+
     assert @channel.room
     assert @channel.subscribed?
 
@@ -104,54 +108,59 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
   end
 
   test "callable action without any argument" do
-    @channel.perform_action 'action' => :leave
+    @channel.perform_action "action" => :leave
     assert_equal [ :leave ], @channel.last_action
   end
 
   test "callable action with arguments" do
-    data = { 'action' => :speak, 'content' => "Hello World" }
+    data = { "action" => :speak, "content" => "Hello World" }
 
     @channel.perform_action data
     assert_equal [ :speak, data ], @channel.last_action
   end
 
   test "should not dispatch a private method" do
-    @channel.perform_action 'action' => :rm_rf
+    @channel.perform_action "action" => :rm_rf
     assert_nil @channel.last_action
   end
 
   test "should not dispatch a public method defined on Base" do
-    @channel.perform_action 'action' => :kick
+    @channel.perform_action "action" => :kick
     assert_nil @channel.last_action
   end
 
   test "should dispatch a public method defined on Base and redefined on channel" do
-    data = { 'action' => :topic, 'content' => "This is Sparta!" }
+    data = { "action" => :topic, "content" => "This is Sparta!" }
 
     @channel.perform_action data
     assert_equal [ :topic, data ], @channel.last_action
   end
 
   test "should dispatch calling a public method defined in an ancestor" do
-    @channel.perform_action 'action' => :chatters
+    @channel.perform_action "action" => :chatters
     assert_equal [ :chatters ], @channel.last_action
   end
 
   test "should dispatch receive action when perform_action is called with empty action" do
-    data = { 'content' => 'hello' }
+    data = { "content" => "hello" }
     @channel.perform_action data
     assert_equal [ :receive ], @channel.last_action
   end
 
   test "transmitting data" do
-    @channel.perform_action 'action' => :get_latest
+    @channel.perform_action "action" => :get_latest
 
-    expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "message" => { "data" => "latest" }
+    expected = { "identifier" => "{id: 1}", "message" => { "data" => "latest" } }
     assert_equal expected, @connection.last_transmission
   end
 
-  test "subscription confirmation" do
-    expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
+  test "do not send subscription confirmation on initialize" do
+    assert_nil @connection.last_transmission
+  end
+
+  test "subscription confirmation on subscribe_to_channel" do
+    expected = { "identifier" => "{id: 1}", "type" => "confirm_subscription" }
+    @channel.subscribe_to_channel
     assert_equal expected, @connection.last_transmission
   end
 
@@ -162,7 +171,84 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
 
   test "invalid action on Channel" do
     assert_logged("Unable to process ActionCable::Channel::BaseTest::ChatChannel#invalid_action") do
-      @channel.perform_action 'action' => :invalid_action
+      @channel.perform_action "action" => :invalid_action
+    end
+  end
+
+  test "notification for perform_action" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe "perform_action.action_cable" do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      data = { "action" => :speak, "content" => "hello" }
+      @channel.perform_action data
+
+      assert_equal 1, events.length
+      assert_equal "perform_action.action_cable", events[0].name
+      assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class]
+      assert_equal :speak, events[0].payload[:action]
+      assert_equal data, events[0].payload[:data]
+    ensure
+      ActiveSupport::Notifications.unsubscribe "perform_action.action_cable"
+    end
+  end
+
+  test "notification for transmit" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe "transmit.action_cable" do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      @channel.perform_action "action" => :get_latest
+      expected_data = { data: "latest" }
+
+      assert_equal 1, events.length
+      assert_equal "transmit.action_cable", events[0].name
+      assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class]
+      assert_equal expected_data, events[0].payload[:data]
+      assert_nil events[0].payload[:via]
+    ensure
+      ActiveSupport::Notifications.unsubscribe "transmit.action_cable"
+    end
+  end
+
+  test "notification for transmit_subscription_confirmation" do
+    begin
+      @channel.subscribe_to_channel
+
+      events = []
+      ActiveSupport::Notifications.subscribe "transmit_subscription_confirmation.action_cable" do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      @channel.stubs(:subscription_confirmation_sent?).returns(false)
+      @channel.send(:transmit_subscription_confirmation)
+
+      assert_equal 1, events.length
+      assert_equal "transmit_subscription_confirmation.action_cable", events[0].name
+      assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class]
+    ensure
+      ActiveSupport::Notifications.unsubscribe "transmit_subscription_confirmation.action_cable"
+    end
+  end
+
+  test "notification for transmit_subscription_rejection" do
+    begin
+      events = []
+      ActiveSupport::Notifications.subscribe "transmit_subscription_rejection.action_cable" do |*args|
+        events << ActiveSupport::Notifications::Event.new(*args)
+      end
+
+      @channel.send(:transmit_subscription_rejection)
+
+      assert_equal 1, events.length
+      assert_equal "transmit_subscription_rejection.action_cable", events[0].name
+      assert_equal "ActionCable::Channel::BaseTest::ChatChannel", events[0].payload[:channel_class]
+    ensure
+      ActiveSupport::Notifications.unsubscribe "transmit_subscription_rejection.action_cable"
     end
   end
 
diff --git a/actioncable/test/channel/broadcasting_test.rb b/actioncable/test/channel/broadcasting_test.rb
index 1de04243e5..3476c1db31 100644
--- a/actioncable/test/channel/broadcasting_test.rb
+++ b/actioncable/test/channel/broadcasting_test.rb
@@ -1,6 +1,6 @@
-require 'test_helper'
-require 'stubs/test_connection'
-require 'stubs/room'
+require "test_helper"
+require "stubs/test_connection"
+require "stubs/room"
 
 class ActionCable::Channel::BroadcastingTest < ActiveSupport::TestCase
   class ChatChannel < ActionCable::Channel::Base
@@ -11,7 +11,7 @@ class ActionCable::Channel::BroadcastingTest < ActiveSupport::TestCase
   end
 
   test "broadcasts_to" do
-    ActionCable.stubs(:server).returns mock().tap { |m| m.expects(:broadcast).with('action_cable:channel:broadcasting_test:chat:Room#1-Campfire', "Hello World") }
+    ActionCable.stubs(:server).returns mock().tap { |m| m.expects(:broadcast).with("action_cable:channel:broadcasting_test:chat:Room#1-Campfire", "Hello World") }
     ChatChannel.broadcast_to(Room.new(1), "Hello World")
   end
 
diff --git a/actioncable/test/channel/naming_test.rb b/actioncable/test/channel/naming_test.rb
index 89ef6ad8b0..08f0e7be48 100644
--- a/actioncable/test/channel/naming_test.rb
+++ b/actioncable/test/channel/naming_test.rb
@@ -1,4 +1,4 @@
-require 'test_helper'
+require "test_helper"
 
 class ActionCable::Channel::NamingTest < ActiveSupport::TestCase
   class ChatChannel < ActionCable::Channel::Base
diff --git a/actioncable/test/channel/periodic_timers_test.rb b/actioncable/test/channel/periodic_timers_test.rb
index 64f0247cd6..2ee711fd29 100644
--- a/actioncable/test/channel/periodic_timers_test.rb
+++ b/actioncable/test/channel/periodic_timers_test.rb
@@ -1,12 +1,21 @@
-require 'test_helper'
-require 'stubs/test_connection'
-require 'stubs/room'
+require "test_helper"
+require "stubs/test_connection"
+require "stubs/room"
+require "active_support/time"
 
 class ActionCable::Channel::PeriodicTimersTest < ActiveSupport::TestCase
   class ChatChannel < ActionCable::Channel::Base
-    periodically -> { ping }, every: 5
+    # Method name arg
     periodically :send_updates, every: 1
 
+    # Proc arg
+    periodically -> { ping }, every: 2
+
+    # Block arg
+    periodically every: 3 do
+      ping
+    end
+
     private
       def ping
       end
@@ -19,22 +28,42 @@ class ActionCable::Channel::PeriodicTimersTest < ActiveSupport::TestCase
   test "periodic timers definition" do
     timers = ChatChannel.periodic_timers
 
-    assert_equal 2, timers.size
+    assert_equal 3, timers.size
 
-    first_timer = timers[0]
-    assert_kind_of Proc, first_timer[0]
-    assert_equal 5, first_timer[1][:every]
+    timers.each_with_index do |timer, i|
+      assert_kind_of Proc, timer[0]
+      assert_equal i+1, timer[1][:every]
+    end
+  end
 
-    second_timer = timers[1]
-    assert_equal :send_updates, second_timer[0]
-    assert_equal 1, second_timer[1][:every]
+  test "disallow negative and zero periods" do
+    [ 0, 0.0, 0.seconds, -1, -1.seconds, "foo", :foo, Object.new ].each do |invalid|
+      assert_raise ArgumentError, /Expected every:/ do
+        ChatChannel.periodically :send_updates, every: invalid
+      end
+    end
+  end
+
+  test "disallow block and arg together" do
+    assert_raise ArgumentError, /not both/ do
+      ChatChannel.periodically(:send_updates, every: 1) { ping }
+    end
+  end
+
+  test "disallow unknown args" do
+    [ "send_updates", Object.new, nil ].each do |invalid|
+      assert_raise ArgumentError, /Expected a Symbol/ do
+        ChatChannel.periodically invalid, every: 1
+      end
+    end
   end
 
   test "timer start and stop" do
-    Concurrent::TimerTask.expects(:new).times(2).returns(true)
-    channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
+    @connection.server.event_loop.expects(:timer).times(3).returns(stub(shutdown: nil))
+    channel = ChatChannel.new @connection, "{id: 1}", id: 1
 
-    channel.expects(:stop_periodic_timers).once
+    channel.subscribe_to_channel
     channel.unsubscribe_from_channel
+    assert_equal [], channel.send(:active_periodic_timers)
   end
 end
diff --git a/actioncable/test/channel/rejection_test.rb b/actioncable/test/channel/rejection_test.rb
index aa93396d44..99c4a7603a 100644
--- a/actioncable/test/channel/rejection_test.rb
+++ b/actioncable/test/channel/rejection_test.rb
@@ -1,12 +1,15 @@
-require 'test_helper'
-require 'stubs/test_connection'
-require 'stubs/room'
+require "test_helper"
+require "stubs/test_connection"
+require "stubs/room"
 
 class ActionCable::Channel::RejectionTest < ActiveSupport::TestCase
   class SecretChannel < ActionCable::Channel::Base
     def subscribed
       reject if params[:id] > 0
     end
+
+    def secret_action
+    end
   end
 
   setup do
@@ -16,10 +19,23 @@ class ActionCable::Channel::RejectionTest < ActiveSupport::TestCase
 
   test "subscription rejection" do
     @connection.expects(:subscriptions).returns mock().tap { |m| m.expects(:remove_subscription).with instance_of(SecretChannel) }
-    @channel = SecretChannel.new @connection, "{id: 1}", { id: 1 }
+    @channel = SecretChannel.new @connection, "{id: 1}", id: 1
+    @channel.subscribe_to_channel
 
-    expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "reject_subscription"
+    expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" }
     assert_equal expected, @connection.last_transmission
   end
 
+  test "does not execute action if subscription is rejected" do
+    @connection.expects(:subscriptions).returns mock().tap { |m| m.expects(:remove_subscription).with instance_of(SecretChannel) }
+    @channel = SecretChannel.new @connection, "{id: 1}", id: 1
+    @channel.subscribe_to_channel
+
+    expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" }
+    assert_equal expected, @connection.last_transmission
+    assert_equal 1, @connection.transmissions.size
+
+    @channel.perform_action("action" => :secret_action)
+    assert_equal 1, @connection.transmissions.size
+  end
 end
diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb
index 947efd96d4..31dcde2e29 100644
--- a/actioncable/test/channel/stream_test.rb
+++ b/actioncable/test/channel/stream_test.rb
@@ -1,13 +1,21 @@
-require 'test_helper'
-require 'stubs/test_connection'
-require 'stubs/room'
+require "test_helper"
+require "stubs/test_connection"
+require "stubs/room"
+
+module ActionCable::StreamTests
+  class Connection < ActionCable::Connection::Base
+    attr_reader :websocket
+
+    def send_async(method, *args)
+      send method, *args
+    end
+  end
 
-class ActionCable::Channel::StreamTest < ActionCable::TestCase
   class ChatChannel < ActionCable::Channel::Base
     def subscribed
       if params[:id]
         @room = Room.new params[:id]
-        stream_from "test_room_#{@room.id}"
+        stream_from "test_room_#{@room.id}", coder: pick_coder(params[:coder])
       end
     end
 
@@ -15,60 +23,180 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase
       transmit_subscription_confirmation
     end
 
+    private def pick_coder(coder)
+      case coder
+      when nil, "json"
+        ActiveSupport::JSON
+      when "custom"
+        DummyEncoder
+      when "none"
+        nil
+      end
+    end
   end
 
-  test "streaming start and stop" do
-    run_in_eventmachine do
-      connection = TestConnection.new
-      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
-      channel = ChatChannel.new connection, "{id: 1}", { id: 1 }
+  module DummyEncoder
+    extend self
+    def encode(*) '{ "foo": "encoded" }' end
+    def decode(*) { foo: "decoded" } end
+  end
 
-      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
-      channel.unsubscribe_from_channel
+  class SymbolChannel < ActionCable::Channel::Base
+    def subscribed
+      stream_from :channel
     end
   end
 
-  test "stream_for" do
-    run_in_eventmachine do
-      connection = TestConnection.new
-      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
+  class StreamTest < ActionCable::TestCase
+    test "streaming start and stop" do
+      run_in_eventmachine do
+        connection = TestConnection.new
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
+        channel = ChatChannel.new connection, "{id: 1}", id: 1
+        channel.subscribe_to_channel
 
-      channel = ChatChannel.new connection, ""
-      channel.stream_for Room.new(1)
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
+        channel.unsubscribe_from_channel
+      end
     end
-  end
 
-  test "stream_from subscription confirmation" do
-    run_in_eventmachine do
-      connection = TestConnection.new
+    test "stream from non-string channel" do
+      run_in_eventmachine do
+        connection = TestConnection.new
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("channel", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
+        channel = SymbolChannel.new connection, ""
+        channel.subscribe_to_channel
 
-      ChatChannel.new connection, "{id: 1}", { id: 1 }
-      assert_nil connection.last_transmission
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
+        channel.unsubscribe_from_channel
+      end
+    end
 
-      wait_for_async
+    test "stream_for" do
+      run_in_eventmachine do
+        connection = TestConnection.new
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:stream_tests:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
 
-      expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
-      connection.transmit(expected)
+        channel = ChatChannel.new connection, ""
+        channel.subscribe_to_channel
+        channel.stream_for Room.new(1)
+      end
+    end
 
-      assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation within 0.1s"
+    test "stream_from subscription confirmation" do
+      run_in_eventmachine do
+        connection = TestConnection.new
+
+        channel = ChatChannel.new connection, "{id: 1}", id: 1
+        channel.subscribe_to_channel
+
+        assert_nil connection.last_transmission
+
+        wait_for_async
+
+        confirmation = { "identifier" => "{id: 1}", "type" => "confirm_subscription" }
+        connection.transmit(confirmation)
+
+        assert_equal confirmation, connection.last_transmission, "Did not receive subscription confirmation within 0.1s"
+      end
     end
-  end
 
-  test "subscription confirmation should only be sent out once" do
-    run_in_eventmachine do
-      connection = TestConnection.new
+    test "subscription confirmation should only be sent out once" do
+      run_in_eventmachine do
+        connection = TestConnection.new
 
-      channel = ChatChannel.new connection, "test_channel"
-      channel.send_confirmation
-      channel.send_confirmation
+        channel = ChatChannel.new connection, "test_channel"
+        channel.send_confirmation
+        channel.send_confirmation
+
+        wait_for_async
+
+        expected = { "identifier" => "test_channel", "type" => "confirm_subscription" }
+        assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation"
+
+        assert_equal 1, connection.transmissions.size
+      end
+    end
+  end
 
-      wait_for_async
+  require "action_cable/subscription_adapter/async"
 
-      expected = ActiveSupport::JSON.encode "identifier" => "test_channel", "type" => "confirm_subscription"
-      assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation"
+  class UserCallbackChannel < ActionCable::Channel::Base
+    def subscribed
+      stream_from :channel do
+        Thread.current[:ran_callback] = true
+      end
+    end
+  end
 
-      assert_equal 1, connection.transmissions.size
+  class MultiChatChannel < ActionCable::Channel::Base
+    def subscribed
+      stream_from "main_room"
+      stream_from "test_all_rooms"
     end
   end
 
+  class StreamFromTest < ActionCable::TestCase
+    setup do
+      @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Async)
+      @server.config.allowed_request_origins = %w( http://rubyonrails.com )
+    end
+
+    test "custom encoder" do
+      run_in_eventmachine do
+        connection = open_connection
+        subscribe_to connection, identifiers: { id: 1 }
+
+        connection.websocket.expects(:transmit)
+        @server.broadcast "test_room_1", { foo: "bar" }, coder: DummyEncoder
+        wait_for_async
+        wait_for_executor connection.server.worker_pool.executor
+      end
+    end
+
+    test "user supplied callbacks are run through the worker pool" do
+      run_in_eventmachine do
+        connection = open_connection
+        receive(connection, command: "subscribe", channel: UserCallbackChannel.name, identifiers: { id: 1 })
+
+        @server.broadcast "channel", {}
+        wait_for_async
+        refute Thread.current[:ran_callback], "User callback was not run through the worker pool"
+      end
+    end
+
+    test "subscription confirmation should only be sent out once with muptiple stream_from" do
+      run_in_eventmachine do
+        connection = open_connection
+        expected = { "identifier" => { "channel" => MultiChatChannel.name }.to_json, "type" => "confirm_subscription" }
+        connection.websocket.expects(:transmit).with(expected.to_json)
+        receive(connection, command: "subscribe", channel: MultiChatChannel.name, identifiers: {})
+
+        wait_for_async
+      end
+    end
+
+    private
+      def subscribe_to(connection, identifiers:)
+        receive connection, command: "subscribe", identifiers: identifiers
+      end
+
+      def open_connection
+        env = Rack::MockRequest.env_for "/test", "HTTP_HOST" => "localhost", "HTTP_CONNECTION" => "upgrade", "HTTP_UPGRADE" => "websocket", "HTTP_ORIGIN" => "http://rubyonrails.com"
+
+        Connection.new(@server, env).tap do |connection|
+          connection.process
+          assert connection.websocket.possible?
+
+          wait_for_async
+          assert connection.websocket.alive?
+        end
+      end
+
+      def receive(connection, command:, identifiers:, channel: "ActionCable::StreamTests::ChatChannel")
+        identifier = JSON.generate(channel: channel, **identifiers)
+        connection.dispatch_websocket_message JSON.generate(command: command, identifier: identifier)
+        wait_for_async
+      end
+  end
 end