Class: MessageBus::Redis::ReliablePubSub

Inherits:
Object
  • Object
show all
Defined in:
lib/message_bus/backends/redis.rb

Defined Under Namespace

Classes: BackLogOutOfOrder, NoMoreRetries

Constant Summary collapse

UNSUB_MESSAGE = 
"$$UNSUBSCRIBE"
LUA_PUBLISH = 
<<LUA

  local start_payload = ARGV[1]
  local max_backlog_age = ARGV[2]
  local max_backlog_size = tonumber(ARGV[3])
  local max_global_backlog_size = tonumber(ARGV[4])
  local channel = ARGV[5]

  local global_id_key = KEYS[1]
  local backlog_id_key = KEYS[2]
  local backlog_key = KEYS[3]
  local global_backlog_key = KEYS[4]
  local redis_channel_name = KEYS[5]

  local global_id = redis.call("INCR", global_id_key)
  local backlog_id = redis.call("INCR", backlog_id_key)
  local payload = string.format("%i|%i|%s", global_id, backlog_id, start_payload)
  local global_backlog_message = string.format("%i|%s", backlog_id, channel)

  redis.call("ZADD", backlog_key, backlog_id, payload)
  redis.call("EXPIRE", backlog_key, max_backlog_age)

  redis.call("ZADD", global_backlog_key, global_id, global_backlog_message)
  redis.call("EXPIRE", global_backlog_key, max_backlog_age)

  redis.call("PUBLISH", redis_channel_name, payload)

  if backlog_id > max_backlog_size then
    redis.call("ZREMRANGEBYSCORE", backlog_key, 1, backlog_id - max_backlog_size)
  end

  if global_id > max_global_backlog_size then
    redis.call("ZREMRANGEBYSCORE", global_backlog_key, 1, global_id - max_global_backlog_size)
  end

  return backlog_id

LUA
LUA_PUBLISH_SHA1 = 
Digest::SHA1.hexdigest(LUA_PUBLISH)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis_config = {}, max_backlog_size = 1000) ⇒ ReliablePubSub

max_backlog_size is per multiplexed channel



31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/message_bus/backends/redis.rb', line 31

def initialize(redis_config = {}, max_backlog_size = 1000)
  @redis_config = redis_config.dup
  unless @redis_config[:enable_redis_logger]
    @redis_config[:logger] = nil
  end
  @max_backlog_size = max_backlog_size
  @max_global_backlog_size = 2000
  @max_in_memory_publish_backlog = 1000
  @in_memory_backlog = []
  @lock = Mutex.new
  @flush_backlog_thread = nil
  # after 7 days inactive backlogs will be removed
  @max_backlog_age = 604800
end

Instance Attribute Details

#max_backlog_age ⇒ Object

Returns the value of attribute max_backlog_age.



17
18
19
 
# File 'lib/message_bus/backends/redis.rb', line 17

def max_backlog_age
  @max_backlog_age
end

#max_backlog_size ⇒ Object

Returns the value of attribute max_backlog_size.



17
18
19
 
# File 'lib/message_bus/backends/redis.rb', line 17

def max_backlog_size
  @max_backlog_size
end

#max_global_backlog_size ⇒ Object

Returns the value of attribute max_global_backlog_size.



17
18
19
 
# File 'lib/message_bus/backends/redis.rb', line 17

def max_global_backlog_size
  @max_global_backlog_size
end

#max_in_memory_publish_backlog ⇒ Object

Returns the value of attribute max_in_memory_publish_backlog.



17
18
19
 
# File 'lib/message_bus/backends/redis.rb', line 17

def max_in_memory_publish_backlog
  @max_in_memory_publish_backlog
end

#subscribed ⇒ Object (readonly)

Returns the value of attribute subscribed.



16
17
18
 
# File 'lib/message_bus/backends/redis.rb', line 16

def subscribed
  @subscribed
end

Instance Method Details

#after_fork ⇒ Object 



50
51
52
 
# File 'lib/message_bus/backends/redis.rb', line 50

def after_fork
  pub_redis.disconnect!
end

#backlog(channel, last_id = nil) ⇒ Object 



224
225
226
227
228
229
230
231
232
 
# File 'lib/message_bus/backends/redis.rb', line 224

def backlog(channel, last_id = nil)
  redis = pub_redis
  backlog_key = backlog_key(channel)
  items = redis.zrangebyscore backlog_key, last_id.to_i + 1, "+inf"

  items.map do |i|
    MessageBus::Message.decode(i)
  end
