Better implementation for async loop control

This commit is contained in:
Zaid Abdulla committed 2017-01-05 18:22:24 -08:00
1 parent 7d30ffd176
commit 12aa43c379
4 files changed
+94 -85

No files matched your search

+30 -23
View File
@@ -93,18 +93,18 @@ refresh = (callback) ->
lock('presets_hash_lock', 3000, (err, unlockOrExtend) -> lock('presets_hash_lock', 3000, (err, unlockOrExtend) ->
return callback(err) if err return callback(err) if err
counter = 4 counter = 3
counter += 2
db.presetsCollection.find().toArray((err, res) -> db.presetsCollection.find().toArray((err, res) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if counter & 1
counter = 0 return counter = 0
return
db.objectsCollection.find().toArray((err, objects) -> db.objectsCollection.find().toArray((err, objects) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if (counter & 1)
counter = 0 return counter = 0
return
res.sort((a, b) -> res.sort((a, b) ->
if a.weight == b.weight if a.weight == b.weight
@@ -167,13 +167,12 @@ refresh = (callback) ->
p = "#{c.name}.[#{alias}]" p = "#{c.name}.[#{alias}]"
_provisions.push(['instances', p, 0]) _provisions.push(['instances', p, 0])
else else
callback(new Error("Unknown configuration type #{c.type}")) if -- counter >= 0 callback(new Error("Unknown configuration type #{c.type}")) if counter & 1
counter = 0 return counter = 0
return
presets.push({name: preset._id, channel: preset.channel or 'default', schedule: schedule, events: events, precondition: precondition, provisions: _provisions}) presets.push({name: preset._id, channel: preset.channel or 'default', schedule: schedule, events: events, precondition: precondition, provisions: _provisions})
if -- counter == 0 if (counter -= 2) == 1
computeHash() computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) -> db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0) unlockOrExtend(0)
@@ -183,11 +182,11 @@ refresh = (callback) ->
) )
) )
counter += 2
db.provisionsCollection.find().toArray((err, res) -> db.provisionsCollection.find().toArray((err, res) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if counter & 1
counter = 0 return counter = 0
return
provisions = {} provisions = {}
for r in res for r in res
@@ -195,7 +194,7 @@ refresh = (callback) ->
provisions[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex') provisions[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex')
provisions[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: 1, timeout: 50}) provisions[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: 1, timeout: 50})
if -- counter == 0 if (counter -= 2) == 1
computeHash() computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) -> db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0) unlockOrExtend(0)
@@ -204,11 +203,11 @@ refresh = (callback) ->
) )
) )
counter += 2
db.virtualParametersCollection.find().toArray((err, res) -> db.virtualParametersCollection.find().toArray((err, res) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if counter & 1
counter = 0 return counter = 0
return
virtualParameters = {} virtualParameters = {}
for r in res for r in res
@@ -216,7 +215,7 @@ refresh = (callback) ->
virtualParameters[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex') virtualParameters[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex')
virtualParameters[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: 1, timeout: 50}) virtualParameters[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: 1, timeout: 50})
if -- counter == 0 if (counter -= 2) == 1
computeHash() computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) -> db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0) unlockOrExtend(0)
@@ -225,11 +224,11 @@ refresh = (callback) ->
) )
) )
counter += 2
db.filesCollection.find().toArray((err, res) -> db.filesCollection.find().toArray((err, res) ->
if err if err
callback(err) if -- counter >= 0 callback(err) if counter & 1
counter = 0 return counter = 0
return
files = {} files = {}
for r in res for r in res
@@ -239,7 +238,7 @@ refresh = (callback) ->
files[id].md5 = r.md5 files[id].md5 = r.md5
files[id].contentType = r.contentType files[id].contentType = r.contentType
if -- counter == 0 if (counter -= 2) == 1
computeHash() computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) -> db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0) unlockOrExtend(0)
@@ -247,6 +246,14 @@ refresh = (callback) ->
return callback() return callback()
) )
) )
if (counter -= 2) == 1
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
) )
) )
+11 -11
View File
@@ -429,31 +429,31 @@ sendRpcRequest = (currentRequest, id, rpcRequest) ->
db.clearTasks(currentRequest.sessionData.doneTasks, (err) -> db.clearTasks(currentRequest.sessionData.doneTasks, (err) ->
return throwError(err, currentRequest.httpResponse) if err return throwError(err, currentRequest.httpResponse) if err
counter = 1 counter = 3
for k of currentRequest.sessionData.faultsTouched for k of currentRequest.sessionData.faultsTouched
++ counter counter += 2
if currentRequest.sessionData.faults[k] if currentRequest.sessionData.faults[k]
currentRequest.sessionData.faults[k].retries = currentRequest.sessionData.retries[k] currentRequest.sessionData.faults[k].retries = currentRequest.sessionData.retries[k]
db.saveFault(currentRequest.sessionData.deviceId, k, currentRequest.sessionData.faults[k], (err) -> db.saveFault(currentRequest.sessionData.deviceId, k, currentRequest.sessionData.faults[k], (err) ->
-- counter if err
if err and counter >= 0 throwError(err, currentRequest.httpResponse) if counter & 1
return throwError(err, currentRequest.httpResponse) if err return counter = 0
if counter == 0 if (counter -= 2) == 1
return writeResponse(currentRequest, soap.response(null)) return writeResponse(currentRequest, soap.response(null))
) )
else else
db.deleteFault(currentRequest.sessionData.deviceId, k, (err) -> db.deleteFault(currentRequest.sessionData.deviceId, k, (err) ->
-- counter if err
if err and counter >= 0 throwError(err, currentRequest.httpResponse) if counter & 1
return throwError(err, currentRequest.httpResponse) if err return counter = 0
if counter == 0 if (counter -= 2) == 1
return writeResponse(currentRequest, soap.response(null)) return writeResponse(currentRequest, soap.response(null))
) )
if -- counter == 0 if (counter -= 2) == 1
return writeResponse(currentRequest, soap.response(null)) return writeResponse(currentRequest, soap.response(null))
) )
) )
+8 -7
View File
@@ -275,16 +275,17 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
let extCache = sandbox.extensionsCache; let extCache = sandbox.extensionsCache;
let args = arguments; let args = arguments;
let counter = 1; let counter = 3;
for (let key in sandbox.extensions) { for (let key in sandbox.extensions) {
++ counter; counter += 2;
extensions.run(sandbox.extensions[key], function(err, fault, res) { extensions.run(sandbox.extensions[key], function(err, fault, res) {
if ((err || fault) && counter > 0) { if (err || fault) {
counter = 0; if (counter & 1)
return callback(err, fault); callback(err, fault);
return counter = 0;
} }
extCache[key] = res; extCache[key] = res;
if (-- counter === 0) if ((counter -= 2) === 1)
if (status === 2) if (status === 2)
return run.apply(null, args); return run.apply(null, args);
else else
@@ -292,7 +293,7 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis
}); });
} }
if (-- counter === 0) if ((counter -= 2) === 1)
if (status === 2) if (status === 2)
return run.apply(null, args); return run.apply(null, args);
else else
+45 -44
View File
@@ -238,7 +238,7 @@ revertDownloadParameters = (sessionData, instance, callback) ->
timeoutOperations = (sessionData, callback) -> timeoutOperations = (sessionData, callback) ->
faults = [] faults = []
operations = [] operations = []
counter = 1 counter = 3
for commandKey, operation of sessionData.operations for commandKey, operation of sessionData.operations
if operation.type isnt 'Download' if operation.type isnt 'Download'
@@ -259,17 +259,16 @@ timeoutOperations = (sessionData, callback) ->
operations.push(operation) operations.push(operation)
++ counter counter += 2
revertDownloadParameters(sessionData, operation.args.instance, (err) -> revertDownloadParameters(sessionData, operation.args.instance, (err) ->
-- counter if err
if err and counter > 0 callback(err) if counter & 1
counter = 0 return counter = 0
return callback(err, faults, operations) if counter == 0 return callback(null, faults, operations) if (counter -= 2) == 1
) )
-- counter return callback(null, faults, operations) if (counter -= 2) == 1
return callback(null, faults, operations) if counter == 0
addProvisions = (sessionData, channel, provisions) -> addProvisions = (sessionData, channel, provisions) ->
@@ -322,24 +321,21 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback)
done = true done = true
allDeclarations = [] allDeclarations = []
allClear = [] allClear = []
counter = 1 counter = 3
for provision in provisions for provision in provisions
if not sessionData.cache.provisions[provision[0]]? if not sessionData.cache.provisions[provision[0]]?
if defaultProvisions[provision[0]] if defaultProvisions[provision[0]]
done = defaultProvisions[provision[0]](sessionData, provision, allDeclarations, startRevision, endRevision) and done done = defaultProvisions[provision[0]](sessionData, provision, allDeclarations, startRevision, endRevision) and done
continue continue
++ counter counter += 2
sandbox.run(sessionData.cache.provisions[provision[0]].script, sandbox.run(sessionData.cache.provisions[provision[0]].script,
{args: provision[1]}, sessionData.timestamp, sessionData.deviceData, {args: provision[1]}, sessionData.timestamp, sessionData.deviceData,
sessionData.extensionsCache, startRevision, endRevision, sessionData.extensionsCache, startRevision, endRevision,
(err, _fault, _clear, _declarations, _done) -> (err, _fault, _clear, _declarations, _done) ->
-- counter
if err or _fault if err or _fault
if counter >= 0 callback(err, _fault) if counter & 1
counter = 0 return counter = 0
return callback(err, _fault)
return
done &&= _done done &&= _done
@@ -349,11 +345,11 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback)
if _clear if _clear
allClear = allClear.concat(_clear) allClear = allClear.concat(_clear)
if counter == 0 if (counter -= 2) == 1
return callback(null, null, done, allDeclarations, allClear) return callback(null, null, done, allDeclarations, allClear)
) )
if -- counter == 0 if (counter -= 2) == 1
return callback(null, null, done, allDeclarations, allClear) return callback(null, null, done, allDeclarations, allClear)
@@ -362,20 +358,17 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
virtualParameterUpdates = [] virtualParameterUpdates = []
allDeclarations = [] allDeclarations = []
allClear = [] allClear = []
counter = 1 counter = 3
for provision in provisions for provision in provisions
++ counter counter += 2
globals = {TIMESTAMPS: provision[1], VALUES: provision[2]} globals = {TIMESTAMPS: provision[1], VALUES: provision[2]}
sandbox.run(sessionData.cache.virtualParameters[provision[0]].script, globals, sandbox.run(sessionData.cache.virtualParameters[provision[0]].script, globals,
sessionData.timestamp, sessionData.deviceData, sessionData.timestamp, sessionData.deviceData,
sessionData.extensionsCache, startRevision, endRevision, sessionData.extensionsCache, startRevision, endRevision,
(err, _fault, _clear, _declarations, _done, _returnValue) -> (err, _fault, _clear, _declarations, _done, _returnValue) ->
-- counter
if err or _fault if err or _fault
if counter >= 0 callback(err, _fault) if counter & 1
counter = 0 return counter = 0
return callback(err, _fault)
return
done &&= _done done &&= _done
@@ -388,7 +381,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
if _done if _done
virtualParameterUpdates.push(_returnValue) virtualParameterUpdates.push(_returnValue)
if counter == 0 if (counter -= 2) == 1
toClear = null toClear = null
if virtualParameterUpdates.length == provisions.length if virtualParameterUpdates.length == provisions.length
for vpu, i in virtualParameterUpdates for vpu, i in virtualParameterUpdates
@@ -399,7 +392,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal
) )
) )
if -- counter == 0 if (counter -= 2) == 1
toClear = null toClear = null
if virtualParameterUpdates.length == provisions.length if virtualParameterUpdates.length == provisions.length
for vpu, i in virtualParameterUpdates for vpu, i in virtualParameterUpdates
@@ -1566,38 +1559,46 @@ serialize = (sessionData, callback) ->
end = (sessionData, callback) -> end = (sessionData, callback) ->
counter = 2 counter = 3
counter += 2
db.saveDevice(sessionData.deviceId, sessionData.deviceData, sessionData.new, (err) -> db.saveDevice(sessionData.deviceId, sessionData.deviceData, sessionData.new, (err) ->
-- counter if err
if err and counter > 0 callback(err) if counter & 1
counter = 0 return counter = 0
return callback(err, sessionData.new) if counter == 0
return callback(null, sessionData.new) if (counter -= 2) == 1
) )
counter += 2
db.redisClient.del("session_#{sessionData.id}", (err) -> db.redisClient.del("session_#{sessionData.id}", (err) ->
-- counter if err
if err and counter > 0 callback(err) if counter & 1
counter = 0 return counter = 0
return callback(err, sessionData.new) if counter == 0
return callback(null, sessionData.new) if (counter -= 2) == 1
) )
for k of sessionData.operationsTouched for k of sessionData.operationsTouched
++ counter counter += 2
if sessionData.operations[k]? if sessionData.operations[k]?
db.saveOperation(sessionData.deviceId, k, sessionData.operations[k], (err) -> db.saveOperation(sessionData.deviceId, k, sessionData.operations[k], (err) ->
-- counter if err
if err and counter > 0 callback(err) if counter & 1
counter = 0 return counter = 0
return callback(err, sessionData.new) if counter == 0
return callback(null, sessionData.new) if (counter -= 2) == 1
) )
else else
db.deleteOperation(sessionData.deviceId, k, (err) -> db.deleteOperation(sessionData.deviceId, k, (err) ->
-- counter if err
if err and counter > 0 callback(err) if counter & 1
counter = 0 return counter = 0
return callback(err, sessionData.new) if counter == 0
return callback(null, sessionData.new) if (counter -= 2) == 1
) )
callback(null, sessionData.new) if (counter -= 2) == 1
exports.init = init exports.init = init
exports.timeoutOperations = timeoutOperations exports.timeoutOperations = timeoutOperations