123456789_123456789_123456789_123456789_123456789_

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

Class Method Summary

::ActionCable::SubscriptionAdapter::SubscriberMap::Async - Inherited

.new

::ActionCable::SubscriptionAdapter::SubscriberMap - Inherited

.new

Instance Method Summary

::ActionCable::SubscriptionAdapter::SubscriberMap::Async - Inherited

#add_subscriber, #invoke_callback, #remove_subscriber

::ActionCable::SubscriptionAdapter::SubscriberMap - Inherited

#add_channel, #add_subscriber, #broadcast, #invoke_callback, #remove_channel, #remove_subscriber

Constructor Details

.new(adapter, executor) ⇒ Listener

[ GitHub ] 

  


# File 'actioncable/lib/action_cable/subscription_adapter/postgresql.rb', line 82



def initialize(adapter, executor)
  super(executor)

  @adapter = adapter
  @queue = Queue.new

  @thread = Thread.new do
    Thread.current.abort_on_exception = true
    listen
  end
end


  







Instance Method Details



  

    #add_channel(channel, on_success)  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/postgresql.rb', line 125



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


  








  

    #listen  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/postgresql.rb', line 94



def listen
  @adapter.with_subscriptions_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}")
            @executor.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)  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/postgresql.rb', line 129



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


  








  

    #shutdown  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/postgresql.rb', line 120



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