Class: RubySkynet::Zookeeper::Registry

Inherits:
Object
  • Object
show all
Includes:
SemanticLogger::Loggable
Defined in:
lib/ruby_skynet/zookeeper/registry.rb

Overview

Registry

Store information in Zookeepr and subscribe to future changes

Notifies registered subscribers when information has changed

All paths specified are relative to the root. As such the root key is never returned, nor is it required when a key is supplied as input. For example, with a root of /foo/bar, any paths passed in will leave out the root: host/name

Direct Known Subclasses

CachedRegistry

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(params, &block) ⇒ Registry

Create a Registry instance to manage a information within Zookeeper

:root [String] Root key to load and then monitor for changes It is not recommended to set the root to "/" as it will generate significant traffic since it will also monitor ZooKeeper Admin changes Mandatory

:zookeeper [Hash|ZooKeeper] ZooKeeper configuration information, or an existing ZooKeeper ( ZooKeeper client) instance

:servers [Array of String]
Array of URL's of ZooKeeper servers to connect to with port numbers
['server1:2181', 'server2:2181']

:connect_timeout [Float]
Time in seconds to timeout when trying to connect to the server

Optional Block The block will be called for every key found in the registry on startup

Example:

require 'ruby_skynet/zookeeper'
registry = RubySkynet::Zookeeper::Registry.new(root: '/registry') do |key, value, version|
puts "Found #{key} => '#{value}' V#{version}"
end


58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 58

def initialize(params, &block)
  params = params.dup
  @root = params.delete(:root)
  raise "Missing mandatory parameter :root" unless @root

  # Add leading '/' to root if missing
  @root = "/#{@root}" unless @root.start_with?('/')

  # Strip trailing '/' if supplied
  @root = @root[0..-2] if @root.end_with?("/")
  @root_with_trail = "#{@root}/"
  @root = '/' if @root == ''

  registry_config = params.delete(:registry) || {}
  if registry_config.is_a?(::Zookeeper::Client)
    @zookeeper = registry_config
  else
    servers = registry_config.delete(:servers) || ['127.0.0.1:2181']
    connect_timeout = (registry_config.delete(:connect_timeout) || 10).to_f

    # Generate warning log entries for any unknown configuration options
    registry_config.each_pair {|k,v| logger.warn "Ignoring unknown configuration option: zookeeper.#{k}"}

    # Create Zookeeper connection
    #   server1:2181,server2:2181,server3:2181
    @zookeeper = ::Zookeeper.new(servers.join(','), connect_timeout, watcher)
  end

  # Allow the serializer and deserializer implementations to be replaced
  @serializer   = params.delete(:serializer)   || RubySkynet::Zookeeper::Json::Serializer
  @deserializer = params.delete(:deserializer) || RubySkynet::Zookeeper::Json::Deserializer

  # Generate warning log entries for any unknown configuration options
  params.each_pair {|k,v| logger.warn "Ignoring unknown configuration option: #{k}"}

  # Hash with Array values containing the list of children for each node, if any
  @children = ThreadSafe::Hash.new

  # Start watching registry for any changes
  get_recursive(@root, watch=true, create_path=true, &block)

  at_exit do
    close
  end
end

Instance Attribute Details

#root ⇒ Object (readonly)

Returns the value of attribute root.



27
28
29
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 27

def root
  @root
end

Instance Method Details

#[](key) ⇒ Object

Retrieve the latest value from a specific path from the registry Returns nil when the key is not present in the registry



106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 106

def [](key)
  result = @zookeeper.get(:path => full_key(key))
  case result[:rc]
  when ::Zookeeper::ZOK
    @deserializer.deserialize(result[:data])
  when ::Zookeeper::ZNONODE
    # Return nil if node not present
  else
    check_rc(result)
  end
end

#[]=(key, value) ⇒ Object

Replace the latest value at a specific key Supplying a nil value will result in the key being deleted in ZooKeeper



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 120

