Class: Shikibu::Notify::PostgresNotifyListener

Inherits:
NotifyListener show all
Defined in:
lib/shikibu/notify/pg_notify.rb

Overview

PostgreSQL LISTEN/NOTIFY listener using pg gem Uses a dedicated connection separate from the Sequel connection pool

Examples:

listener = PostgresNotifyListener.new('postgres://localhost/mydb')
listener.subscribe('workflow_resumable') { |payload| puts payload }
listener.start
# ... later
listener.stop

Constant Summary collapse

DEFAULT_RECONNECT_INTERVAL = 
5
MAX_RECONNECT_INTERVAL = 
60

Instance Method Summary collapse

Constructor Details

#initialize(database_url, reconnect_interval: DEFAULT_RECONNECT_INTERVAL) ⇒ PostgresNotifyListener

Returns a new instance of PostgresNotifyListener.

Parameters:

  • PostgreSQL connection URL

  • (defaults to: DEFAULT_RECONNECT_INTERVAL)

    Seconds between reconnection attempts



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/shikibu/notify/pg_notify.rb', line 25

def initialize(database_url, reconnect_interval: DEFAULT_RECONNECT_INTERVAL)
  super()
  @database_url = database_url
  @reconnect_interval = reconnect_interval
  @connection = nil
  @callbacks = {} # channel => [callbacks]
  @running = false
  @listen_thread = nil
  @mutex = Mutex.new
  @reconnect_count = 0
  @executor = Concurrent::ThreadPoolExecutor.new(
    min_threads: 1,
    max_threads: 4,
    max_queue: 100,
    fallback_policy: :discard
  )
end

Instance Method Details

#connected? ⇒ Boolean

Check if listener is connected to PostgreSQL

Returns:



91
92
93
 
# File 'lib/shikibu/notify/pg_notify.rb', line 91

def connected?
  !@connection.nil? && !@connection.finished?
end

#start ⇒ Object

Start the notification listener Establishes connection and begins listening for notifications



45
46
47
48
49
50
51
52
 
# File 'lib/shikibu/notify/pg_notify.rb', line 45

def start
  return if @running

  @running = true
  establish_connection
  start_listen_thread
  log_info('PostgresNotifyListener started')
end

#stop ⇒ Object

Stop the notification listener Closes connection and stops the listener thread



56
57
58
59
60
61
62
63
64
65
 
# File 'lib/shikibu/notify/pg_notify.rb', line 56

def stop
  @running = false
  @executor.shutdown
  @executor.wait_for_termination(5)
  @listen_thread&.join(5)
  @listen_thread = nil
  close_connection
  @callbacks.clear
  log_info('PostgresNotifyListener stopped')
end

#subscribe(channel) {|payload| ... } ⇒ Object

Subscribe to notifications on a channel

Parameters:

  • PostgreSQL channel name

Yields:

  • (payload) —

    Block called when notification received

Yield Parameters:

  • payload (String) —

    Notification payload (JSON string)



71
72
73
74
75
76
77
78
 
# File 'lib/shikibu/notify/pg_notify.rb', line 71

def subscribe(channel, &callback)
  @mutex.synchronize do
    @callbacks[channel] ||= []
    @callbacks[channel] << callback

    listen_to_channel(channel) if connected?
  end
end

#unsubscribe(channel) ⇒ Object

Unsubscribe from notifications on a channel

Parameters:

  • PostgreSQL channel name



82
83
84
85
86
87
 
# File 'lib/shikibu/notify/pg_notify.rb', line 82

def unsubscribe(channel)
  @mutex.synchronize do
    @callbacks.delete(channel)
    unlisten_from_channel(channel) if connected?
  end
end