-
Notifications
You must be signed in to change notification settings - Fork 109
/
synced-cron-server.js
351 lines (285 loc) · 8.72 KB
/
synced-cron-server.js
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
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
// A package for running jobs synchronized across multiple processes
SyncedCron = {
_entries: {},
running: false,
options: {
//Log job run details to console
log: true,
logger: null,
//Name of collection to use for synchronisation and logging
collectionName: 'cronHistory',
//Default to using localTime
utc: false,
//TTL in seconds for history records in collection to expire
//NOTE: Unset to remove expiry but ensure you remove the index from
//mongo by hand
collectionTTL: 172800
},
config: function(opts) {
this.options = _.extend({}, this.options, opts);
}
}
Later = Npm.require('later');
/*
Logger factory function. Takes a prefix string and options object
and uses an injected `logger` if provided, else falls back to
Meteor's `Log` package.
Will send a log object to the injected logger, on the following form:
message: String
level: String (info, warn, error, debug)
tag: 'SyncedCron'
*/
function createLogger(prefix) {
check(prefix, String);
// Return noop if logging is disabled.
if(SyncedCron.options.log === false) {
return function() {};
}
return function(level, message) {
check(level, Match.OneOf('info', 'error', 'warn', 'debug'));
check(message, String);
var logger = SyncedCron.options && SyncedCron.options.logger;
if(logger && _.isFunction(logger)) {
logger({
level: level,
message: message,
tag: prefix
});
} else {
Log[level]({ message: prefix + ': ' + message });
}
}
}
var log;
Meteor.startup(function() {
var options = SyncedCron.options;
log = createLogger('SyncedCron');
['info', 'warn', 'error', 'debug'].forEach(function(level) {
log[level] = _.partial(log, level);
});
// Don't allow TTL less than 5 minutes so we don't break synchronization
var minTTL = 300;
// Use UTC or localtime for evaluating schedules
if (options.utc)
Later.date.UTC();
else
Later.date.localTime();
// collection holding the job history records
SyncedCron._collection = new Mongo.Collection(options.collectionName);
SyncedCron._collection._ensureIndex({intendedAt: 1, name: 1}, {unique: true});
if (options.collectionTTL) {
if (options.collectionTTL > minTTL)
SyncedCron._collection._ensureIndex({startedAt: 1 },
{ expireAfterSeconds: options.collectionTTL } );
else
log.warn('Not going to use a TTL that is shorter than:' + minTTL);
}
});
var scheduleEntry = function(entry) {
var schedule = entry.schedule(Later.parse);
entry._timer =
SyncedCron._laterSetInterval(SyncedCron._entryWrapper(entry), schedule);
log.info('Scheduled "' + entry.name + '" next run @'
+ Later.schedule(schedule).next(1));
}
// add a scheduled job
// SyncedCron.add({
// name: String, //*required* unique name of the job
// schedule: function(laterParser) {},//*required* when to run the job
// job: function() {}, //*required* the code to run
// });
SyncedCron.add = function(entry) {
check(entry.name, String);
check(entry.schedule, Function);
check(entry.job, Function);
check(entry.persist, Match.Optional(Boolean));
if (entry.persist === undefined) {
entry.persist = true;
}
// check
if (!this._entries[entry.name]) {
this._entries[entry.name] = entry;
// If cron is already running, start directly.
if (this.running) {
scheduleEntry(entry);
}
}
}
// Start processing added jobs
SyncedCron.start = function() {
var self = this;
Meteor.startup(function() {
// Schedule each job with later.js
_.each(self._entries, function(entry) {
scheduleEntry(entry);
});
self.running = true;
});
}
// Return the next scheduled date of the first matching entry or undefined
SyncedCron.nextScheduledAtDate = function(jobName) {
var entry = this._entries[jobName];
if (entry)
return Later.schedule(entry.schedule(Later.parse)).next(1);
}
// Remove and stop the entry referenced by jobName
SyncedCron.remove = function(jobName) {
var entry = this._entries[jobName];
if (entry) {
if (entry._timer)
entry._timer.clear();
delete this._entries[jobName];
log.info('Removed "' + entry.name + '"');
}
}
// Pause processing, but do not remove jobs so that the start method will
// restart existing jobs
SyncedCron.pause = function() {
if (this.running) {
_.each(this._entries, function(entry) {
entry._timer.clear();
});
this.running = false;
}
}
// Stop processing and remove ALL jobs
SyncedCron.stop = function() {
_.each(this._entries, function(entry, name) {
SyncedCron.remove(name);
});
this.running = false;
}
// The meat of our logic. Checks if the specified has already run. If not,
// records that it's running the job, runs it, and records the output
SyncedCron._entryWrapper = function(entry) {
var self = this;
return function(intendedAt) {
intendedAt = new Date(intendedAt.getTime());
intendedAt.setMilliseconds(0);
var jobHistory;
if (entry.persist) {
jobHistory = {
intendedAt: intendedAt,
name: entry.name,
startedAt: new Date()
};
// If we have a dup key error, another instance has already tried to run
// this job.
try {
jobHistory._id = self._collection.insert(jobHistory);
} catch(e) {
// http://www.mongodb.org/about/contributors/error-codes/
// 11000 == duplicate key error
if (e.code === 11000) {
log.info('Not running "' + entry.name + '" again.');
return;
}
throw e;
};
}
// run and record the job
try {
log.info('Starting "' + entry.name + '".');
var output = entry.job(intendedAt,entry.name); // <- Run the actual job
log.info('Finished "' + entry.name + '".');
if(entry.persist) {
self._collection.update({_id: jobHistory._id}, {
$set: {
finishedAt: new Date(),
result: output
}
});
}
} catch(e) {
log.info('Exception "' + entry.name +'" ' + ((e && e.stack) ? e.stack : e));
if(entry.persist) {
self._collection.update({_id: jobHistory._id}, {
$set: {
finishedAt: new Date(),
error: (e && e.stack) ? e.stack : e
}
});
}
}
};
}
// for tests
SyncedCron._reset = function() {
this._entries = {};
this._collection.remove({});
this.running = false;
}
// ---------------------------------------------------------------------------
// The following two functions are lifted from the later.js package, however
// I've made the following changes:
// - Use Meteor.setTimeout and Meteor.clearTimeout
// - Added an 'intendedAt' parameter to the callback fn that specifies the precise
// time the callback function *should* be run (so we can co-ordinate jobs)
// between multiple, potentially laggy and unsynced machines
// From: https://github.com/bunkat/later/blob/master/src/core/setinterval.js
SyncedCron._laterSetInterval = function(fn, sched) {
var t = SyncedCron._laterSetTimeout(scheduleTimeout, sched),
done = false;
/**
* Executes the specified function and then sets the timeout for the next
* interval.
*/
function scheduleTimeout(intendedAt) {
if(!done) {
try {
fn(intendedAt);
} catch(e) {
log.info('Exception running scheduled job ' + ((e && e.stack) ? e.stack : e));
}
t = SyncedCron._laterSetTimeout(scheduleTimeout, sched);
}
}
return {
/**
* Clears the timeout.
*/
clear: function() {
done = true;
t.clear();
}
};
};
// From: https://github.com/bunkat/later/blob/master/src/core/settimeout.js
SyncedCron._laterSetTimeout = function(fn, sched) {
var s = Later.schedule(sched), t;
scheduleTimeout();
/**
* Schedules the timeout to occur. If the next occurrence is greater than the
* max supported delay (2147483647 ms) than we delay for that amount before
* attempting to schedule the timeout again.
*/
function scheduleTimeout() {
var now = Date.now(),
next = s.next(2, now);
// don't schedlue another occurence if no more exist synced-cron#41
if (! next[0])
return;
var diff = next[0].getTime() - now,
intendedAt = next[0];
// minimum time to fire is one second, use next occurrence instead
if(diff < 1000) {
diff = next[1].getTime() - now;
intendedAt = next[1];
}
if(diff < 2147483647) {
t = Meteor.setTimeout(function() { fn(intendedAt); }, diff);
}
else {
t = Meteor.setTimeout(scheduleTimeout, 2147483647);
}
}
return {
/**
* Clears the timeout.
*/
clear: function() {
Meteor.clearTimeout(t);
}
};
};
// ---------------------------------------------------------------------------