Class: Sublist

Inherits:
Object
  • Object
show all
Defined in:
lib/nats/server/sublist.rb

Overview

–

Sublist implementation for a publish-subscribe system. This container class holds subscriptions and matches candidate subjects to those subscriptions. Certain wildcards are supported for subscriptions. ‘*’ will match any given token at any level. ‘>’ will match all subsequent tokens. – See included test for example usage:

Defined Under Namespace

Classes: SublistLevel, SublistNode

Constant Summary collapse

PWC =

:nodoc:

'*'.freeze
FWC = 
'>'.freeze
CACHE_SIZE = 
4096
EMPTY_LEVEL = 
SublistLevel.new({})

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Sublist

Returns a new instance of Sublist.



25
26
27
28
29
30
 
# File 'lib/nats/server/sublist.rb', line 25

def initialize(options = {})
  @count = 0
  @results = []
  @root = SublistLevel.new({})
  @cache = {}
end

Instance Attribute Details

#count ⇒ Object (readonly)

Returns the value of attribute count.



18
19
20
 
# File 'lib/nats/server/sublist.rb', line 18

def count
  @count
end

Instance Method Details

#clear_cache ⇒ Object 



44
 
# File 'lib/nats/server/sublist.rb', line 44

def clear_cache; @cache = {} if @cache; end

#disable_cache ⇒ Object

Ruby is a great language to make selective trade offs of space versus time. We do that here with a low tech front end cache. The cache holds results until it is exhausted or if the instance inserts or removes a subscription. The assumption is that the cache is best suited for high speed matching, and that once it is cleared out it will naturally fill with the high speed matches. This can obviously be improved with a smarter LRU structure that does not need to completely go away when a remove happens..

front end caching is on by default, but we can turn it off here if needed



42
 
# File 'lib/nats/server/sublist.rb', line 42

def disable_cache; @cache = nil; end

#enable_cache ⇒ Object 



43
 
# File 'lib/nats/server/sublist.rb', line 43

def enable_cache;  @cache ||= {};  end

#insert(subject, subscriber) ⇒ Object

Insert a subscriber into the sublist for the given subject.



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/nats/server/sublist.rb', line 54

def insert(subject, subscriber)
  # TODO - validate subject as correct.
  level, tokens = @root, subject.split('.')
  for token in tokens
    # This is slightly slower than direct if statements, but looks cleaner.
    case token
      when FWC then node = (level.fwc || (level.fwc = SublistNode.new([])))
      when PWC then node = (level.pwc || (level.pwc = SublistNode.new([])))
      else node  = ((level.nodes[token]) || (level.nodes[token] = SublistNode.new([])))
    end
    level = (node.next_level || (node.next_level = SublistLevel.new({})))
  end
  node.leaf_nodes.push(subscriber)
  @count += 1
  clear_cache # Clear the cache
  node.next_level = nil if node.next_level == EMPTY_LEVEL
end

#match(subject) ⇒ Object

Match a subject to all subscribers, return the array of matches.



79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/nats/server/sublist.rb', line 79

def match(subject)
  return @cache[subject] if (@cache && @cache[subject])
  tokens = subject.split('.')
  @results.clear
  matchAll(@root, tokens)
  # FIXME: This is too low tech, will revisit when needed.
  if @cache
    prune_cache if @cache.size > CACHE_SIZE
    @cache[subject] = Array.new(@results).freeze # Avoid tampering of copy
  end
  @results
end

#prune_cache ⇒ Object

Random removal



47
48
49
50
51
 
# File 'lib/nats/server/sublist.rb', line 47

def prune_cache
  return unless @cache
  keys = @cache.keys
  @cache.delete(keys[rand(keys.size)])
end

#remove(subject, subscriber) ⇒ Object

Remove a given subscriber from the sublist for the given subject.



73
74
75
76
 
# File 'lib/nats/server/sublist.rb', line 73

def remove(subject, subscriber)
  return unless subject && subscriber
  remove_level(@root, subject.split('.'), subscriber)
end