Class: MessageBus::Backends::Postgres::Client
- Inherits:
-
Object
- Object
- MessageBus::Backends::Postgres::Client
- Defined in:
- lib/message_bus/backends/postgres.rb
Defined Under Namespace
Classes: Listener
Constant Summary collapse
- INHERITED_CONNECTIONS =
[]
Instance Method Summary collapse
- #add(channel, value) ⇒ Object
- #backlog(channel, backlog_id) ⇒ Object
- #clear_channel_backlog(channel, backlog_id, num_to_keep) ⇒ Object
- #clear_global_backlog(backlog_id, num_to_keep) ⇒ Object
- #expire(max_backlog_age) ⇒ Object
-
#expire_all_backlogs! ⇒ Object
use with extreme care, will nuke all of the data.
- #get_value(channel, id) ⇒ Object
- #global_backlog(backlog_id) ⇒ Object
-
#initialize(config) ⇒ Client
constructor
A new instance of Client.
- #max_id(channel = nil) ⇒ Object
- #publish(channel, data) ⇒ Object
- #reconnect ⇒ Object
-
#reset! ⇒ Object
Dangerous, drops the message_bus table containing the backlog if it exists.
- #subscribe(channel) {|listener| ... } ⇒ Object
- #unsubscribe ⇒ Object
Constructor Details
#initialize(config) ⇒ Client
Returns a new instance of Client.
44 45 46 47 48 49 50 51 |
# File 'lib/message_bus/backends/postgres.rb', line 44 def initialize(config) @config = config @listening_on = {} @available = [] @allocated = {} @mutex = Mutex.new @pid = Process.pid end |
Instance Method Details
#add(channel, value) ⇒ Object
53 54 55 |
# File 'lib/message_bus/backends/postgres.rb', line 53 def add(channel, value) hold { |conn| exec_prepared(conn, 'insert_message', [channel, value]) { |r| r.getvalue(0, 0).to_i } } end |
#backlog(channel, backlog_id) ⇒ Object
74 75 76 77 78 |
# File 'lib/message_bus/backends/postgres.rb', line 74 def backlog(channel, backlog_id) hold do |conn| exec_prepared(conn, 'channel_backlog', [channel, backlog_id]) { |r| r.values.each { |a| a[0] = a[0].to_i } } end || [] end |
#clear_channel_backlog(channel, backlog_id, num_to_keep) ⇒ Object
64 65 66 67 |
# File 'lib/message_bus/backends/postgres.rb', line 64 def clear_channel_backlog(channel, backlog_id, num_to_keep) hold { |conn| exec_prepared(conn, 'clear_channel_backlog', [channel, backlog_id, num_to_keep]) } nil end |
#clear_global_backlog(backlog_id, num_to_keep) ⇒ Object
57 58 59 60 61 62 |
# File 'lib/message_bus/backends/postgres.rb', line 57 def clear_global_backlog(backlog_id, num_to_keep) if backlog_id > num_to_keep hold { |conn| exec_prepared(conn, 'clear_global_backlog', [backlog_id - num_to_keep]) } nil end end |
#expire(max_backlog_age) ⇒ Object
69 70 71 72 |
# File 'lib/message_bus/backends/postgres.rb', line 69 def expire(max_backlog_age) hold { |conn| exec_prepared(conn, 'expire', [max_backlog_age]) } nil end |
#expire_all_backlogs! ⇒ Object
use with extreme care, will nuke all of the data
106 107 108 |
# File 'lib/message_bus/backends/postgres.rb', line 106 def expire_all_backlogs! reset! end |
#get_value(channel, id) ⇒ Object
86 87 88 |
# File 'lib/message_bus/backends/postgres.rb', line 86 def get_value(channel, id) hold { |conn| exec_prepared(conn, 'get_message', [channel, id]) { |r| r.getvalue(0, 0) } } end |
#global_backlog(backlog_id) ⇒ Object
80 81 82 83 84 |
# File 'lib/message_bus/backends/postgres.rb', line 80 def global_backlog(backlog_id) hold do |conn| exec_prepared(conn, 'global_backlog', [backlog_id]) { |r| r.values.each { |a| a[0] = a[0].to_i } } end || [] end |
#max_id(channel = nil) ⇒ Object
110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
# File 'lib/message_bus/backends/postgres.rb', line 110 def max_id(channel = nil) block = proc do |r| if r.ntuples > 0 r.getvalue(0, 0).to_i else 0 end end if channel hold { |conn| exec_prepared(conn, 'max_channel_id', [channel], &block) } else hold { |conn| exec_prepared(conn, 'max_id', &block) } end end |
#publish(channel, data) ⇒ Object
126 127 128 |
# File 'lib/message_bus/backends/postgres.rb', line 126 def publish(channel, data) hold { |conn| exec_prepared(conn, 'publish', [channel, data]) } end |
#reconnect ⇒ Object
90 91 92 93 94 95 |
# File 'lib/message_bus/backends/postgres.rb', line 90 def reconnect sync do @listening_on.clear @available.clear end end |
#reset! ⇒ Object
Dangerous, drops the message_bus table containing the backlog if it exists.
98 99 100 101 102 103 |
# File 'lib/message_bus/backends/postgres.rb', line 98 def reset! hold do |conn| conn.exec 'DROP TABLE IF EXISTS message_bus' create_table(conn) end end |
#subscribe(channel) {|listener| ... } ⇒ Object
130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
# File 'lib/message_bus/backends/postgres.rb', line 130 def subscribe(channel) obj = Object.new sync { @listening_on[channel] = obj } listener = Listener.new yield listener conn = raw_pg_connection conn.exec "LISTEN #{channel}" listener.do_sub.call while listening_on?(channel, obj) conn.wait_for_notify(10) do |_, _, payload| break unless listening_on?(channel, obj) listener..call(nil, payload) end end listener.do_unsub.call conn.exec "UNLISTEN #{channel}" nil end |
#unsubscribe ⇒ Object
152 153 154 |
# File 'lib/message_bus/backends/postgres.rb', line 152 def unsubscribe sync { @listening_on.clear } end |