Repository navigation
Expand file tree
/
Copy pathserver.js
More file actions
235 lines (204 loc) · 7.67 KB
/
Copy pathserver.js
File metadata and controls
235 lines (204 loc) · 7.67 KB
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
var assign = require('lodash.assign'),
bind = require('lodash.bind'),
isEmpty = require('lodash.isempty'),
jsondiffpatch = require('jsondiffpatch'),
COMMANDS = require('./commands'),
utils = require('./utils'),
Server;
Server = function(adapter, transport, diffOptions){
if(!(adapter && transport)){ throw new Error('Need to specify an adapter and a transport'); }
if(!diffOptions){ diffOptions = {}; }
this.adapter = adapter;
this.transport = transport;
this.data = {};
this.requests = {};
this.saveRequests = {};
this.saveQueue = {};
// bind functions
this.trackConnection = bind(this.trackConnection, this);
// set up the jsondiffpatch options
// see here for options: https://github.com/benjamine/jsondiffpatch#options
diffOptions = assign({
objectHash: function(obj) { return obj.id || obj._id || JSON.stringify(obj); }
}, diffOptions);
this.jsondiffpatch = jsondiffpatch.create(diffOptions);
this.transport.on('connection', this.trackConnection);
};
/**
* Registers the correct event listeners
* @param {Connection} connection The connection that should get tracked
*/
Server.prototype.trackConnection = function(connection){
connection.on(COMMANDS.join, this.joinConnection.bind(this, connection));
connection.on(COMMANDS.syncWithServer, this.receiveEdit.bind(this, connection));
};
/**
* Joins a connection to a room and send the initial data
* @param {Connection} connection
* @param {String} room room identifier
* @param {Function} initializeClient Callback that is being used for initialization of the client
*/
Server.prototype.joinConnection = function(connection, room, initializeClient){
this.getData(room, function(error, data){
// connect to the room
connection.join(room);
// set up the client version for this socket
// each connection has a backup and a shadow
// and a set of edits
data.clientVersions[connection.id] = {
backup: {
doc: utils.deepCopy(data.serverCopy),
serverVersion: 0
},
shadow: {
doc: utils.deepCopy(data.serverCopy),
serverVersion: 0,
localVersion: 0
},
edits: []
};
// send the current server version
initializeClient(data.serverCopy);
});
};
/**
* Gets data for a room from the internal cache or from the adapter
* @param {String} room room identifier
* @param {Function} callback notifier-callback
*/
Server.prototype.getData = function(room, callback){
var cachedVersion = this.data[room],
cache = this.data,
requests = this.requests;
if(cachedVersion){
callback(null, cachedVersion);
}else{
// if there is no request for this room
// ask the adapter for the data
// do nothing in the else case because this operation
// should only happen once
if(!requests[room]){
requests[room] = true;
this.adapter.getData(room, function(error, data){
cache[room] = {
registeredSockets: [],
clientVersions: {},
serverCopy: data
};
requests[room] = false;
callback(null, cache[room]);
});
}else{
requests[room] = true;
}
}
};
/**
* Applies the sent edits to the shadow and the server copy, notifies all connected sockets and saves a snapshot
* @param {Object} connection The connection that sent the edits
* @param {Object} editMessage The message containing all edits
* @param {Function} sendToClient The callback that sends the server changes back to the client
*/
Server.prototype.receiveEdit = function(connection, editMessage, sendToClient){
// -1) The algorithm actually says we should use a checksum here, I don't think that's necessary
// 0) get the relevant doc
this.getData(editMessage.room, function(err, doc){
// 0.a) get the client versions
var clientDoc = doc.clientVersions[connection.id];
// no client doc could be found, client needs to re-auth
if(err || !clientDoc){
connection.emit(COMMANDS.error, 'Need to re-connect!');
return;
}
// when the versions match, remove old edits stack
if(editMessage.serverVersion === clientDoc.shadow.serverVersion){
clientDoc.edits = [];
}
// 1) iterate over all edits
editMessage.edits.forEach(function(edit){
// 2) check the version numbers
if(edit.serverVersion === clientDoc.shadow.serverVersion &&
edit.localVersion === clientDoc.shadow.localVersion){
// versions match
// backup! TODO: is this the right place to do that?
clientDoc.backup.doc = utils.deepCopy(clientDoc.shadow.doc);
// 3) patch the shadow
// var snapshot = utils.deepCopy(clientDoc.shadow.doc);
this.jsondiffpatch.patch(clientDoc.shadow.doc, utils.deepCopy(edit.diff));
// clientDoc.shadow.doc = snapshot;
// apply the patch to the server's document
// snapshot = utils.deepCopy(doc.serverCopy);
this.jsondiffpatch.patch(doc.serverCopy, utils.deepCopy(edit.diff));
// doc.serverCopy = snapshot;
// 3.a) increase the version number for the shadow if diff not empty
if(!isEmpty(edit.diff)){
clientDoc.shadow.localVersion++;
}
}else{
// TODO: implement backup workflow
// has a low priority since `packets are not lost` - but don't quote me on that :P
console.log('error', 'patch rejected!!', edit.serverVersion, '->', clientDoc.shadow.serverVersion, ':',
edit.localVersion, '->', clientDoc.shadow.localVersion);
}
}.bind(this));
// 4) save a snapshot of the document
this.saveSnapshot(editMessage.room);
// notify all sockets about the update, if not empty
if(editMessage.edits.length > 0){
this.transport.to(editMessage.room).emit(COMMANDS.remoteUpdateIncoming, connection.id);
}
this.sendServerChanges(doc, clientDoc, sendToClient);
}.bind(this));
};
Server.prototype.saveSnapshot = function(room){
var noRequestInProgress = !this.saveRequests[room],
checkQueueAndSaveAgain = function(){
// if another save request is in the queue, save again
var anotherRequestScheduled = this.saveQueue[room] === true;
this.saveRequests[room] = false;
if(anotherRequestScheduled){
this.saveQueue[room] = false;
this.saveSnapshot(room);
}
}.bind(this);
// only save if no save going on at the moment
if(noRequestInProgress){
this.saveRequests[room] = true;
// get data for saving
this.getData(room, function(err, data){
// store data
if(!err && data){
this.adapter.storeData(room, data.serverCopy, checkQueueAndSaveAgain);
}else{
checkQueueAndSaveAgain();
}
}.bind(this));
}else{
// schedule a new save request
this.saveQueue[room] = true;
}
};
Server.prototype.sendServerChanges = function(doc, clientDoc, send){
// create a diff from the current server version to the client's shadow
var diff = this.jsondiffpatch.diff(clientDoc.shadow.doc, doc.serverCopy);
var basedOnServerVersion = clientDoc.shadow.serverVersion;
// add the difference to the server's edit stack
if(!isEmpty(diff)){
clientDoc.edits.push({
serverVersion: basedOnServerVersion,
localVersion: clientDoc.shadow.localVersion,
diff: diff
});
// update the server version
clientDoc.shadow.serverVersion++;
// apply the patch to the server shadow
this.jsondiffpatch.patch(clientDoc.shadow.doc, utils.deepCopy(diff));
}
// we explicitly want empty diffs to get sent as well
send({
localVersion: clientDoc.shadow.localVersion,
serverVersion: basedOnServerVersion,
edits: clientDoc.edits
});
};
module.exports = Server;