end

#backlog_id_key(channel) ⇒ Object 



68
69
70
 
# File 'lib/message_bus/backends/redis.rb', line 68

def backlog_id_key(channel)
  "__mb_backlog_id_n_#{channel}"
end

#backlog_key(channel) ⇒ Object 



64
65
66
 
# File 'lib/message_bus/backends/redis.rb', line 64

def backlog_key(channel)
  "__mb_backlog_n_#{channel}"
end

#ensure_backlog_flushed ⇒ Object 



180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
 
# File 'lib/message_bus/backends/redis.rb', line 180

def ensure_backlog_flushed
  flushed = false

  while !flushed
    try_again = false

    if is_readonly?
      sleep 1
      next
    end

    @lock.synchronize do
      if @in_memory_backlog.length == 0
        flushed = true
        break
      end

      begin
        # TODO recover special options
        publish(*@in_memory_backlog[0], queue_in_memory: false)
      rescue Redis::CommandError => e
        if e.message =~ /^READONLY/
          try_again = true
        else
          MessageBus.logger.warn("Dropping undeliverable message: #{e.message}\n#{e.backtrace.join('\n')}")
        end
      rescue => e
        MessageBus.logger.warn("Dropping undeliverable message: #{e.message}\n#{e.backtrace.join('\n')}")
      end

      @in_memory_backlog.delete_at(0) unless try_again
    end
  end
ensure
  @lock.synchronize do
    @flush_backlog_thread = nil
  end
end

#get_message(channel, message_id) ⇒ Object 



252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/message_bus/backends/redis.rb', line 252

def get_message(channel, message_id)
  redis = pub_redis
  backlog_key = backlog_key(channel)

  items = redis.zrangebyscore backlog_key, message_id, message_id
  if items && items[0]
    MessageBus::Message.decode(items[0])
  else
    nil
  end
end

#global_backlog(last_id = nil) ⇒ Object 



234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/message_bus/backends/redis.rb', line 234

def global_backlog(last_id = nil)
  last_id = last_id.to_i
  redis = pub_redis

  items = redis.zrangebyscore global_backlog_key, last_id.to_i + 1, "+inf"

  items.map! do |i|
    pipe = i.index "|"
    message_id = i[0..pipe].to_i
    channel = i[pipe + 1..-1]
    m = get_message(channel, message_id)
    m
  end

  items.compact!
  items
end

#global_backlog_key ⇒ Object 



76
77
78
 
# File 'lib/message_bus/backends/redis.rb', line 76

def global_backlog_key
  "__mb_global_backlog_n"
end

#global_id_key ⇒ Object 



72
73
74
 
# File 'lib/message_bus/backends/redis.rb', line 72

def global_id_key
  "__mb_global_id_n"
end

#global_subscribe(last_id = nil, &blk) ⇒ Object

Raises:

  • (ArgumentError) 


313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
 
# File 'lib/message_bus/backends/redis.rb', line 313

def global_subscribe(last_id = nil, &blk)
  raise ArgumentError unless block_given?
  highest_id = last_id

  clear_backlog = lambda do
    retries = 4
    begin
      highest_id = process_global_backlog(highest_id, retries > 0, &blk)
    rescue BackLogOutOfOrder => e
      highest_id = e.highest_id
      retries -= 1
      sleep(rand(50) / 1000.0)
      retry
    end
  end

  begin
    @redis_global = new_redis_connection

    if highest_id
      clear_backlog.call(&blk)
    end

    @redis_global.subscribe(redis_channel_name) do |on|
      on.subscribe do
        if highest_id
          clear_backlog.call(&blk)
        end
        @subscribed = true
      end

      on.unsubscribe do
        @subscribed = false
      end

      on.message do |c, m|
        if m == UNSUB_MESSAGE
          @redis_global.unsubscribe
          return
        end
        m = MessageBus::Message.decode m

        # we have 3 options
        #
        # 1. message came in the correct order GREAT, just deal with it
        # 2. message came in the incorrect order COMPLICATED, wait a tiny bit and clear backlog
        # 3. message came in the incorrect order and is lowest than current highest id, reset

        if highest_id.nil? || m.global_id == highest_id + 1
          highest_id = m.global_id
          yield m
        else
          clear_backlog.call(&blk)
        end
      end
    end
  rescue => error
    MessageBus.logger.warn "#{error} subscribe failed, reconnecting in 1 second. Call stack #{error.backtrace}"
    sleep 1
    retry
  end
end

#global_unsubscribe ⇒ Object 



