class ActionCable::SubscriptionAdapter::PostgreSQL

def broadcast(channel, payload)
  with_broadcast_connection do |pg_conn|
    pg_conn.exec("NOTIFY #{pg_conn.escape_identifier(channel_identifier(channel))}, '#{pg_conn.escape_string(payload)}'")
  end
end

def channel_identifier(channel)
  channel.size > 63 ? OpenSSL::Digest::SHA1.hexdigest(channel) : channel
end

def initialize(*)
  super
  @listener = nil
end

def listener
  @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) }
end

def shutdown
  listener.shutdown
end

def subscribe(channel, callback, success_callback = nil)
  listener.add_subscriber(channel_identifier(channel), callback, success_callback)
end

def unsubscribe(channel, callback)
  listener.remove_subscriber(channel_identifier(channel), callback)
end

def verify!(pg_conn)
  unless pg_conn.is_a?(PG::Connection)
    raise "The Active Record database must be PostgreSQL in order to use the PostgreSQL Action Cable storage adapter"
  end
end

def with_broadcast_connection(&block) # :nodoc:
  ActiveRecord::Base.connection_pool.with_connection do |ar_conn|
    pg_conn = ar_conn.raw_connection
    verify!(pg_conn)
    yield pg_conn
  end
end

def with_subscriptions_connection(&block) # :nodoc:
  ar_conn = ActiveRecord::Base.connection_pool.checkout.tap do |conn|
    # Action Cable is taking ownership over this database connection, and
    # will perform the necessary cleanup tasks
    ActiveRecord::Base.connection_pool.remove(conn)
  end
  pg_conn = ar_conn.raw_connection
  verify!(pg_conn)
  pg_conn.exec("SET application_name = #{pg_conn.escape_identifier(identifier)}")
  yield pg_conn
ensure
  ar_conn.disconnect!
end