Class: PgNotifier::Manager

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_notifier/manager.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(attrs = {}) ⇒ Manager

Returns a new instance of Manager.



9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/pg_notifier/manager.rb', line 9

def initialize(attrs = {})
  @logger = attrs.fetch :logger, Logger.new(STDOUT)
  @db_config = attrs.fetch :db_config , {}
  @timeout = attrs.fetch :timeout, 1

  @finish = false
  @mutex = Mutex.new
  @resource = ConditionVariable.new

  Thread.abort_on_exception = true
end

Instance Attribute Details

#db_config ⇒ Object

Returns the value of attribute db_config.



7
8
9
 
# File 'lib/pg_notifier/manager.rb', line 7

def db_config
  @db_config
end

#logger ⇒ Object

Returns the value of attribute logger.



7
8
9
 
# File 'lib/pg_notifier/manager.rb', line 7

def logger
  @logger
end

#timeout ⇒ Object

Returns the value of attribute timeout.



7
8
9
 
# File 'lib/pg_notifier/manager.rb', line 7

def timeout
  @timeout
end

Instance Method Details

#channels ⇒ Object 



30
31
32
 
# File 'lib/pg_notifier/manager.rb', line 30

def channels
  subscriptions_by_channels.keys
end

#close_connection ⇒ Object 



100
101
102
103
104
105
106
107
108
 
# File 'lib/pg_notifier/manager.rb', line 100

def close_connection
  unless connection.finished?
    channels.each do |channel|
      connection.exec "UNLISTEN #{channel};"
    end

    connection.finish
  end
end

#connection ⇒ Object 



34
35
36
 
# File 'lib/pg_notifier/manager.rb', line 34

def connection
  @connection ||= PG::Connection.open db_config
end

#graceful_shutdown ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/pg_notifier/manager.rb', line 110

def graceful_shutdown
  logger.info 'Gracefully shutting down'

  @finish = true

  @mutex.synchronize do
    close_connection
    @resource.signal
  end

  exit(0)
end

#handle_signal(sig) ⇒ Object 



78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/pg_notifier/manager.rb', line 78

def handle_signal(sig)
  logger.debug "Got #{sig} signal"

  case sig
  when 'INT'
    shutdown
  when 'TERM'
    graceful_shutdown
  when 'HUP'
    graceful_shutdown
  end
end

#notify(channel, options = {}, &block) ⇒ Object 



21
22
23
24
 
# File 'lib/pg_notifier/manager.rb', line 21

def notify(channel, options = {}, &block)
  subscriptions_by_channels[channel] ||= []
  subscriptions_by_channels[channel] << Subscription.new(channel, options, &block)
end

#run ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/pg_notifier/manager.rb', line 38

def run
  logger.info "Starting pg_notifier for #{channels.count} channels: [ #{channels.join(' ')} ]"

  sig_read, sig_write = IO.pipe

  (%w[INT TERM HUP] & Signal.list.keys).each do |sig|
    trap sig do
      sig_write.puts(sig)
    end
  end

  Thread.new do
    channels.each do |channel|
      pg_result = connection.exec "LISTEN #{channel};"

      unless pg_result.result_status.eql? PG::PGRES_COMMAND_OK
        raise ChannelNotLaunched, "Channel ##{channel} not launched"
      end
    end

    @mutex.synchronize do
      until @finish do
        connection.wait_for_notify do |channel, pid, payload|
          logger.info "Notifying channel: #{channel}, pid: #{pid}, payload: #{payload}"

          subscriptions = subscriptions_by_channels.fetch channel, []
          subscriptions.each { |subscription| subscription.notify(channel, pid, payload) }
        end

        @resource.wait @mutex, timeout
      end
    end
  end

  while io = IO.select([sig_read])
    sig = io.first[0].gets.chomp
    handle_signal(sig)
  end
end

#shutdown ⇒ Object 



91
92
93
94
95
96
97
98
 
# File 'lib/pg_notifier/manager.rb', line 91

def shutdown
  logger.info 'Shutting down'

  @finish = true
  close_connection

  exit(0)
end

#subscriptions_by_channels ⇒ Object 



26
27
28
 
# File 'lib/pg_notifier/manager.rb', line 26

def subscriptions_by_channels
  @subscriptions_by_channels ||= {}
end