提交 4769fc4b 编写于 作者: S Sean Griffin

Merge pull request #24540 from sgrif/sg-actioncable-callbacks

Run Action Cable callbacks through the worker pool
......@@ -76,8 +76,11 @@ def stream_from(broadcasting, callback = nil, coder: nil, &block)
# Don't send the confirmation until pubsub#subscribe is successful
defer_subscription_confirmation!
if handler = callback || block
handler = -> message { handler.(coder.decode(message)) } if coder
if user_handler = callback || block
user_handler = -> message { handler.(coder.decode(message)) } if coder
handler = -> message do
connection.worker_pool.async_invoke(user_handler, :call, message)
end
else
handler = default_stream_handler(broadcasting, coder: coder)
end
......
......@@ -116,11 +116,22 @@ class StreamTest < ActionCable::TestCase
require 'action_cable/subscription_adapter/inline'
class UserCallbackChannel < ActionCable::Channel::Base
def subscribed
stream_from :channel do
Thread.current[:ran_callback] = true
end
end
end
class StreamEncodingTest < ActionCable::TestCase
setup do
@server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Inline)
@server.config.allowed_request_origins = %w( http://rubyonrails.com )
@server.stubs(:channel_classes).returns(ChatChannel.name => ChatChannel)
@server.stubs(:channel_classes).returns(
ChatChannel.name => ChatChannel,
UserCallbackChannel.name => UserCallbackChannel,
)
end
test 'custom encoder' do
......@@ -134,6 +145,17 @@ class StreamEncodingTest < ActionCable::TestCase
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
private
def subscribe_to(connection, identifiers:)
receive connection, command: 'subscribe', identifiers: identifiers
......@@ -151,8 +173,8 @@ def open_connection
end
end
def receive(connection, command:, identifiers:)
identifier = JSON.generate(channel: 'ActionCable::StreamTests::ChatChannel', **identifiers)
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
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册