def []=(key,value)
  if value.nil?
    delete(key)
    return value
  end
  v = @serializer.serialize(value)
  k = full_key(key)
  result = @zookeeper.set(:path => k, :data => v)
  if result[:rc] == ::Zookeeper::ZNONODE
    create_path(k, v)
  else
    check_rc(result)
  end
  value
end

#close ⇒ Object

Cleanup on process termination



190
191
192
193
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 190

def close
  @zookeeper.close if @zookeeper
  @zookeeper = nil
end

#delete(key, remove_empty_parents = true) ⇒ Object

Delete the value at a specific key and any parent nodes if they don't have any children or values

Params remove_empty_parents If set to true it will also delete any parent nodes that have no children or value

Returns nil



145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 145

def delete(key, remove_empty_parents=true)
  result = @zookeeper.delete(:path => full_key(key))
  return if result[:rc] == ::Zookeeper::ZNONODE
  check_rc(result)

  if remove_empty_parents
    paths = key.split('/')
    paths.pop
    while paths.size > 0
      parent_path = full_key(paths.join('/'))
      result = @zookeeper.get(:path => parent_path)
      break if (result[:rc] == ::Zookeeper::ZNONODE) || (result[:data] != nil)

      delete(parent_path)
      paths.pop
    end
  end
  nil
end

#each_pair(relative_path = '', &block) ⇒ Object

Iterate over every key, value pair in the registry Optional relative path can be supplied Returns the number of nodes iterated over

Example:

registry.each_pair {|k,v| puts "#{k} => #{v}"}


171
172
173
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 171

def each_pair(relative_path = '', &block)
  get_recursive(full_key(relative_path), watch=false, &block)
end

#keys ⇒ Object

Returns [Array] all keys in the registry



176
177
178
179
180
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 176

def keys
  keys = []
  each_pair {|k,v| keys << k}
  keys
end

#on_create(key = '*', &block) ⇒ Object

When an entry is created the block will be called Parameters key The relative key to watch for changes block The block to be called

Parameters passed to the block:

key
 The key that was created
 Supplying a key of '*' means all paths
 Default: '*'

value
 New value from doozer

version
 The version number of this node

Example:

registry.on_update do |key, value, revision|
puts "#{key} was created with #{value}"
end

Note: They key must either be the exact path or '*' for all keys



220
221
222
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 220

def on_create(key='*', &block)
  ((@create_subscribers ||= ThreadSafe::Hash.new)[key] ||= ThreadSafe::Array.new) << block
end

#on_delete(key = '*', &block) ⇒ Object

When an entry is deleted the block will be called Parameters key The relative key to watch for changes block The block to be called

Parameters passed to the block:

key
 The key that was deleted from doozer
 Supplying a key of '*' means all paths
 Default: '*'

Example:

registry.on_delete do |key, revision|
puts "#{key} was deleted"
end

Note: They key must either be the exact path or '*' for all keys



272
273
274
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 272

def on_delete(key='*', &block)
  ((@delete_subscribers ||= ThreadSafe::Hash.new)[key] ||= ThreadSafe::Array.new) << block
end

#on_update(key = '*', &block) ⇒ Object

When an entry is updated the block will be called Parameters key The relative key to watch for changes block The block to be called

Parameters passed to the block:

key
 The key that was updated in doozer
 Supplying a key of '*' means all paths
 Default: '*'

value
 New value from doozer

version
 The version number of this node

Example:

registry.on_update do |key, value, version|
puts "#{key} was updated to #{value}"
end

Note: They key must either be the exact path or '*' for all keys



249
250
251
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 249

def on_update(key='*', &block)
  ((@update_subscribers ||= ThreadSafe::Hash.new)[key] ||= ThreadSafe::Array.new) << block
end

#to_h ⇒ Object

Returns a copy of the registry as a Hash



183
184
185
186
187
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 183

def to_h
  h = {}
  each_pair {|k,v| h[k] = v}
  h
end