Class: SeapigClient

Inherits:
Object
  • Object
show all
Defined in:
lib/seapig-client-ruby/client.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(uri, options = {}) ⇒ SeapigClient

Returns a new instance of SeapigClient.



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/seapig-client-ruby/client.rb', line 30

def initialize(uri, options={})
	@uri = uri
	@options = options
	@slave_objects = {}
	@master_objects = {}
	@connected = false
	@socket = nil
	@error = nil
	connect
end

Instance Attribute Details

#connected ⇒ Object (readonly)

Returns the value of attribute connected.



28
29
30
 
# File 'lib/seapig-client-ruby/client.rb', line 28

def connected
  @connected
end

#error ⇒ Object (readonly)

Returns the value of attribute error.



28
29
30
 
# File 'lib/seapig-client-ruby/client.rb', line 28

def error
  @error
end

#socket ⇒ Object (readonly)

Returns the value of attribute socket.



28
29
30
 
# File 'lib/seapig-client-ruby/client.rb', line 28

def socket
  @socket
end

#uri ⇒ Object (readonly)

Returns the value of attribute uri.



28
29
30
 
# File 'lib/seapig-client-ruby/client.rb', line 28

def uri
  @uri
end

Instance Method Details

#connect ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/seapig-client-ruby/client.rb', line 42

def connect

	disconnect if @socket
	@reconnect_on_close = true

	@timeout_timer ||= EM.add_periodic_timer(10) {
		next if not @socket
		next if Time.new.to_f - @last_communication_at < 20
		puts "Seapig ping timeout, reconnecting" if @options[:debug]
		connect
	}

	@last_communication_at = Time.new.to_f

	puts 'Connecting to seapig server' if @options[:debug]
	@socket = WebSocket::EventMachine::Client.connect(uri: @uri)

	@socket.onopen {
		puts 'Connected to seapig server' if @options[:debug]
		@connected = true
		@error = nil
		@onstatuschange_proc.call(self) if @onstatuschange_proc
		@socket.send JSON.dump(action: 'client-options-set', options: @options)
		@slave_objects.each_pair { |id, object|
			@socket.send JSON.dump(action: 'object-consumer-register', pattern: id, :"version-known" => object.version)
			object.validate
		}
		@master_objects.each_pair { |id, object|
			@socket.send JSON.dump(action: 'object-producer-register', pattern: id, :"version-known" => object.version)
			object.children.each { |child_id, child| child.upload(0, {}, child.version, true) } if id.include?('*')
		}
		@last_communication_at = Time.new.to_f
	}

	@socket.onclose(&(@socket_onclose = Proc.new { |code, reason|
		puts 'Seapig connection closed (code:'+code.inspect+', reason:'+reason.inspect+')' if @options[:debug]
		@connected = false
		@socket = nil
		@timeout_timer.cancel if @timeout_timer
		@timeout_timer = nil
		@slave_objects.values.each { |object| object.invalidate }
		@onstatuschange_proc.call(self) if @onstatuschange_proc
		EM.cancel_timer(@reconnection_timer) if @reconnection_timer
		@reconnection_timer = nil
		@reconnection_timer = EM.add_timer(1) { connect } if @reconnect_on_close
	}))

	@socket.onerror { |error|
		puts 'Seapig socket error: '+error.inspect if @options[:debug]
		@error = { while: @socket ? "connecting" : "connected", error: error }
		@socket_onclose.call(nil, error) if @socket
	}

	@socket.onmessage { |data|
		message = JSON.load(data)
		case message['action']
		when 'object-update'
			@slave_objects.each_pair { |id, object|
				object.patch(message) if object.matches(message['id'])
			}
		when 'object-destroy'
			@slave_objects.each_pair { |id, object|
				object.destroy(message['id']) if object.matches(message['id'])
			}
			@master_objects.each_pair { |id, object|
				object.destroy(message['id']) if object.matches(message['id'])
			}
		when 'object-produce'
			handler = @master_objects.values.find { |object| object.matches(message['id']) }
			puts 'Seapig server submitted invalid "produce" request: '+message.inspect if (not handler) and @options[:debug]
			handler.produce(message['id'],message['version-inferred']) if handler
		else
			raise 'Seapig server submitted an unsupported message: '+message.inspect if @options[:debug]
		end
		@last_communication_at = Time.new.to_f
	}

	@socket.onping {
		@last_communication_at = Time.new.to_f
	}

end

#detach_fd ⇒ Object

FIXME: this is, most likely, broken



150
151
152
 
# File 'lib/seapig-client-ruby/client.rb', line 150

def detach_fd #FIXME: this is, most likely, broken
	disconnect(true)
end

#disconnect(detach_fd = false) ⇒ Object 



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/seapig-client-ruby/client.rb', line 126

def disconnect(detach_fd = false)
	@reconnect_on_close = false
	if @timeout_timer
		@timeout_timer.cancel
		@timeout_timer = nil
	end
	if @reconnection_timer
		EM.cancel_timer(@reconnection_timer)
		@reconnection_timer = nil
	end
	if @socket
		if detach_fd #FIXME: this is, most likely, broken
			IO.new(@socket.detach).close
			@socket.onclose {}
			@socket_onclose.call("fd detach", "fd detach")
		else
			@socket.onclose {}
			@socket.close
			@socket_onclose.call("close","close")
		end
	end
end

#master(id, options = {}) ⇒ Object 



169
170
171
172
173
 
# File 'lib/seapig-client-ruby/client.rb', line 169

def master(id, options={})
	@master_objects[id] = if id.include?('*') then SeapigWildcardMasterObject.new(self, id, options) else SeapigMasterObject.new(self, id, options) end
	@socket.send JSON.dump(action: 'object-producer-register', pattern: id, :"version-known" => @master_objects[id].version) if @connected
	@master_objects[id]
end

#onstatuschange(&block) ⇒ Object 



155
156
157
158
 
# File 'lib/seapig-client-ruby/client.rb', line 155

def onstatuschange(&block)
	@onstatuschange_proc = block
	self
end

#slave(id, options = {}) ⇒ Object 



161
162
163
164
165
166
 
# File 'lib/seapig-client-ruby/client.rb', line 161

def slave(id, options={})
	raise "Both or none of 'object' and 'version' are needed" if (options[:object] and not options[:version]) or (not options[:object] and options[:version])
	@slave_objects[id] = if id.include?('*') then SeapigWildcardSlaveObject.new(self, id, options) else SeapigSlaveObject.new(self, id, options) end
	@socket.send JSON.dump(action: 'object-consumer-register', pattern: id, :"version-known" => @slave_objects[id].version) if @connected
	@slave_objects[id]
end

#unlink(id) ⇒ Object 



176
177
178
179
180
181
182
183
184
185
 
# File 'lib/seapig-client-ruby/client.rb', line 176

def unlink(id)
	if @slave_objects[id]
		@slave_objects.delete(id)
		@socket.send(JSON.stringify(action: 'object-consumer-unregister', pattern: id)) if @connected
	end
	if @master_objects[id]
		@master_objects.delete(id)
		@socket.send(JSON.stringify(action: 'object-producer-unregister', pattern: id)) if @connected
	end
end