Permalink
Browse files

processing NextToken for select queries over 2500

  • Loading branch information...
tracend committed May 28, 2013
1 parent 236aad0 commit ca66186349dd0441a8a39ae41c25797a00ac9b82
Showing with 24 additions and 14 deletions.
  1. +24 −14 lib/simpledb.js
View
@@ -28,13 +28,13 @@ var MARK = 'simpledb: '
* delay: delay before this try
*
* statuscb(done,tryI,last,delay,err)
- * done: boolean return value from your function
+ * done: boolean return value from your function
* tryI: try index, starts from 1
* last: true if this is the last time statuscb will be called
* delay: delay before this try
* err: any captured errors thrown from your function
*
- * Notes:
+ * Notes:
* You need to handle and log simpledb errors yourself inside your own
* function fn. Generally you don't want to log "Service
* Unavailable" messages are these expected, as that's the whole
@@ -48,6 +48,7 @@ exports.expbackoff = function(fn,statuscb,maxtry,expbase,delaymin,delayscale,ran
delaymin = null == delaymin ? 0 : delaymin
delayscale = null == delayscale ? 100 : delayscale
randomdelay = null == randomdelay ? true : randomdelay
+ maxlimit = 2500
function retry(tryI,last,delay) {
try {
@@ -119,7 +120,8 @@ exports.SimpleDB = function(opts,logger) {
// TODO - get aws-lib to support port
awsopts.port = null == opts.port ? 80 : opts.port
-
+ var results = []
+
log('create',opts,awsopts)
self.client = aws.createSimpleDBClient(opts.keyid, opts.secret, awsopts)
@@ -130,7 +132,6 @@ exports.SimpleDB = function(opts,logger) {
var time = new Date().getTime()
var err = null
var meta = {action:act,query:q,result:res,time:time,duration:time-start,trycount:tryI}
-
var retry = false
if( res.Errors ) {
@@ -150,8 +151,16 @@ exports.SimpleDB = function(opts,logger) {
}
if( !err ) {
- stop(true)
- callback(err,res,meta)
+ results = results.concat( res.SelectResult.Item )
+ if( res.SelectResult.Item.length == maxlimit && res.SelectResult.NextToken){
+ // get the next batch of results
+ q.NextToken = res.SelectResult.NextToken
+ makereq(act,q,callback)
+ } else {
+ stop(true)
+ res.SelectResult.Item = results
+ callback(err,res,meta)
+ }
}
else {
log('error',start,act,q,tryI,last,retry,err,res,meta)
@@ -194,7 +203,7 @@ exports.SimpleDB = function(opts,logger) {
opts.statuscb && opts.statuscb(done,tryI,last,delay,err)
},
- opts.maxtry,opts.expbase,opts.delaymin,opts.delayscale,opts.randomdelay
+ opts.maxtry,opts.expbase,opts.delaymin,opts.delayscale,opts.randomdelay
)
}
}
@@ -230,7 +239,7 @@ exports.SimpleDB = function(opts,logger) {
})
}
-
+
function empty(name,value,callback) {
var fail = false
try {
@@ -239,7 +248,7 @@ exports.SimpleDB = function(opts,logger) {
}
if( 'string' != typeof(value) ) {
throw MARK+name+' is not a string'
- }
+ }
if( '' == ''+value ) {
throw MARK+name+' is empty'
}
@@ -407,20 +416,20 @@ exports.SimpleDB = function(opts,logger) {
callback(err,out,meta)
})
}
-
+
self.batchDeleteItem = function(domain, items, override, callback) {
callback = getcallback(override,callback)
if( badargs(callback,domain ))return
-
+
var act = 'BatchDeleteAttributes'
var q = {DomainName:domain}
-
+
for( var itemI = 1; itemI <= items.length; itemI++ ) {
var attrs = items[itemI-1]
q['Item.'+itemI+'.ItemName'] = attrs.$ItemName
putattrs(q,attrs,'Item.'+itemI+'.')
}
-
+
applyoverride(q,override)
makereq(act,q,function(err,res,meta){
var out = err ? null : {}
@@ -473,7 +482,7 @@ exports.SimpleDB = function(opts,logger) {
for( var aI = 1; aI <= attrs.length; aI++ ) {
q['Attribute.'+aI+'.Name'] = attrs[aI-1]
}
- }
+ }
else {
putattrs(q,attrs,'');
}
@@ -492,6 +501,7 @@ exports.SimpleDB = function(opts,logger) {
override = args
args = []
}
+ results = []
callback = getcallback(override,callback)
if( badargs(callback ))return
if( empty('query',query,callback ))return

0 comments on commit ca66186

Please sign in to comment.