Class: ActionCable::SubscriptionAdapter::PostgreSQL::Listener

Inherits:
SubscriberMap
  • Object
show all
Defined in:
lib/action_cable/subscription_adapter/postgresql.rb

Instance Method Summary collapse

Methods inherited from SubscriberMap

#add_subscriber, #broadcast, #remove_subscriber

Constructor Details

#initialize(adapter, event_loop) ⇒ Listener

Returns a new instance of Listener.



56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/action_cable/subscription_adapter/postgresql.rb', line 56

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

Instance Method Details

#add_channel(channel, on_success) ⇒ Object 



100
101
102
 
# File 'lib/action_cable/subscription_adapter/postgresql.rb', line 100

def add_channel(channel, on_success)
  @queue.push([:listen, channel, on_success])
end

#invoke_callback ⇒ Object 



108
109
110
 
# File 'lib/action_cable/subscription_adapter/postgresql.rb', line 108

def invoke_callback(*)
  @event_loop.post { super }
end

#listen ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/action_cable/subscription_adapter/postgresql.rb', line 69

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) ⇒ Object 



104
105
106
 
# File 'lib/action_cable/subscription_adapter/postgresql.rb', line 104

def remove_channel(channel)
  @queue.push([:unlisten, channel])
end

#shutdown ⇒ Object 



95
96
97
98
 
# File 'lib/action_cable/subscription_adapter/postgresql.rb', line 95

def shutdown
  @queue.push([:shutdown])
  Thread.pass while @thread.alive?
end