path: root/actioncable/test/channel/stream_test.rb

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 ChatChannel < ActionCable::Channel::Base
    def subscribed
      if params[:id]
        @room = Room.new params[:id]
        stream_from "test_room_#{@room.id}", coder: pick_coder(params[:coder])
      end
    end

    def send_confirmation
      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

  module DummyEncoder
    extend self
    def encode(*) '{ "foo": "encoded" }' end
    def decode(*) { foo: "decoded" } end
  end

  class SymbolChannel < ActionCable::Channel::Base
    def subscribed
      stream_from :channel
    end
  end

  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

        wait_for_async

        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
        channel.unsubscribe_from_channel
      end
    end

    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

        wait_for_async

        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) }
        channel.unsubscribe_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:stream_tests:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }

        channel = ChatChannel.new connection, ""
        channel.subscribe_to_channel
        channel.stream_for Room.new(1)
      end
    end

    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

    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

        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

  require "action_cable/subscription_adapter/async"

  class UserCallbackChannel < ActionCable::Channel::Base
    def subscribed
      stream_from :channel do
        Thread.current[:ran_callback] = true
      end
    end
  end

  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