Initial stab at adding Action Cable to rails/master

20 files changed, 1008 insertions, 0 deletions

diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb
new file mode 100644
index 0000000000..580338b44a
--- /dev/null
+++ b/actioncable/test/channel/base_test.rb
@@ -0,0 +1,148 @@
+require 'test_helper'
+require 'stubs/test_connection'
+require 'stubs/room'
+
+class ActionCable::Channel::BaseTest < ActiveSupport::TestCase
+  class ActionCable::Channel::Base
+    def kick
+      @last_action = [ :kick ]
+    end
+
+    def topic
+    end
+  end
+
+  class BasicChannel < ActionCable::Channel::Base
+    def chatters
+      @last_action = [ :chatters ]
+    end
+  end
+
+  class ChatChannel < BasicChannel
+    attr_reader :room, :last_action
+    after_subscribe :toggle_subscribed
+    after_unsubscribe :toggle_subscribed
+
+    def initialize(*)
+      @subscribed = false
+      super
+    end
+
+    def subscribed
+      @room = Room.new params[:id]
+      @actions = []
+    end
+
+    def unsubscribed
+      @room = nil
+    end
+
+    def toggle_subscribed
+      @subscribed = !@subscribed
+    end
+
+    def leave
+      @last_action = [ :leave ]
+    end
+
+    def speak(data)
+      @last_action = [ :speak, data ]
+    end
+
+    def topic(data)
+      @last_action = [ :topic, data ]
+    end
+
+    def subscribed?
+      @subscribed
+    end
+
+    def get_latest
+      transmit data: 'latest'
+    end
+
+    private
+      def rm_rf
+        @last_action = [ :rm_rf ]
+      end
+  end
+
+  setup do
+    @user = User.new "lifo"
+    @connection = TestConnection.new(@user)
+    @channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
+  end
+
+  test "should subscribe to a channel on initialize" do
+    assert_equal 1, @channel.room.id
+  end
+
+  test "on subscribe callbacks" do
+    assert @channel.subscribed
+  end
+
+  test "channel params" do
+    assert_equal({ id: 1 }, @channel.params)
+  end
+
+  test "unsubscribing from a channel" do
+    assert @channel.room
+    assert @channel.subscribed?
+
+    @channel.unsubscribe_from_channel
+
+    assert ! @channel.room
+    assert ! @channel.subscribed?
+  end
+
+  test "connection identifiers" do
+    assert_equal @user.name, @channel.current_user.name
+  end
+
+  test "callable action without any argument" do
+    @channel.perform_action 'action' => :leave
+    assert_equal [ :leave ], @channel.last_action
+  end
+
+  test "callable action with arguments" do
+    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
+    assert_nil @channel.last_action
+  end
+
+  test "should not dispatch a public method defined on Base" do
+    @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!" }
+
+    @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
+    assert_equal [ :chatters ], @channel.last_action
+  end
+
+  test "transmitting data" do
+    @channel.perform_action 'action' => :get_latest
+
+    expected = ActiveSupport::JSON.encode "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"
+    assert_equal expected, @connection.last_transmission
+  end
+
+end
diff --git a/actioncable/test/channel/broadcasting_test.rb b/actioncable/test/channel/broadcasting_test.rb
new file mode 100644
index 0000000000..1de04243e5
--- /dev/null
+++ b/actioncable/test/channel/broadcasting_test.rb
@@ -0,0 +1,29 @@
+require 'test_helper'
+require 'stubs/test_connection'
+require 'stubs/room'
+
+class ActionCable::Channel::BroadcastingTest < ActiveSupport::TestCase
+  class ChatChannel < ActionCable::Channel::Base
+  end
+
+  setup do
+    @connection = TestConnection.new
+  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") }
+    ChatChannel.broadcast_to(Room.new(1), "Hello World")
+  end
+
+  test "broadcasting_for with an object" do
+    assert_equal "Room#1-Campfire", ChatChannel.broadcasting_for(Room.new(1))
+  end
+
+  test "broadcasting_for with an array" do
+    assert_equal "Room#1-Campfire:Room#2-Campfire", ChatChannel.broadcasting_for([ Room.new(1), Room.new(2) ])
+  end
+
+  test "broadcasting_for with a string" do
+    assert_equal "hello", ChatChannel.broadcasting_for("hello")
+  end
+end
diff --git a/actioncable/test/channel/naming_test.rb b/actioncable/test/channel/naming_test.rb
new file mode 100644
index 0000000000..89ef6ad8b0
--- /dev/null
+++ b/actioncable/test/channel/naming_test.rb
@@ -0,0 +1,10 @@
+require 'test_helper'
+
+class ActionCable::Channel::NamingTest < ActiveSupport::TestCase
+  class ChatChannel < ActionCable::Channel::Base
+  end
+
+  test "channel_name" do
+    assert_equal "action_cable:channel:naming_test:chat", ChatChannel.channel_name
+  end
+end
diff --git a/actioncable/test/channel/periodic_timers_test.rb b/actioncable/test/channel/periodic_timers_test.rb
new file mode 100644
index 0000000000..1590a12f09
--- /dev/null
+++ b/actioncable/test/channel/periodic_timers_test.rb
@@ -0,0 +1,40 @@
+require 'test_helper'
+require 'stubs/test_connection'
+require 'stubs/room'
+
+class ActionCable::Channel::PeriodicTimersTest < ActiveSupport::TestCase
+  class ChatChannel < ActionCable::Channel::Base
+    periodically -> { ping }, every: 5
+    periodically :send_updates, every: 1
+
+    private
+      def ping
+      end
+  end
+
+  setup do
+    @connection = TestConnection.new
+  end
+
+  test "periodic timers definition" do
+    timers = ChatChannel.periodic_timers
+
+    assert_equal 2, timers.size
+
+    first_timer = timers[0]
+    assert_kind_of Proc, first_timer[0]
+    assert_equal 5, first_timer[1][:every]
+
+    second_timer = timers[1]
+    assert_equal :send_updates, second_timer[0]
+    assert_equal 1, second_timer[1][:every]
+  end
+
+  test "timer start and stop" do
+    EventMachine::PeriodicTimer.expects(:new).times(2).returns(true)
+    channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
+
+    channel.expects(:stop_periodic_timers).once
+    channel.unsubscribe_from_channel
+  end
+end
diff --git a/actioncable/test/channel/rejection_test.rb b/actioncable/test/channel/rejection_test.rb
new file mode 100644
index 0000000000..aa93396d44
--- /dev/null
+++ b/actioncable/test/channel/rejection_test.rb
@@ -0,0 +1,25 @@
+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
+  end
+
+  setup do
+    @user = User.new "lifo"
+    @connection = TestConnection.new(@user)
+  end
+
+  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 }
+
+    expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "reject_subscription"
+    assert_equal expected, @connection.last_transmission
+  end
+
+end
diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb
new file mode 100644
index 0000000000..5e4e01abbf
--- /dev/null
+++ b/actioncable/test/channel/stream_test.rb
@@ -0,0 +1,80 @@
+require 'test_helper'
+require 'stubs/test_connection'
+require 'stubs/room'
+
+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}"
+      end
+    end
+
+    def send_confirmation
+      transmit_subscription_confirmation
+    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").returns stub_everything(:pubsub) }
+      channel = ChatChannel.new connection, "{id: 1}", { id: 1 }
+
+      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe_proc) }
+      channel.unsubscribe_from_channel
+    end
+  end
+
+  test "stream_for" do
+    run_in_eventmachine do
+      connection = TestConnection.new
+      EM.next_tick do
+        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire").returns stub_everything(:pubsub) }
+      end
+
+      channel = ChatChannel.new connection, ""
+      channel.stream_for Room.new(1)
+    end
+  end
+
+  test "stream_from subscription confirmation" do
+    EM.run do
+      connection = TestConnection.new
+      connection.expects(:pubsub).returns EM::Hiredis.connect.pubsub
+
+      channel = ChatChannel.new connection, "{id: 1}", { id: 1 }
+      assert_nil connection.last_transmission
+
+      EM::Timer.new(0.1) do
+        expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
+        assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation within 0.1s"
+
+        EM.run_deferred_callbacks
+        EM.stop
+      end
+    end
+  end
+
+  test "subscription confirmation should only be sent out once" do
+    EM.run do
+      connection = TestConnection.new
+      connection.stubs(:pubsub).returns EM::Hiredis.connect.pubsub
+
+      channel = ChatChannel.new connection, "test_channel"
+      channel.send_confirmation
+      channel.send_confirmation
+
+      EM.run_deferred_callbacks
+
+      expected = ActiveSupport::JSON.encode "identifier" => "test_channel", "type" => "confirm_subscription"
+      assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation"
+
+      assert_equal 1, connection.transmissions.size
+      EM.stop
+    end
+  end
+
+end
diff --git a/actioncable/test/connection/authorization_test.rb b/actioncable/test/connection/authorization_test.rb
new file mode 100644
index 0000000000..68668b2835
--- /dev/null
+++ b/actioncable/test/connection/authorization_test.rb
@@ -0,0 +1,32 @@
+require 'test_helper'
+require 'stubs/test_server'
+
+class ActionCable::Connection::AuthorizationTest < ActionCable::TestCase
+  class Connection < ActionCable::Connection::Base
+    attr_reader :websocket
+
+    def connect
+      reject_unauthorized_connection
+    end
+
+    def send_async(method, *args)
+      # Bypass Celluloid
+      send method, *args
+    end
+  end
+
+  test "unauthorized connection" do
+    run_in_eventmachine do
+      server = TestServer.new
+      server.config.allowed_request_origins = %w( http://rubyonrails.com )
+
+      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket',
+        'HTTP_ORIGIN' => 'http://rubyonrails.com'
+
+      connection = Connection.new(server, env)
+      connection.websocket.expects(:close)
+
+      connection.process
+    end
+  end
+end
diff --git a/actioncable/test/connection/base_test.rb b/actioncable/test/connection/base_test.rb
new file mode 100644
index 0000000000..da6041db4a
--- /dev/null
+++ b/actioncable/test/connection/base_test.rb
@@ -0,0 +1,118 @@
+require 'test_helper'
+require 'stubs/test_server'
+
+class ActionCable::Connection::BaseTest < ActionCable::TestCase
+  class Connection < ActionCable::Connection::Base
+    attr_reader :websocket, :subscriptions, :message_buffer, :connected
+
+    def connect
+      @connected = true
+    end
+
+    def disconnect
+      @connected = false
+    end
+
+    def send_async(method, *args)
+      # Bypass Celluloid
+      send method, *args
+    end
+  end
+
+  setup do
+    @server = TestServer.new
+    @server.config.allowed_request_origins = %w( http://rubyonrails.com )
+  end
+
+  test "making a connection with invalid headers" do
+    run_in_eventmachine do
+      connection = ActionCable::Connection::Base.new(@server, Rack::MockRequest.env_for("/test"))
+      response = connection.process
+      assert_equal 404, response[0]
+    end
+  end
+
+  test "websocket connection" do
+    run_in_eventmachine do
+      connection = open_connection
+      connection.process
+
+      assert connection.websocket.possible?
+      assert connection.websocket.alive?
+    end
+  end
+
+  test "rack response" do
+    run_in_eventmachine do
+      connection = open_connection
+      response = connection.process
+
+      assert_equal [ -1, {}, [] ], response
+    end
+  end
+
+  test "on connection open" do
+    run_in_eventmachine do
+      connection = open_connection
+      connection.process
+
+      connection.websocket.expects(:transmit).with(regexp_matches(/\_ping/))
+      connection.message_buffer.expects(:process!)
+
+      # Allow EM to run on_open callback
+      EM.next_tick do
+        assert_equal [ connection ], @server.connections
+        assert connection.connected
+      end
+    end
+  end
+
+  test "on connection close" do
+    run_in_eventmachine do
+      connection = open_connection
+      connection.process
+
+      # Setup the connection
+      EventMachine.stubs(:add_periodic_timer).returns(true)
+      connection.send :on_open
+      assert connection.connected
+
+      connection.subscriptions.expects(:unsubscribe_from_all)
+      connection.send :on_close
+
+      assert ! connection.connected
+      assert_equal [], @server.connections
+    end
+  end
+
+  test "connection statistics" do
+    run_in_eventmachine do
+      connection = open_connection
+      connection.process
+
+      statistics = connection.statistics
+
+      assert statistics[:identifier].blank?
+      assert_kind_of Time, statistics[:started_at]
+      assert_equal [], statistics[:subscriptions]
+    end
+  end
+
+  test "explicitly closing a connection" do
+    run_in_eventmachine do
+      connection = open_connection
+      connection.process
+
+      connection.websocket.expects(:close)
+      connection.close
+    end
+  end
+
+  private
+    def open_connection
+      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket',
+        'HTTP_ORIGIN' => 'http://rubyonrails.com'
+
+      Connection.new(@server, env)
+    end
+end
diff --git a/actioncable/test/connection/cross_site_forgery_test.rb b/actioncable/test/connection/cross_site_forgery_test.rb
new file mode 100644
index 0000000000..ede3057e30
--- /dev/null
+++ b/actioncable/test/connection/cross_site_forgery_test.rb
@@ -0,0 +1,82 @@
+require 'test_helper'
+require 'stubs/test_server'
+
+class ActionCable::Connection::CrossSiteForgeryTest < ActionCable::TestCase
+  HOST = 'rubyonrails.com'
+
+  class Connection < ActionCable::Connection::Base
+    def send_async(method, *args)
+      # Bypass Celluloid
+      send method, *args
+    end
+  end
+
+  setup do
+    @server = TestServer.new
+    @server.config.allowed_request_origins = %w( http://rubyonrails.com )
+  end
+
+  teardown do
+    @server.config.disable_request_forgery_protection = false
+    @server.config.allowed_request_origins = []
+  end
+
+  test "disable forgery protection" do
+    @server.config.disable_request_forgery_protection = true
+    assert_origin_allowed 'http://rubyonrails.com'
+    assert_origin_allowed 'http://hax.com'
+  end
+
+  test "explicitly specified a single allowed origin" do
+    @server.config.allowed_request_origins = 'http://hax.com'
+    assert_origin_not_allowed 'http://rubyonrails.com'
+    assert_origin_allowed 'http://hax.com'
+  end
+
+  test "explicitly specified multiple allowed origins" do
+    @server.config.allowed_request_origins = %w( http://rubyonrails.com http://www.rubyonrails.com )
+    assert_origin_allowed 'http://rubyonrails.com'
+    assert_origin_allowed 'http://www.rubyonrails.com'
+    assert_origin_not_allowed 'http://hax.com'
+  end
+
+  test "explicitly specified a single regexp allowed origin" do
+    @server.config.allowed_request_origins = /.*ha.*/
+    assert_origin_not_allowed 'http://rubyonrails.com'
+    assert_origin_allowed 'http://hax.com'
+  end
+
+  test "explicitly specified multiple regexp allowed origins" do
+    @server.config.allowed_request_origins = [/http:\/\/ruby.*/, /.*rai.s.*com/, 'string' ]
+    assert_origin_allowed 'http://rubyonrails.com'
+    assert_origin_allowed 'http://www.rubyonrails.com'
+    assert_origin_not_allowed 'http://hax.com'
+    assert_origin_not_allowed 'http://rails.co.uk'
+  end
+
+  private
+    def assert_origin_allowed(origin)
+      response = connect_with_origin origin
+      assert_equal -1, response[0]
+    end
+
+    def assert_origin_not_allowed(origin)
+      response = connect_with_origin origin
+      assert_equal 404, response[0]
+    end
+
+    def connect_with_origin(origin)
+      response = nil
+
+      run_in_eventmachine do
+        response = Connection.new(@server, env_for_origin(origin)).process
+      end
+
+      response
+    end
+
+    def env_for_origin(origin)
+      Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket', 'SERVER_NAME' => HOST,
+        'HTTP_ORIGIN' => origin
+    end
+end
diff --git a/actioncable/test/connection/identifier_test.rb b/actioncable/test/connection/identifier_test.rb
new file mode 100644
index 0000000000..02e6b21845
--- /dev/null
+++ b/actioncable/test/connection/identifier_test.rb
@@ -0,0 +1,77 @@
+require 'test_helper'
+require 'stubs/test_server'
+require 'stubs/user'
+
+class ActionCable::Connection::IdentifierTest < ActionCable::TestCase
+  class Connection < ActionCable::Connection::Base
+    identified_by :current_user
+    attr_reader :websocket
+
+    public :process_internal_message
+
+    def connect
+      self.current_user = User.new "lifo"
+    end
+  end
+
+  test "connection identifier" do
+    run_in_eventmachine do
+      open_connection_with_stubbed_pubsub
+      assert_equal "User#lifo", @connection.connection_identifier
+    end
+  end
+
+  test "should subscribe to internal channel on open and unsubscribe on close" do
+    run_in_eventmachine do
+      pubsub = mock('pubsub')
+      pubsub.expects(:subscribe).with('action_cable/User#lifo')
+      pubsub.expects(:unsubscribe_proc).with('action_cable/User#lifo', kind_of(Proc))
+
+      server = TestServer.new
+      server.stubs(:pubsub).returns(pubsub)
+
+      open_connection server: server
+      close_connection
+    end
+  end
+
+  test "processing disconnect message" do
+    run_in_eventmachine do
+      open_connection_with_stubbed_pubsub
+
+      @connection.websocket.expects(:close)
+      message = ActiveSupport::JSON.encode('type' => 'disconnect')
+      @connection.process_internal_message message
+    end
+  end
+
+  test "processing invalid message" do
+    run_in_eventmachine do
+      open_connection_with_stubbed_pubsub
+
+      @connection.websocket.expects(:close).never
+      message = ActiveSupport::JSON.encode('type' => 'unknown')
+      @connection.process_internal_message message
+    end
+  end
+
+  protected
+    def open_connection_with_stubbed_pubsub
+      server = TestServer.new
+      server.stubs(:pubsub).returns(stub_everything('pubsub'))
+
+      open_connection server: server
+    end
+
+    def open_connection(server:)
+      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      @connection = Connection.new(server, env)
+
+      @connection.process
+      @connection.send :on_open
+    end
+
+    def close_connection
+      @connection.send :on_close
+    end
+end
diff --git a/actioncable/test/connection/multiple_identifiers_test.rb b/actioncable/test/connection/multiple_identifiers_test.rb
new file mode 100644
index 0000000000..55a9f96cb3
--- /dev/null
+++ b/actioncable/test/connection/multiple_identifiers_test.rb
@@ -0,0 +1,41 @@
+require 'test_helper'
+require 'stubs/test_server'
+require 'stubs/user'
+
+class ActionCable::Connection::MultipleIdentifiersTest < ActionCable::TestCase
+  class Connection < ActionCable::Connection::Base
+    identified_by :current_user, :current_room
+
+    def connect
+      self.current_user = User.new "lifo"
+      self.current_room = Room.new "my", "room"
+    end
+  end
+
+  test "multiple connection identifiers" do
+    run_in_eventmachine do
+      open_connection_with_stubbed_pubsub
+      assert_equal "Room#my-room:User#lifo", @connection.connection_identifier
+    end
+  end
+
+  protected
+    def open_connection_with_stubbed_pubsub
+      server = TestServer.new
+      server.stubs(:pubsub).returns(stub_everything('pubsub'))
+
+      open_connection server: server
+    end
+
+    def open_connection(server:)
+      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      @connection = Connection.new(server, env)
+
+      @connection.process
+      @connection.send :on_open
+    end
+
+    def close_connection
+      @connection.send :on_close
+    end
+end
diff --git a/actioncable/test/connection/string_identifier_test.rb b/actioncable/test/connection/string_identifier_test.rb
new file mode 100644
index 0000000000..ab69df57b3
--- /dev/null
+++ b/actioncable/test/connection/string_identifier_test.rb
@@ -0,0 +1,44 @@
+require 'test_helper'
+require 'stubs/test_server'
+
+class ActionCable::Connection::StringIdentifierTest < ActionCable::TestCase
+  class Connection < ActionCable::Connection::Base
+    identified_by :current_token
+
+    def connect
+      self.current_token = "random-string"
+    end
+
+    def send_async(method, *args)
+      # Bypass Celluloid
+      send method, *args
+    end
+  end
+
+  test "connection identifier" do
+    run_in_eventmachine do
+      open_connection_with_stubbed_pubsub
+      assert_equal "random-string", @connection.connection_identifier
+    end
+  end
+
+  protected
+    def open_connection_with_stubbed_pubsub
+      @server = TestServer.new
+      @server.stubs(:pubsub).returns(stub_everything('pubsub'))
+
+      open_connection
+    end
+
+    def open_connection
+      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      @connection = Connection.new(@server, env)
+
+      @connection.process
+      @connection.send :on_open
+    end
+
+    def close_connection
+      @connection.send :on_close
+    end
+end
diff --git a/actioncable/test/connection/subscriptions_test.rb b/actioncable/test/connection/subscriptions_test.rb
new file mode 100644
index 0000000000..4f6760827e
--- /dev/null
+++ b/actioncable/test/connection/subscriptions_test.rb
@@ -0,0 +1,116 @@
+require 'test_helper'
+
+class ActionCable::Connection::SubscriptionsTest < ActionCable::TestCase
+  class Connection < ActionCable::Connection::Base
+    attr_reader :websocket
+
+    def send_async(method, *args)
+      # Bypass Celluloid
+      send method, *args
+    end
+  end
+
+  class ChatChannel < ActionCable::Channel::Base
+    attr_reader :room, :lines
+
+    def subscribed
+      @room = Room.new params[:id]
+      @lines = []
+    end
+
+    def speak(data)
+      @lines << data
+    end
+  end
+
+  setup do
+    @server = TestServer.new
+    @server.stubs(:channel_classes).returns(ChatChannel.name => ChatChannel)
+
+    @chat_identifier = ActiveSupport::JSON.encode(id: 1, channel: 'ActionCable::Connection::SubscriptionsTest::ChatChannel')
+  end
+
+  test "subscribe command" do
+    run_in_eventmachine do
+      setup_connection
+      channel = subscribe_to_chat_channel
+
+      assert_kind_of ChatChannel, channel
+      assert_equal 1, channel.room.id
+    end
+  end
+
+  test "subscribe command without an identifier" do
+    run_in_eventmachine do
+      setup_connection
+
+      @subscriptions.execute_command 'command' => 'subscribe'
+      assert @subscriptions.identifiers.empty?
+    end
+  end
+
+  test "unsubscribe command" do
+    run_in_eventmachine do
+      setup_connection
+      subscribe_to_chat_channel
+
+      channel = subscribe_to_chat_channel
+      channel.expects(:unsubscribe_from_channel)
+
+      @subscriptions.execute_command 'command' => 'unsubscribe', 'identifier' => @chat_identifier
+      assert @subscriptions.identifiers.empty?
+    end
+  end
+
+  test "unsubscribe command without an identifier" do
+    run_in_eventmachine do
+      setup_connection
+
+      @subscriptions.execute_command 'command' => 'unsubscribe'
+      assert @subscriptions.identifiers.empty?
+    end
+  end
+
+  test "message command" do
+    run_in_eventmachine do
+      setup_connection
+      channel = subscribe_to_chat_channel
+
+      data = { 'content' => 'Hello World!', 'action' => 'speak' }
+      @subscriptions.execute_command 'command' => 'message', 'identifier' => @chat_identifier, 'data' => ActiveSupport::JSON.encode(data)
+
+      assert_equal [ data ], channel.lines
+    end
+  end
+
+  test "unsubscrib from all" do
+    run_in_eventmachine do
+      setup_connection
+
+      channel1 = subscribe_to_chat_channel
+
+      channel2_id = ActiveSupport::JSON.encode(id: 2, channel: 'ActionCable::Connection::SubscriptionsTest::ChatChannel')
+      channel2 = subscribe_to_chat_channel(channel2_id)
+
+      channel1.expects(:unsubscribe_from_channel)
+      channel2.expects(:unsubscribe_from_channel)
+
+      @subscriptions.unsubscribe_from_all
+    end
+  end
+
+  private
+    def subscribe_to_chat_channel(identifier = @chat_identifier)
+      @subscriptions.execute_command 'command' => 'subscribe', 'identifier' => identifier
+      assert_equal identifier, @subscriptions.identifiers.last
+
+      @subscriptions.send :find, 'identifier' => identifier
+    end
+
+    def setup_connection
+      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      @connection = Connection.new(@server, env)
+
+      @subscriptions = ActionCable::Connection::Subscriptions.new(@connection)
+    end
+end
diff --git a/actioncable/test/stubs/global_id.rb b/actioncable/test/stubs/global_id.rb
new file mode 100644
index 0000000000..334f0d03e8
--- /dev/null
+++ b/actioncable/test/stubs/global_id.rb
@@ -0,0 +1,8 @@
+class GlobalID
+  attr_reader :uri
+  delegate :to_param, :to_s, to: :uri
+
+  def initialize(gid, options = {})
+    @uri = gid
+  end
+end
diff --git a/actioncable/test/stubs/room.rb b/actioncable/test/stubs/room.rb
new file mode 100644
index 0000000000..cd66a0b687
--- /dev/null
+++ b/actioncable/test/stubs/room.rb
@@ -0,0 +1,16 @@
+class Room
+  attr_reader :id, :name
+
+  def initialize(id, name='Campfire')
+    @id = id
+    @name = name
+  end
+
+  def to_global_id
+    GlobalID.new("Room##{id}-#{name}")
+  end
+
+  def to_gid_param
+    to_global_id.to_param
+  end
+end
diff --git a/actioncable/test/stubs/test_connection.rb b/actioncable/test/stubs/test_connection.rb
new file mode 100644
index 0000000000..384abc5e76
--- /dev/null
+++ b/actioncable/test/stubs/test_connection.rb
@@ -0,0 +1,21 @@
+require 'stubs/user'
+
+class TestConnection
+  attr_reader :identifiers, :logger, :current_user, :transmissions
+
+  def initialize(user = User.new("lifo"))
+    @identifiers = [ :current_user ]
+
+    @current_user = user
+    @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
+    @transmissions = []
+  end
+
+  def transmit(data)
+    @transmissions << data
+  end
+
+  def last_transmission
+    @transmissions.last
+  end
+end
diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb
new file mode 100644
index 0000000000..f9168f9b78
--- /dev/null
+++ b/actioncable/test/stubs/test_server.rb
@@ -0,0 +1,15 @@
+require 'ostruct'
+
+class TestServer
+  include ActionCable::Server::Connections
+
+  attr_reader :logger, :config
+
+  def initialize
+    @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
+    @config = OpenStruct.new(log_tags: [])
+  end
+
+  def send_async
+  end
+end
diff --git a/actioncable/test/stubs/user.rb b/actioncable/test/stubs/user.rb
new file mode 100644
index 0000000000..a66b4f87d5
--- /dev/null
+++ b/actioncable/test/stubs/user.rb
@@ -0,0 +1,15 @@
+class User
+  attr_reader :name
+
+  def initialize(name)
+    @name = name
+  end
+
+  def to_global_id
+    GlobalID.new("User##{name}")
+  end
+
+  def to_gid_param
+    to_global_id.to_param
+  end
+end
diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb
new file mode 100644
index 0000000000..88319fd4a7
--- /dev/null
+++ b/actioncable/test/test_helper.rb
@@ -0,0 +1,42 @@
+require File.expand_path('../../../load_paths', __FILE__)
+
+require 'action_cable'
+require 'active_support/testing/autorun'
+
+
+require 'puma'
+require 'em-hiredis'
+
+require 'mocha/setup'
+
+require 'rack/mock'
+
+# Require all the stubs and models
+Dir[File.dirname(__FILE__) + '/stubs/*.rb'].each {|file| require file }
+
+require 'celluloid'
+$CELLULOID_DEBUG = false
+$CELLULOID_TEST = false
+Celluloid.logger = Logger.new(StringIO.new)
+
+require 'faye/websocket'
+class << Faye::WebSocket
+  remove_method :ensure_reactor_running
+
+  # We don't want Faye to start the EM reactor in tests because it makes testing much harder.
+  # We want to be able to start and stop EM loop in tests to make things simpler.
+  def ensure_reactor_running
+    # no-op
+  end
+end
+
+class ActionCable::TestCase < ActiveSupport::TestCase
+  def run_in_eventmachine
+    EM.run do
+      yield
+
+      EM.run_deferred_callbacks
+      EM.stop
+    end
+  end
+end
diff --git a/actioncable/test/worker_test.rb b/actioncable/test/worker_test.rb
new file mode 100644
index 0000000000..69c4b6529d
--- /dev/null
+++ b/actioncable/test/worker_test.rb
@@ -0,0 +1,49 @@
+require 'test_helper'
+
+class WorkerTest < ActiveSupport::TestCase
+  class Receiver
+    attr_accessor :last_action
+
+    def run
+      @last_action = :run
+    end
+
+    def process(message)
+      @last_action =  [ :process, message ]
+    end
+
+    def connection
+    end
+  end
+
+  setup do
+    Celluloid.boot
+
+    @worker = ActionCable::Server::Worker.new
+    @receiver = Receiver.new
+  end
+
+  teardown do
+    @receiver.last_action = nil
+  end
+
+  test "invoke" do
+    @worker.invoke @receiver, :run
+    assert_equal :run, @receiver.last_action
+  end
+
+  test "invoke with arguments" do
+    @worker.invoke @receiver, :process, "Hello"
+    assert_equal [ :process, "Hello" ], @receiver.last_action
+  end
+
+  test "running periodic timers with a proc" do
+    @worker.run_periodic_timer @receiver, @receiver.method(:run)
+    assert_equal :run, @receiver.last_action
+  end
+
+  test "running periodic timers with a method" do
+    @worker.run_periodic_timer @receiver, :run
+    assert_equal :run, @receiver.last_action
+  end
+end