Class: RightAMQP::HABrokerClient::Published

Inherits:
Object
  • Object
show all
Defined in:
lib/right_amqp/ha_client/ha_broker_client.rb

Overview

Cache for context of recently published messages for use with message returns Applies LRU for managing cache size but only deletes entries when old enough

Constant Summary collapse

MAX_AGE =

Number of seconds since a cache entry was last used before it is deleted

60

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Published

Initialize cache



1154
1155
1156
1157
 
# File 'lib/right_amqp/ha_client/ha_broker_client.rb', line 1154

def initialize
  @cache = {}
  @lru = []
end

Instance Method Details

#fetch(message) ⇒ Object

Fetch context of previously published message

Parameters

message(String)

Serialized message that was published

Return

(Context|nil)

Context of message, or nil if not found in cache



1188
1189
1190
1191
1192
1193
1194
1195
 
# File 'lib/right_amqp/ha_client/ha_broker_client.rb', line 1188

def fetch(message)
  key = identify(message)
  if entry = @cache[key]
    entry[0] = Time.now.to_i
    @lru.push(@lru.delete(key))
    entry[1]
  end
end

#identify(message) ⇒ Object

Obtain a unique identifier for this message

Parameters

message(String)

Serialized message that was published

Returns

(String)

Unique id for message



1204
1205
1206
 
# File 'lib/right_amqp/ha_client/ha_broker_client.rb', line 1204

def identify(message)
  Digest::MD5.hexdigest(message)
end

#store(message, context) ⇒ Object

Store message context in cache

Parameters

message(String)

Serialized message that was published

context(Context)

Message publishing context

Return

true

Always return true



1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
 
# File 'lib/right_amqp/ha_client/ha_broker_client.rb', line 1167

def store(message, context)
  key = identify(message)
  now = Time.now.to_i
  if entry = @cache[key]
    entry[0] = now
    @lru.push(@lru.delete(key))
  else
    @cache[key] = [now, context]
    @lru.push(key)
    @cache.delete(@lru.shift) while (now - @cache[@lru.first][0]) > MAX_AGE
  end
  true
end