Class: Celluloid::Presence::ZkService

Inherits:
Object
  • Object
show all
Includes:
Celluloid, Notifications
Defined in:
lib/celluloid-presence/service.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ ZkService

Create a single connection to Zookeeper that persists through Actor crashes

servers: a list of Zookeeper servers to connect to. Each server in the

list has a host/port configuration



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/celluloid-presence/service.rb', line 28

def initialize(options)
	@@zk ||= begin
		Fanout.supervise_as :notifications_fanout if Actor[:notifications_fanout].nil?
		
		# Let them specify a single server instead of many
		servers = options[:server].nil? ? options[:servers] : [options[:server]]
		raise "no Zookeeper servers given" unless servers
		
		# Add the default Zookeeper port unless specified
		servers.map! do |server|
			if server[/:\d+$/]
				server
			else
				"#{server}:#{DEFAULT_PORT}"
			end
		end
		
		ZK.new(*servers) do |zk|
			zk.on_connecting &proc { Actor[:zk_service].async.on_connecting }
			zk.on_connected  &proc { Actor[:zk_service].async.on_connected }
		end
	end.tap do |zk|
		@@connected = zk.connected?
	end
end

Class Method Details

.init(options) ⇒ Object

Optional init function to supervise this actor



20
21
22
 
# File 'lib/celluloid-presence/service.rb', line 20

def self.init(options)
	ZkService.supervise_as :zk_service, options
end

Instance Method Details

#connected? ⇒ Boolean

Our abstracted connected status to avoid race conditions

Returns:

  • (Boolean) 


113
114
115
 
# File 'lib/celluloid-presence/service.rb', line 113

def connected?
	@@connected
end

#ensure(func, *args) ⇒ Object

Ensures important requests are followed through in the case of disconnect / reconnect Ignores request if the node does not exist USE SPARINGLY



122
123
124
125
126
127
128
 
# File 'lib/celluloid-presence/service.rb', line 122

def ensure(func, *args)
	if connected?
		ensured(func, *args)
	else
		@@ensure.push [func, args]
	end
end

#event_callback(path, event) ⇒ Object

Publishes any zookeeper events that have been registered



79
80
81
 
# File 'lib/celluloid-presence/service.rb', line 79

def event_callback(path, event)
	publish("zk_event_#{path}", event)
end

#on_connected ⇒ Object

Called once a connection is made to the zookeeper cluster Runs any impending ensures before informing any subscribed actors



58
59
60
61
62
63
64
65
 
# File 'lib/celluloid-presence/service.rb', line 58

def on_connected
	@@connected = true
	while not @@ensure.empty?
		func = @@ensure.shift
		ensured func[0], *func[1]	# func[0] == function name, func[1] == arguments
	end
	publish('zk_connected')
end

#on_connecting ⇒ Object

Called on disconnect from zookeeper Might be that the zookeeper node crashed or zookeeper is down, may not effect any other connections



71
72
73
74
 
# File 'lib/celluloid-presence/service.rb', line 71

def on_connecting
	@@connected = false
	publish('zk_connecting')
end

#register(path) ⇒ Object

Uses a closure to cleanly pull the zookeeper event back into the protection of celluloid



86
87
88
89
90
91
92
93
 
# File 'lib/celluloid-presence/service.rb', line 86

def register(path)
	path = path.to_sym
	if @@registry[path].nil?
		callback = proc { |event| Actor[:zk_service].async.event_callback(path, event) }
		@@registry[path] = @@zk.register(path.to_s, &callback)
	end
	@@registry_count[path] = (@@registry_count[path] || 0) + 1
end

#unregister(path) ⇒ Object

unsubscribes from zookeeper events once there are no more listeners



98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/celluloid-presence/service.rb', line 98

def unregister(path)
	path = path.to_sym
	if @@registry[path]
		@@registry_count[path] -= 1
		if @@registry_count[path] <= 0
			sub = @@registry.delete(path)
			sub.unsubscribe
			@@registry_count.delete(path)
		end
	end
end