| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207 |
1
1
50
50
1
51
52
52
57
57
57
57
109
109
109
109
109
109
84
84
109
109
109
109
109
109
109
9
9
109
109
1
140
140
140
140
140
1
84
84
84
84
84
32
24
24
24
24
24
24
24
24
24
8
5
5
4
3
3
3
9
9
12
12
12
264
| /*
* Copyright 2014-2016, Sébastien Piquemal <sebpiq@gmail.com>
*
* rhizome is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* rhizome is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with rhizome. If not, see <http://www.gnu.org/licenses/>.
*/
"use strict";
var EventEmitter = require('events').EventEmitter
, _ = require('underscore')
, async = require('async')
, connections = require('../connections')
, coreMessages = require('./messages')
, coreUtils = require('./utils')
// Base class for all Servers.
// Subclasses must extend at least `start(done)` and `stop(done)`.
// Events:
// - 'connection' : emitted when a new connection has been opened.
var Server = exports.Server = function(config) {
EventEmitter.apply(this)
this.connections = []
}
_.extend(Server.prototype, EventEmitter.prototype, {
// Class used to represent server connections
ConnectionClass: null,
// Overwrite with `debug` function from node-debug
debug: function() {},
// Starts the server, and calls `done(err)` when done.
// This is just a stub that subclasses must complement.
start: function(done) {
this.debug('starting')
},
// Stops the server, and calls `done(err)` when done.
// This is just a stub that subclasses must complement.
// !!! Subclasses implementing their own `stop` must call this last,
// because it unbinds event handlers, including error handlers.
stop: function(done) {
this.debug('stopping')
async.each(this.connections.slice(0), (connection, next) => {
connection.once('close', () => {
this.connections = _.without(this.connections, connection)
next()
})
connection.close()
}, done)
},
// Creates and opens a new connection with arguments `args`, then calls `done(err, connection)`
openConnection: function(args, done) {
var connection = new this.ConnectionClass(args)
, queuedSysMessages = []
, onSysMessage
this.connections.push(connection)
// Replace `onSysMessage` handler with a method that will just queue received messages.
// We will then send these queued messages to the actual `onSysMessage` handler,
// after connection is opened.
onSysMessage = connection.onSysMessage
connection.onSysMessage = function() { queuedSysMessages.push(arguments) }
connection.on('error', (err) => this.emit('error', err))
connection.on('close', () => {
this.connections = _.without(this.connections, connection)
this.debug(connection.toString() + ' closed - total connections ' + this.connections.length)
})
connections.manager.open(connection, (err) => {
Iif (err) return done(err)
connection.status = 'open'
this.debug(connection.toString() + ' opened - total connections ' + this.connections.length)
// Restore `onSysMessages` method and execute queued messages
connection.onSysMessage = onSysMessage
queuedSysMessages.forEach((args) => connection.onSysMessage.apply(connection, args))
// Restore subscriptions
connection._subscriptions.forEach((address) => {
var err = connections.manager.subscribe(connection, address)
Iif (err) this.emit('error', new Error('invalid subscription : ' + err))
})
done(null, connection)
this.emit('connection', connection)
})
}
})
// Base class for different types of connections.
// Subclasses must implement at least `send(address, args)`.
// Events:
// - 'close' : emitted when the connection is closing.
var Connection = exports.Connection = function(args) {
EventEmitter.apply(this)
this._subscriptions = []
this.status = 'closed'
this.id = null
// An object to store all sorts of infos about the connection,
// those will be persisted.
this.infos = {}
}
_.extend(Connection.prototype, EventEmitter.prototype, {
// Namespace for the connection. Allows to separate different types of connections.
namespace: null,
// If true, an id will automatically be assigned when calling `connection.open`,
// if the connection doesn't already have an id.
autoId: false,
// Closes the connection and emit a `close` event.
close: function() {
connections.manager.close(this, (err) => {
Iif (err) return this.emit('error', err)
this.status = 'closed'
this.emit('close')
this.removeAllListeners()
})
},
// Sends a message to the connection (in most cases, to the remotely connected client)
send: function(address, args) {
throw new Error('Implement me')
},
// Handler for common system message between all types of connections.
onSysMessage: function(address, args) {
// When a client wants to receive messages sent to an address,
// we subscribe him and send acknowldgement.
if (address === coreMessages.subscribeAddress) {
var toAddress = args[0]
, err = connections.manager.subscribe(this, toAddress)
Iif (err) this.send(coreMessages.errorAddress, [err])
else {
this.send(coreMessages.subscribedAddress, [toAddress])
// Add the new subscription to `_subscriptions` list and save changes.
var countBefore = this._subscriptions.length
this._subscriptions.push(toAddress)
this._subscriptions = _.uniq(this._subscriptions)
Eif (countBefore < this._subscriptions.length) {
connections.manager.connectionUpdate(this, (err) => {
Iif (err) this.emit('error', err)
})
}
}
// Resends last messages received at the given address.
} else if (address === coreMessages.resendAddress) {
var fromAddress = args[0]
, resentArgs = connections.manager.getLastMessage(fromAddress)
if (resentArgs !== null)
this.send(fromAddress, resentArgs)
// Sends a list of the ids of the open connections for the given namespace
} else Eif (address === coreMessages.connectionsSendListAddress) {
var namespace = args[0]
, idList = connections.manager.getOpenConnectionsIds(namespace)
this.send(coreMessages.connectionsTakeListAddress + '/' + namespace, idList)
}
},
// Persist connection
save: function() {
connections.manager.connectionUpdate(this, (err) => {
err && this.emit('error', new Error('error updating connection : ' + err.message))
})
},
// Serializes the connection data to persist it to the database
serialize: function() {
return {
subscriptions: this._subscriptions,
infos: this.infos
}
},
// Deserialize the persisted connection data
deserialize: function(data) {
this.infos = data.infos
this._subscriptions = data.subscriptions
},
toString: function() { return 'Connection(' + this.namespace + ':' + this.id + ')' }
}) |