@@ -7,9 +7,12 @@ VertexDB Queue := Object clone do( target ::= nil messageName ::= nil expiresActive ::= true - concurrency ::= 10 + concurrency ::= 1 node ::= nil qNodes ::= nil + itemsPerProcess ::= nil //if non nil then process at most itemsPerProcess + jobs ::= nil + processedCount ::= 0 waitingNode ::= method(node nodeAt("waiting")) activeNode ::= method(node nodeAt("active")) @@ -52,12 +55,17 @@ VertexDB Queue := Object clone do( jobError ::= nil + processCountLimited := method( + if(itemsPerProcess, processedCount >= itemsPerProcess, false) + ) + process := method( setJobError(nil) if(expiresActive, expireActive) processedOne := false - self jobs := List clone - while(k := waitingNode queueToNode(activeNode), + setJobs(List clone) + setProcessedCount(0) + while(processCountLimited not and (k := waitingNode queueToNode(activeNode)), debugWriteln("waitingNode<", waitingNode path, "> queueToNode(<", activeNode path, ">) == <", k, ">") if(k == nil, break) debugWriteln("jobs size: ", jobs size) @@ -113,6 +121,7 @@ VertexDB Queue := Object clone do( ) finishedJob := method(job, + setProcessedCount(processedCount + 1) self jobs remove(job) ) addons/VertexDB/io/protos/Queue.io bd050fa9ccf225a6a8794495f4e348d525731ace Rich Collins richcollins@gmail.com http://github.com/stevedekorte/io/commit/8104948b668b605bf5d0af7350502941cfe13de8 8104948b668b605bf5d0af7350502941cfe13de8 2009-10-30T16:55:55-07:00 2009-10-30T16:55:55-07:00 VertexDB: Queue: Use default concurrency of 1. Added maxItemsPerProcess 417d48107e5a4d0612fa4f154fcf1828688b64e0 Rich Collins richcollins@gmail.com