304
305
306
307
308
309
310
311
 
# File 'lib/message_bus/backends/redis.rb', line 304

def global_unsubscribe
  if @redis_global
    # new connection to avoid deadlock
    new_redis_connection.publish(redis_channel_name, UNSUB_MESSAGE)
    @redis_global.disconnect
    @redis_global = nil
  end
end

#last_id(channel) ⇒ Object 



219
220
221
222
 
# File 'lib/message_bus/backends/redis.rb', line 219

def last_id(channel)
  backlog_id_key = backlog_id_key(channel)
  pub_redis.get(backlog_id_key).to_i
end

#new_redis_connection ⇒ Object 



46
47
48
 
# File 'lib/message_bus/backends/redis.rb', line 46

def new_redis_connection
  ::Redis.new(@redis_config)
end

#process_global_backlog(highest_id, raise_error, &blk) ⇒ Object 



283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
 
# File 'lib/message_bus/backends/redis.rb', line 283

def process_global_backlog(highest_id, raise_error, &blk)
  if highest_id > pub_redis.get(global_id_key).to_i
    highest_id = 0
  end

  global_backlog(highest_id).each do |old|
    if highest_id + 1 == old.global_id
      yield old
      highest_id = old.global_id
    else
      raise BackLogOutOfOrder.new(highest_id) if raise_error
      if old.global_id > highest_id
        yield old
        highest_id = old.global_id
      end
    end
  end

  highest_id
end

#pub_redis ⇒ Object

redis connection used for publishing messages



60
61
62
 
# File 'lib/message_bus/backends/redis.rb', line 60

def pub_redis
  @pub_redis ||= new_redis_connection
end

#publish(channel, data, opts = nil) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/message_bus/backends/redis.rb', line 128

def publish(channel, data, opts = nil)
  queue_in_memory = (opts && opts[:queue_in_memory]) != false

  max_backlog_age = (opts && opts[:max_backlog_age]) || self.max_backlog_age
  max_backlog_size = (opts && opts[:max_backlog_size]) || self.max_backlog_size

  redis = pub_redis
  backlog_id_key = backlog_id_key(channel)
  backlog_key = backlog_key(channel)

  msg = MessageBus::Message.new nil, nil, channel, data

  cached_eval(redis, LUA_PUBLISH, LUA_PUBLISH_SHA1,
             argv: [
               msg.encode_without_ids,
               max_backlog_age,
               max_backlog_size,
               max_global_backlog_size,
               channel
             ],
             keys: [
               global_id_key,
               backlog_id_key,
               backlog_key,
               global_backlog_key,
               redis_channel_name
             ]
  )

rescue Redis::CommandError => e
  if queue_in_memory && e.message =~ /READONLY/
    @lock.synchronize do
      @in_memory_backlog << [channel, data]
      if @in_memory_backlog.length > @max_in_memory_publish_backlog
        @in_memory_backlog.delete_at(0)
        MessageBus.logger.warn("Dropping old message cause max_in_memory_publish_backlog is full: #{e.message}\n#{e.backtrace.join('\n')}")
      end
    end

    if @flush_backlog_thread == nil
      @lock.synchronize do
        if @flush_backlog_thread == nil
          @flush_backlog_thread = Thread.new { ensure_backlog_flushed }
        end
      end
    end
    nil
  else
    raise
  end
end

#redis_channel_name ⇒ Object 



54
55
56
57
 
# File 'lib/message_bus/backends/redis.rb', line 54

def redis_channel_name
  db = @redis_config[:db] || 0
  "_message_bus_#{db}"
end

#reset! ⇒ Object

use with extreme care, will nuke all of the data



81
82
83
84
85
 
# File 'lib/message_bus/backends/redis.rb', line 81

def reset!
  pub_redis.keys("__mb_*").each do |k|
    pub_redis.del k
  end
end

#subscribe(channel, last_id = nil) ⇒ Object

Raises:

  • (ArgumentError) 


264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/message_bus/backends/redis.rb', line 264

def subscribe(channel, last_id = nil)
  # trivial implementation for now,
  #   can cut down on connections if we only have one global subscriber
  raise ArgumentError unless block_given?

  if last_id
    # we need to translate this to a global id, at least give it a shot
    #   we are subscribing on global and global is always going to be bigger than local
    #   so worst case is a replay of a few messages
    message = get_message(channel, last_id)
    if message
      last_id = message.global_id
    end
  end
  global_subscribe(last_id) do |m|
    yield m if m.channel == channel
  end
end