class ActionCable::SubscriptionAdapter::PostgreSQL::Listener
Public Class Methods
new(adapter, event_loop)
click to toggle source
Calls superclass method
# File lib/action_cable/subscription_adapter/postgresql.rb, line 49 def initialize(adapter, event_loop) super() @adapter = adapter @event_loop = event_loop @queue = Queue.new @thread = Thread.new do Thread.current.abort_on_exception = true listen end end
Public Instance Methods
add_channel(channel, on_success)
click to toggle source
# File lib/action_cable/subscription_adapter/postgresql.rb, line 93 def add_channel(channel, on_success) @queue.push([:listen, channel, on_success]) end
invoke_callback(*)
click to toggle source
Calls superclass method
# File lib/action_cable/subscription_adapter/postgresql.rb, line 101 def invoke_callback(*) @event_loop.post { super } end
listen()
click to toggle source
# File lib/action_cable/subscription_adapter/postgresql.rb, line 62 def listen @adapter.with_connection do |pg_conn| catch :shutdown do loop do until @queue.empty? action, channel, callback = @queue.pop(true) case action when :listen pg_conn.exec("LISTEN #{pg_conn.escape_identifier channel}") @event_loop.post(&callback) if callback when :unlisten pg_conn.exec("UNLISTEN #{pg_conn.escape_identifier channel}") when :shutdown throw :shutdown end end pg_conn.wait_for_notify(1) do |chan, pid, message| broadcast(chan, message) end end end end end
remove_channel(channel)
click to toggle source
# File lib/action_cable/subscription_adapter/postgresql.rb, line 97 def remove_channel(channel) @queue.push([:unlisten, channel]) end
shutdown()
click to toggle source
# File lib/action_cable/subscription_adapter/postgresql.rb, line 88 def shutdown @queue.push([:shutdown]) Thread.pass while @thread.alive? end