Merge 3704f269a1320af88cb9aa05653b83124a9cf105 into 39000fd4fc5c9a61d1641f8cc2ce7948777d1e80

datastore/fsds.nim

@@ -190,6 +190,9 @@ method query*(
     iter = QueryIter.new()
 
   proc next(): Future[?!QueryResponse] {.async.} =
+    if iter.finished:
+      return success (Key.none, EmptyBytes)
+
     let
       path = walker()
 
@@ -216,6 +219,10 @@ method query*(
     return success (key.some, data)
 
   iter.next = next
+  iter.dispose = proc(): Future[?!void] {.async.} =
+    iter.finished = true
+    return success()
+  
   return success iter
 
 method modifyGet*(

datastore/leveldb/leveldbds.nim

@@ -100,7 +100,7 @@ method query*(
 
   proc next(): Future[?!QueryResponse] {.async.} =
     if iter.finished:
-      return failure(newException(QueryEndedError, "Calling next on a finished query!"))
+      return success (Key.none, EmptyBytes)
 
     try:
       let (keyStr, valueStr) = dbIter.next()
@@ -116,6 +116,7 @@ method query*(
 
   proc dispose(): Future[?!void] {.async.} =
     dbIter.dispose()
+    iter.finished = true
     return success()
 
   iter.next = next

datastore/sql/sqliteds.nim

@@ -269,7 +269,7 @@ method query*(
 
   proc next(): Future[?!QueryResponse] {.async.} =
     if iter.finished:
-      return failure(newException(QueryEndedError, "Calling next on a finished query!"))
+      return success (Key.none, EmptyBytes)
 
     let
       v = sqlite3_step(s)
@@ -324,7 +324,7 @@ method query*(
   iter.dispose = proc(): Future[?!void] {.async.} =
     discard sqlite3_reset(s)
     discard sqlite3_clear_bindings(s)
-    iter.next = nil
+    iter.finished = true
     return success()
 
   iter.next = next

tests/datastore/querycommontests.nim

@@ -44,6 +44,139 @@ template queryTests*(ds: Datastore, testLimitsAndOffsets = true, testSortOrder =
 
     (await iter.dispose()).tryGet
 
+  test "Concurrent queries":
+    let
+      q1 = Query.init(key1)
+      q2 = Query.init(key1)
+
+    (await ds.put(key1, val1)).tryGet
+    (await ds.put(key2, val2)).tryGet
+    (await ds.put(key3, val3)).tryGet
+
+    let
+      iter1 = (await ds.query(q1)).tryGet
+      iter2 = (await ds.query(q1)).tryGet
+
+    let one1 = (await iter1.next()).tryGet
+    check not iter1.finished
+
+    let one2 = (await iter2.next()).tryGet
+    check not iter2.finished
+
+    let two1 = (await iter1.next()).tryGet
+    check not iter1.finished
+    let three1 = (await iter1.next()).tryGet
+    check not iter1.finished
+
+    let two2 = (await iter2.next()).tryGet
+    check not iter2.finished
+    let three2 = (await iter2.next()).tryGet
+    check not iter2.finished
+
+    let four1 = (await iter1.next()).tryGet
+    check iter1.finished
+    let four2 = (await iter2.next()).tryGet
+    check iter2.finished
+
+    check:
+      one1[0].get == key1
+      one1[1] == val1
+      two1[0].get == key2
+      two1[1] == val2
+      three1[0].get == key3
+      three1[1] == val3
+      four1[0] == Key.none
+
+      one2[0].get == key1
+      one2[1] == val1
+      two2[0].get == key2
+      two2[1] == val2
+      three2[0].get == key3
+      three2[1] == val3
+      four2[0] == Key.none
+
+    (await iter1.dispose()).tryGet
+    (await iter2.dispose()).tryGet
+
+  test "Concurrent queries - dispose":
+    let
+      q1 = Query.init(key1)
+      q2 = Query.init(key1)
+
+    (await ds.put(key1, val1)).tryGet
+    (await ds.put(key2, val2)).tryGet
+    (await ds.put(key3, val3)).tryGet
+
+    let
+      iter1 = (await ds.query(q1)).tryGet
+      iter2 = (await ds.query(q1)).tryGet
+
+    let one1 = (await iter1.next()).tryGet
+    check not iter1.finished
+
+    let one2 = (await iter2.next()).tryGet
+    check not iter2.finished
+
+    (await iter1.dispose()).tryGet
+    check iter1.finished
+
+    let two2 = (await iter2.next()).tryGet
+    check not iter2.finished
+    let three2 = (await iter2.next()).tryGet
+    check not iter2.finished
+
+    let four2 = (await iter2.next()).tryGet
+    check iter2.finished
+
+    check:
+      one1[0].get == key1
+      one1[1] == val1
+
+      one2[0].get == key1
+      one2[1] == val1
+      two2[0].get == key2
+      two2[1] == val2
+      three2[0].get == key3
+      three2[1] == val3
+      four2[0] == Key.none
+
+    (await iter1.dispose()).tryGet
+    (await iter2.dispose()).tryGet
+
+  test "Dispose should discontinue iteration":
+    let
+      q1 = Query.init(key1)
+
+    (await ds.put(key1, val1)).tryGet
+    (await ds.put(key2, val2)).tryGet
+    (await ds.put(key3, val3)).tryGet
+
+    let
+      iter = (await ds.query(q1)).tryGet
+
+    let one = (await iter.next()).tryGet
+    check not iter.finished
+
+    let two = (await iter.next()).tryGet
+    check not iter.finished
+
+    (await iter.dispose()).tryGet
+    check iter.finished
+
+    let three = (await iter.next()).tryGet
+    check iter.finished
+
+    let four = (await iter.next()).tryGet
+    check iter.finished
+
+    check:
+      one[0].get == key1
+      one[1] == val1
+      two[0].get == key2
+      two[1] == val2
+      three[0] == Key.none
+      four[0] == Key.none
+
   test "Key should query all keys without values":
     let
       q = Query.init(key1, value = false)