Created
August 7, 2015 19:42
-
-
Save bengotow/2f1333683010230b6d91 to your computer and use it in GitHub Desktop.
concurrency - rolling average
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
var Promise = require('bluebird'); | |
var request = Promise.promisify(require('request')); | |
var fs = require('fs'); | |
global.Nylas = require('nylas').config({ | |
appId: '1', | |
appSecret: '2', | |
apiServer: 'https://api.nylas.com' | |
}); | |
var MAX_COUNT = 5000; | |
var CONCURRENCY = 20; | |
var deltasProcessed = 0; | |
var deltasProcessedStart = 0; | |
var speedRollingAverage = 0; | |
var startStream = function (id) { | |
var Namespace = null; | |
Nylas.with(<<NAMESPACE ID>>).namespaces.first({}) | |
.then(function(namespace) { | |
Namespace = namespace; | |
return Namespace.deltas.generateCursor(0); | |
}) | |
.then(function(cursor) { | |
console.log("Opening stream "+id); | |
var DELTA_EXCLUDE_TYPES = ['contact', 'event', 'file', 'tag', 'thread']; | |
var stream = Namespace.deltas.startStream(cursor, DELTA_EXCLUDE_TYPES); | |
stream.on('data', function () { | |
console.log("["+id+"] got data"); | |
}); | |
stream.on('delta', function(delta) { | |
// Handle the new delta here. | |
deltasProcessed += 1; | |
if (Date.now() - deltasProcessedStart > 5000) { | |
speedRollingAverage = speedRollingAverage * 0.8 + deltasProcessed * 0.2; | |
console.log("Deltas in last 5000 msec: "+deltasProcessed + ". Rolling average of 5 samples: " +speedRollingAverage); | |
deltasProcessed = 0; | |
deltasProcessedStart = Date.now(); | |
} | |
}).on('error', function(err) { | |
// Handle errors here, such as by restarting the stream at the last cursor. | |
console.error('Delta streaming error:', err); | |
}); | |
}); | |
}; | |
// delayed so I can start the profiler | |
setTimeout(function() { | |
startTime = new Date(); | |
console.log('Running '+CONCURRENCY+' assistants concurrently'); | |
for (var i = 0; i < CONCURRENCY; i++) { | |
startStream(i); | |
} | |
}, 5000); | |
// delay process exit forever so I can view the profile | |
setTimeout(function(){}, 20000000); |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment