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

:ephemeral [Boolean] All set operations of non-nil values will result in ephemeral nodes.

:on_connect [Proc] Block to call after the connection to Zookeeper has been established and every time the connection is re-established

:registry [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


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
103
104
105
106
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 65

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) || {}

  #   server1:2181,server2:2181,server3:2181
  @servers = (registry_config.delete(:servers) || ['127.0.0.1:2181']).join(',')
  @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}"}

  # 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

  @ephemeral = params.delete(:ephemeral)
  @ephemeral = false if @ephemeral.nil?

  @on_connect = params.delete(:on_connect)

  # 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

  # Block is used in init
  @block = block

  self.init
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



110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 110

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



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 124

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



194
195
196
197
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 194

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



149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 149

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}"}


175
176
177
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 175

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



180
181
182
183
184
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 180

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 the registry

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



224
225
226
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 224

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 the registry
 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



276
277
278
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 276

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 the registry
 Supplying a key of '*' means all paths
 Default: '*'

value
 New value from the registry

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



253
254
255
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 253

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



187
188
189
190
191
 
# File 'lib/ruby_skynet/zookeeper/registry.rb', line 187

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