Merge branch 'master' into testapi-delete
This commit is contained in:
Ben Bierens
GitHub
commit
c8b9f6f4a8
|
|
@ -32,6 +32,7 @@ jobs:
|
|||
uses: actions/checkout@v4
|
||||
with:
|
||||
submodules: recursive
|
||||
ref: ${{ github.event.pull_request.head.sha }}
|
||||
|
||||
- name: Setup Nimbus Build System
|
||||
uses: ./.github/actions/nimbus-build-system
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ jobs:
|
|||
uses: actions/checkout@v4
|
||||
with:
|
||||
submodules: recursive
|
||||
ref: ${{ github.event.pull_request.head.sha }}
|
||||
|
||||
- name: Setup Nimbus Build System
|
||||
uses: ./.github/actions/nimbus-build-system
|
||||
|
|
|
|||
|
|
@ -120,7 +120,7 @@ proc getPendingBlocks(
|
|||
CatchableError,
|
||||
"Future for block id not found, tree cid: " & $manifest.treeCid & ", index: " & $index)
|
||||
|
||||
Iter.new(genNext, isFinished)
|
||||
AsyncIter[(?!bt.Block, int)].new(genNext, isFinished)
|
||||
|
||||
proc prepareEncodingData(
|
||||
self: Erasure,
|
||||
|
|
@ -440,8 +440,7 @@ proc decode*(
|
|||
if treeCid != encoded.originalTreeCid:
|
||||
return failure("Original tree root differs from the tree root computed out of recovered data")
|
||||
|
||||
let idxIter = Iter
|
||||
.fromItems(recoveredIndices)
|
||||
let idxIter = Iter[Natural].new(recoveredIndices)
|
||||
.filter((i: Natural) => i < tree.leavesCount)
|
||||
|
||||
if err =? (await self.store.putSomeProofs(tree, idxIter)).errorOption:
|
||||
|
|
|
|||
|
|
@ -39,22 +39,9 @@ func checkIteration(self: IndexingStrategy, iteration: int): void {.raises: [Ind
|
|||
IndexingError,
|
||||
"Indexing iteration can't be greater than or equal to iterations.")
|
||||
|
||||
proc getIter(first, last, step: int): Iter[int] =
|
||||
var
|
||||
finish = false
|
||||
cur = first
|
||||
|
||||
func get(): int =
|
||||
result = cur
|
||||
cur += step
|
||||
|
||||
if cur > last:
|
||||
finish = true
|
||||
|
||||
func isFinished(): bool =
|
||||
finish
|
||||
|
||||
Iter.new(get, isFinished)
|
||||
func getIter(first, last, step: int): Iter[int] =
|
||||
{.cast(noSideEffect).}:
|
||||
Iter[int].new(first, last, step)
|
||||
|
||||
func getLinearIndicies(
|
||||
self: IndexingStrategy,
|
||||
|
|
|
|||
|
|
@ -157,10 +157,8 @@ proc updateExpiry*(
|
|||
|
||||
try:
|
||||
let
|
||||
ensuringFutures = Iter
|
||||
.fromSlice(0..<manifest.blocksCount)
|
||||
.mapIt(
|
||||
self.networkStore.localStore.ensureExpiry( manifest.treeCid, it, expiry ))
|
||||
ensuringFutures = Iter[int].new(0..<manifest.blocksCount)
|
||||
.mapIt(self.networkStore.localStore.ensureExpiry( manifest.treeCid, it, expiry ))
|
||||
await allFuturesThrowing(ensuringFutures)
|
||||
except CancelledError as exc:
|
||||
raise exc
|
||||
|
|
@ -209,7 +207,7 @@ proc fetchBatched*(
|
|||
|
||||
trace "Fetching blocks in batches of", size = batchSize
|
||||
|
||||
let iter = Iter.fromSlice(0..<manifest.blocksCount)
|
||||
let iter = Iter[int].new(0..<manifest.blocksCount)
|
||||
self.fetchBatched(manifest.treeCid, iter, batchSize, onBatch)
|
||||
|
||||
proc streamSingleBlock(
|
||||
|
|
|
|||
|
|
@ -132,51 +132,35 @@ method listBlocks*(
|
|||
## Get the list of blocks in the BlockStore. This is an intensive operation
|
||||
##
|
||||
|
||||
var
|
||||
iter = AsyncIter[?Cid]()
|
||||
|
||||
let
|
||||
cids = self.cids()
|
||||
|
||||
proc next(): Future[?Cid] {.async.} =
|
||||
await idleAsync()
|
||||
proc isFinished(): bool =
|
||||
return finished(cids)
|
||||
|
||||
var cid: Cid
|
||||
while true:
|
||||
if iter.finished:
|
||||
return Cid.none
|
||||
proc genNext(): Future[Cid] {.async.} =
|
||||
cids()
|
||||
|
||||
cid = cids()
|
||||
let iter = await (AsyncIter[Cid].new(genNext, isFinished)
|
||||
.filter(
|
||||
proc (cid: Cid): Future[bool] {.async.} =
|
||||
without isManifest =? cid.isManifest, err:
|
||||
trace "Error checking if cid is a manifest", err = err.msg
|
||||
return false
|
||||
|
||||
if finished(cids):
|
||||
iter.finish
|
||||
return Cid.none
|
||||
case blockType:
|
||||
of BlockType.Both:
|
||||
return true
|
||||
of BlockType.Manifest:
|
||||
return isManifest
|
||||
of BlockType.Block:
|
||||
return not isManifest
|
||||
))
|
||||
|
||||
without isManifest =? cid.isManifest, err:
|
||||
trace "Error checking if cid is a manifest", err = err.msg
|
||||
return Cid.none
|
||||
|
||||
case blockType:
|
||||
of BlockType.Manifest:
|
||||
if not isManifest:
|
||||
trace "Cid is not manifest, skipping", cid
|
||||
continue
|
||||
|
||||
break
|
||||
of BlockType.Block:
|
||||
if isManifest:
|
||||
trace "Cid is a manifest, skipping", cid
|
||||
continue
|
||||
|
||||
break
|
||||
of BlockType.Both:
|
||||
break
|
||||
|
||||
return cid.some
|
||||
|
||||
iter.next = next
|
||||
|
||||
return success iter
|
||||
return success(map[Cid, ?Cid](iter,
|
||||
proc (cid: Cid): Future[?Cid] {.async.} =
|
||||
some(cid)
|
||||
))
|
||||
|
||||
func putBlockSync(self: CacheStore, blk: Block): bool =
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,66 @@
|
|||
import pkg/questionable
|
||||
import pkg/questionable/results
|
||||
import pkg/chronos
|
||||
import pkg/chronicles
|
||||
import pkg/datastore/typedds
|
||||
|
||||
import ../utils/asynciter
|
||||
|
||||
type KeyVal[T] = tuple[key: Key, value: T]
|
||||
|
||||
proc toAsyncIter*[T](
|
||||
queryIter: QueryIter[T],
|
||||
finishOnErr: bool = true
|
||||
): Future[?!AsyncIter[?!QueryResponse[T]]] {.async.} =
|
||||
## Converts `QueryIter[T]` to `AsyncIter[?!QueryResponse[T]]` and automatically
|
||||
## runs dispose whenever `QueryIter` finishes or whenever an error occurs (only
|
||||
## if the flag finishOnErr is set to true)
|
||||
##
|
||||
|
||||
if queryIter.finished:
|
||||
trace "Disposing iterator"
|
||||
if error =? (await queryIter.dispose()).errorOption:
|
||||
return failure(error)
|
||||
return success(AsyncIter[?!QueryResponse[T]].empty())
|
||||
|
||||
var errOccurred = false
|
||||
|
||||
proc genNext: Future[?!QueryResponse[T]] {.async.} =
|
||||
let queryResOrErr = await queryIter.next()
|
||||
|
||||
if queryResOrErr.isErr:
|
||||
errOccurred = true
|
||||
|
||||
if queryIter.finished or (errOccurred and finishOnErr):
|
||||
trace "Disposing iterator"
|
||||
if error =? (await queryIter.dispose()).errorOption:
|
||||
return failure(error)
|
||||
|
||||
return queryResOrErr
|
||||
|
||||
proc isFinished(): bool =
|
||||
queryIter.finished or (errOccurred and finishOnErr)
|
||||
|
||||
AsyncIter[?!QueryResponse[T]].new(genNext, isFinished).success
|
||||
|
||||
proc filterSuccess*[T](
|
||||
iter: AsyncIter[?!QueryResponse[T]]
|
||||
): Future[AsyncIter[tuple[key: Key, value: T]]] {.async.} =
|
||||
## Filters out any items that are not success
|
||||
|
||||
proc mapping(resOrErr: ?!QueryResponse[T]): Future[?KeyVal[T]] {.async.} =
|
||||
without res =? resOrErr, error:
|
||||
error "Error occurred when getting QueryResponse", msg = error.msg
|
||||
return KeyVal[T].none
|
||||
|
||||
without key =? res.key:
|
||||
warn "No key for a QueryResponse"
|
||||
return KeyVal[T].none
|
||||
|
||||
without value =? res.value, error:
|
||||
error "Error occurred when getting a value from QueryResponse", msg = error.msg
|
||||
return KeyVal[T].none
|
||||
|
||||
(key: key, value: value).some
|
||||
|
||||
await mapFilter[?!QueryResponse[T], KeyVal[T]](iter, mapping)
|
||||
|
|
@ -47,4 +47,4 @@ proc putSomeProofs*(store: BlockStore, tree: CodexTree, iter: Iter[Natural]): Fu
|
|||
store.putSomeProofs(tree, iter.map((i: Natural) => i.ord))
|
||||
|
||||
proc putAllProofs*(store: BlockStore, tree: CodexTree): Future[?!void] =
|
||||
store.putSomeProofs(tree, Iter.fromSlice(0..<tree.leavesCount))
|
||||
store.putSomeProofs(tree, Iter[int].new(0..<tree.leavesCount))
|
||||
|
|
|
|||
|
|
@ -3,26 +3,29 @@ import std/sugar
|
|||
import pkg/questionable
|
||||
import pkg/chronos
|
||||
|
||||
type
|
||||
Function*[T, U] = proc(fut: T): U {.raises: [CatchableError], gcsafe, noSideEffect.}
|
||||
IsFinished* = proc(): bool {.raises: [], gcsafe, noSideEffect.}
|
||||
GenNext*[T] = proc(): T {.raises: [CatchableError], gcsafe.}
|
||||
Iter*[T] = ref object
|
||||
finished: bool
|
||||
next*: GenNext[T]
|
||||
AsyncIter*[T] = Iter[Future[T]]
|
||||
import ./iter
|
||||
|
||||
proc finish*[T](self: Iter[T]): void =
|
||||
export iter
|
||||
|
||||
## AsyncIter[T] is similar to `Iter[Future[T]]` with addition of methods specific to asynchronous processing
|
||||
##
|
||||
|
||||
type
|
||||
AsyncIter*[T] = ref object
|
||||
finished: bool
|
||||
next*: GenNext[Future[T]]
|
||||
|
||||
proc finish*[T](self: AsyncIter[T]): void =
|
||||
self.finished = true
|
||||
|
||||
proc finished*[T](self: Iter[T]): bool =
|
||||
proc finished*[T](self: AsyncIter[T]): bool =
|
||||
self.finished
|
||||
|
||||
iterator items*[T](self: Iter[T]): T =
|
||||
iterator items*[T](self: AsyncIter[T]): Future[T] =
|
||||
while not self.finished:
|
||||
yield self.next()
|
||||
|
||||
iterator pairs*[T](self: Iter[T]): tuple[key: int, val: T] {.inline.} =
|
||||
iterator pairs*[T](self: AsyncIter[T]): tuple[key: int, val: Future[T]] {.inline.} =
|
||||
var i = 0
|
||||
while not self.finished:
|
||||
yield (i, self.next())
|
||||
|
|
@ -32,14 +35,25 @@ proc map*[T, U](fut: Future[T], fn: Function[T, U]): Future[U] {.async.} =
|
|||
let t = await fut
|
||||
fn(t)
|
||||
|
||||
proc new*[T](_: type Iter, genNext: GenNext[T], isFinished: IsFinished, finishOnErr: bool = true): Iter[T] =
|
||||
var iter = Iter[T]()
|
||||
proc flatMap*[T, U](fut: Future[T], fn: Function[T, Future[U]]): Future[U] {.async.} =
|
||||
let t = await fut
|
||||
await fn(t)
|
||||
|
||||
proc next(): T {.raises: [CatchableError].} =
|
||||
proc new*[T](_: type AsyncIter[T], genNext: GenNext[Future[T]], isFinished: IsFinished, finishOnErr: bool = true): AsyncIter[T] =
|
||||
## Creates a new Iter using elements returned by supplier function `genNext`.
|
||||
## Iter is finished whenever `isFinished` returns true.
|
||||
##
|
||||
|
||||
var iter = AsyncIter[T]()
|
||||
|
||||
proc next(): Future[T] {.async.} =
|
||||
if not iter.finished:
|
||||
var item: T
|
||||
try:
|
||||
item = genNext()
|
||||
item = await genNext()
|
||||
except CancelledError as err:
|
||||
iter.finish
|
||||
raise err
|
||||
except CatchableError as err:
|
||||
if finishOnErr or isFinished():
|
||||
iter.finish
|
||||
|
|
@ -49,7 +63,7 @@ proc new*[T](_: type Iter, genNext: GenNext[T], isFinished: IsFinished, finishOn
|
|||
iter.finish
|
||||
return item
|
||||
else:
|
||||
raise newException(CatchableError, "Iterator is finished but next item was requested")
|
||||
raise newException(CatchableError, "AsyncIter is finished but next item was requested")
|
||||
|
||||
if isFinished():
|
||||
iter.finish
|
||||
|
|
@ -57,90 +71,95 @@ proc new*[T](_: type Iter, genNext: GenNext[T], isFinished: IsFinished, finishOn
|
|||
iter.next = next
|
||||
return iter
|
||||
|
||||
proc fromItems*[T](_: type Iter, items: seq[T]): Iter[T] =
|
||||
## Create new iterator from items
|
||||
proc mapAsync*[T, U](iter: Iter[T], fn: Function[T, Future[U]]): AsyncIter[U] =
|
||||
AsyncIter[U].new(
|
||||
genNext = () => fn(iter.next()),
|
||||
isFinished = () => iter.finished()
|
||||
)
|
||||
|
||||
proc new*[U, V: Ordinal](_: type AsyncIter[U], slice: HSlice[U, V]): AsyncIter[U] =
|
||||
## Creates new Iter from a slice
|
||||
##
|
||||
|
||||
Iter.fromSlice(0..<items.len)
|
||||
.map((i: int) => items[i])
|
||||
let iter = Iter[U].new(slice)
|
||||
mapAsync[U, U](iter,
|
||||
proc (i: U): Future[U] {.async.} =
|
||||
i
|
||||
)
|
||||
|
||||
proc fromSlice*[U, V: Ordinal](_: type Iter, slice: HSlice[U, V]): Iter[U] =
|
||||
## Creates new iterator from slice
|
||||
proc new*[U, V, S: Ordinal](_: type AsyncIter[U], a: U, b: V, step: S = 1): AsyncIter[U] =
|
||||
## Creates new Iter in range a..b with specified step (default 1)
|
||||
##
|
||||
|
||||
Iter.fromRange(slice.a.int, slice.b.int, 1)
|
||||
let iter = Iter[U].new(a, b, step)
|
||||
mapAsync[U, U](iter,
|
||||
proc (i: U): Future[U] {.async.} =
|
||||
i
|
||||
)
|
||||
|
||||
proc fromRange*[U, V, S: Ordinal](_: type Iter, a: U, b: V, step: S = 1): Iter[U] =
|
||||
## Creates new iterator in range a..b with specified step (default 1)
|
||||
proc empty*[T](_: type AsyncIter[T]): AsyncIter[T] =
|
||||
## Creates an empty AsyncIter
|
||||
##
|
||||
|
||||
var i = a
|
||||
proc genNext(): Future[T] {.raises: [CatchableError].} =
|
||||
raise newException(CatchableError, "Next item requested from an empty AsyncIter")
|
||||
proc isFinished(): bool = true
|
||||
|
||||
proc genNext(): U =
|
||||
let u = i
|
||||
inc(i, step)
|
||||
u
|
||||
AsyncIter[T].new(genNext, isFinished)
|
||||
|
||||
proc isFinished(): bool =
|
||||
(step > 0 and i > b) or
|
||||
(step < 0 and i < b)
|
||||
|
||||
Iter.new(genNext, isFinished)
|
||||
|
||||
proc map*[T, U](iter: Iter[T], fn: Function[T, U]): Iter[U] =
|
||||
Iter.new(
|
||||
genNext = () => fn(iter.next()),
|
||||
proc map*[T, U](iter: AsyncIter[T], fn: Function[T, Future[U]]): AsyncIter[U] =
|
||||
AsyncIter[U].new(
|
||||
genNext = () => iter.next().flatMap(fn),
|
||||
isFinished = () => iter.finished
|
||||
)
|
||||
|
||||
proc filter*[T](iter: Iter[T], predicate: Function[T, bool]): Iter[T] =
|
||||
var nextT: Option[T]
|
||||
proc mapFilter*[T, U](iter: AsyncIter[T], mapPredicate: Function[T, Future[Option[U]]]): Future[AsyncIter[U]] {.async.} =
|
||||
var nextFutU: Option[Future[U]]
|
||||
|
||||
proc tryFetch(): void =
|
||||
nextT = T.none
|
||||
proc tryFetch(): Future[void] {.async.} =
|
||||
nextFutU = Future[U].none
|
||||
while not iter.finished:
|
||||
let t = iter.next()
|
||||
if predicate(t):
|
||||
nextT = some(t)
|
||||
let futT = iter.next()
|
||||
try:
|
||||
if u =? await futT.flatMap(mapPredicate):
|
||||
let futU = newFuture[U]("AsyncIter.mapFilterAsync")
|
||||
futU.complete(u)
|
||||
nextFutU = some(futU)
|
||||
break
|
||||
except CancelledError as err:
|
||||
raise err
|
||||
except CatchableError as err:
|
||||
let errFut = newFuture[U]("AsyncIter.mapFilterAsync")
|
||||
errFut.fail(err)
|
||||
nextFutU = some(errFut)
|
||||
break
|
||||
|
||||
proc genNext(): T =
|
||||
let t = nextT.unsafeGet
|
||||
tryFetch()
|
||||
return t
|
||||
proc genNext(): Future[U] {.async.} =
|
||||
let futU = nextFutU.unsafeGet
|
||||
await tryFetch()
|
||||
await futU
|
||||
|
||||
proc isFinished(): bool =
|
||||
nextT.isNone
|
||||
nextFutU.isNone
|
||||
|
||||
tryFetch()
|
||||
Iter.new(genNext, isFinished)
|
||||
await tryFetch()
|
||||
AsyncIter[U].new(genNext, isFinished)
|
||||
|
||||
proc prefetch*[T](iter: Iter[T], n: Positive): Iter[T] =
|
||||
var ringBuf = newSeq[T](n)
|
||||
var iterLen = int.high
|
||||
var i = 0
|
||||
|
||||
proc tryFetch(j: int): void =
|
||||
if not iter.finished:
|
||||
let item = iter.next()
|
||||
ringBuf[j mod n] = item
|
||||
if iter.finished:
|
||||
iterLen = min(j + 1, iterLen)
|
||||
proc filter*[T](iter: AsyncIter[T], predicate: Function[T, Future[bool]]): Future[AsyncIter[T]] {.async.} =
|
||||
proc wrappedPredicate(t: T): Future[Option[T]] {.async.} =
|
||||
if await predicate(t):
|
||||
some(t)
|
||||
else:
|
||||
if j == 0:
|
||||
iterLen = 0
|
||||
T.none
|
||||
|
||||
proc genNext(): T =
|
||||
let item = ringBuf[i mod n]
|
||||
tryFetch(i + n)
|
||||
inc i
|
||||
return item
|
||||
await mapFilter[T, T](iter, wrappedPredicate)
|
||||
|
||||
proc isFinished(): bool =
|
||||
i >= iterLen
|
||||
proc delayBy*[T](iter: AsyncIter[T], d: Duration): AsyncIter[T] =
|
||||
## Delays emitting each item by given duration
|
||||
##
|
||||
|
||||
# initialize ringBuf with n prefetched values
|
||||
for j in 0..<n:
|
||||
tryFetch(j)
|
||||
|
||||
Iter.new(genNext, isFinished)
|
||||
map[T, T](iter,
|
||||
proc (t: T): Future[T] {.async.} =
|
||||
await sleepAsync(d)
|
||||
t
|
||||
)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,140 @@
|
|||
import std/sugar
|
||||
|
||||
import pkg/questionable
|
||||
import pkg/questionable/results
|
||||
|
||||
type
|
||||
Function*[T, U] = proc(fut: T): U {.raises: [CatchableError], gcsafe, closure.}
|
||||
IsFinished* = proc(): bool {.raises: [], gcsafe, closure.}
|
||||
GenNext*[T] = proc(): T {.raises: [CatchableError], gcsafe.}
|
||||
Iter*[T] = ref object
|
||||
finished: bool
|
||||
next*: GenNext[T]
|
||||
|
||||
proc finish*[T](self: Iter[T]): void =
|
||||
self.finished = true
|
||||
|
||||
proc finished*[T](self: Iter[T]): bool =
|
||||
self.finished
|
||||
|
||||
iterator items*[T](self: Iter[T]): T =
|
||||
while not self.finished:
|
||||
yield self.next()
|
||||
|
||||
iterator pairs*[T](self: Iter[T]): tuple[key: int, val: T] {.inline.} =
|
||||
var i = 0
|
||||
while not self.finished:
|
||||
yield (i, self.next())
|
||||
inc(i)
|
||||
|
||||
proc new*[T](_: type Iter[T], genNext: GenNext[T], isFinished: IsFinished, finishOnErr: bool = true): Iter[T] =
|
||||
## Creates a new Iter using elements returned by supplier function `genNext`.
|
||||
## Iter is finished whenever `isFinished` returns true.
|
||||
##
|
||||
|
||||
var iter = Iter[T]()
|
||||
|
||||
proc next(): T {.raises: [CatchableError].} =
|
||||
if not iter.finished:
|
||||
var item: T
|
||||
try:
|
||||
item = genNext()
|
||||
except CatchableError as err:
|
||||
if finishOnErr or isFinished():
|
||||
iter.finish
|
||||
raise err
|
||||
|
||||
if isFinished():
|
||||
iter.finish
|
||||
return item
|
||||
else:
|
||||
raise newException(CatchableError, "Iter is finished but next item was requested")
|
||||
|
||||
if isFinished():
|
||||
iter.finish
|
||||
|
||||
iter.next = next
|
||||
return iter
|
||||
|
||||
proc new*[U, V, S: Ordinal](_: type Iter[U], a: U, b: V, step: S = 1): Iter[U] =
|
||||
## Creates a new Iter in range a..b with specified step (default 1)
|
||||
##
|
||||
|
||||
var i = a
|
||||
|
||||
proc genNext(): U =
|
||||
let u = i
|
||||
inc(i, step)
|
||||
u
|
||||
|
||||
proc isFinished(): bool =
|
||||
(step > 0 and i > b) or
|
||||
(step < 0 and i < b)
|
||||
|
||||
Iter[U].new(genNext, isFinished)
|
||||
|
||||
proc new*[U, V: Ordinal](_: type Iter[U], slice: HSlice[U, V]): Iter[U] =
|
||||
## Creates a new Iter from a slice
|
||||
##
|
||||
|
||||
Iter[U].new(slice.a.int, slice.b.int, 1)
|
||||
|
||||
proc new*[T](_: type Iter[T], items: seq[T]): Iter[T] =
|
||||
## Creates a new Iter from a sequence
|
||||
##
|
||||
|
||||
Iter[int].new(0..<items.len)
|
||||
.map((i: int) => items[i])
|
||||
|
||||
proc empty*[T](_: type Iter[T]): Iter[T] =
|
||||
## Creates an empty Iter
|
||||
##
|
||||
|
||||
proc genNext(): T {.raises: [CatchableError].} =
|
||||
raise newException(CatchableError, "Next item requested from an empty Iter")
|
||||
proc isFinished(): bool = true
|
||||
|
||||
Iter[T].new(genNext, isFinished)
|
||||
|
||||
proc map*[T, U](iter: Iter[T], fn: Function[T, U]): Iter[U] =
|
||||
Iter[U].new(
|
||||
genNext = () => fn(iter.next()),
|
||||
isFinished = () => iter.finished
|
||||
)
|
||||
|
||||
proc mapFilter*[T, U](iter: Iter[T], mapPredicate: Function[T, Option[U]]): Iter[U] =
|
||||
var nextUOrErr: Option[Result[U, ref CatchableError]]
|
||||
|
||||
proc tryFetch(): void =
|
||||
nextUOrErr = Result[U, ref CatchableError].none
|
||||
while not iter.finished:
|
||||
try:
|
||||
let t = iter.next()
|
||||
if u =? mapPredicate(t):
|
||||
nextUOrErr = some(success(u))
|
||||
break
|
||||
except CatchableError as err:
|
||||
nextUOrErr = some(U.failure(err))
|
||||
|
||||
proc genNext(): U {.raises: [CatchableError].} =
|
||||
# at this point nextUOrErr should always be some(..)
|
||||
without u =? nextUOrErr.unsafeGet, err:
|
||||
raise err
|
||||
|
||||
tryFetch()
|
||||
return u
|
||||
|
||||
proc isFinished(): bool =
|
||||
nextUOrErr.isNone
|
||||
|
||||
tryFetch()
|
||||
Iter[U].new(genNext, isFinished)
|
||||
|
||||
proc filter*[T](iter: Iter[T], predicate: Function[T, bool]): Iter[T] =
|
||||
proc wrappedPredicate(t: T): Option[T] =
|
||||
if predicate(t):
|
||||
some(t)
|
||||
else:
|
||||
T.none
|
||||
|
||||
mapFilter[T, T](iter, wrappedPredicate)
|
||||
10
openapi.yaml
10
openapi.yaml
|
|
@ -289,7 +289,7 @@ components:
|
|||
description: "Root hash of the content"
|
||||
originalBytes:
|
||||
type: integer
|
||||
format: uint64
|
||||
format: int64
|
||||
description: "Length of original content in bytes"
|
||||
blockSize:
|
||||
type: integer
|
||||
|
|
@ -304,18 +304,18 @@ components:
|
|||
totalBlocks:
|
||||
description: "Number of blocks stored by the node"
|
||||
type: integer
|
||||
format: uint64
|
||||
format: int64
|
||||
quotaMaxBytes:
|
||||
type: integer
|
||||
format: uint64
|
||||
format: int64
|
||||
description: "Maximum storage space used by the node"
|
||||
quotaUsedBytes:
|
||||
type: integer
|
||||
format: uint64
|
||||
format: int64
|
||||
description: "Amount of storage space currently in use"
|
||||
quotaReservedBytes:
|
||||
type: integer
|
||||
format: uint64
|
||||
format: int64
|
||||
description: "Amount of storage space reserved"
|
||||
|
||||
servers:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,65 @@
|
|||
import std/sugar
|
||||
|
||||
import pkg/stew/results
|
||||
import pkg/questionable
|
||||
import pkg/chronos
|
||||
import pkg/datastore/typedds
|
||||
import pkg/datastore/sql/sqliteds
|
||||
import pkg/codex/stores/queryiterhelper
|
||||
import pkg/codex/utils/asynciter
|
||||
|
||||
import ../../asynctest
|
||||
import ../helpers
|
||||
|
||||
proc encode(s: string): seq[byte] =
|
||||
s.toBytes()
|
||||
|
||||
proc decode(T: type string, bytes: seq[byte]): ?!T =
|
||||
success(string.fromBytes(bytes))
|
||||
|
||||
asyncchecksuite "Test QueryIter helper":
|
||||
var
|
||||
tds: TypedDatastore
|
||||
|
||||
setupAll:
|
||||
tds = TypedDatastore.init(SQLiteDatastore.new(Memory).tryGet())
|
||||
|
||||
teardownAll:
|
||||
(await tds.close()).tryGet
|
||||
|
||||
test "Should auto-dispose when QueryIter finishes":
|
||||
let
|
||||
source = {
|
||||
"a": "11",
|
||||
"b": "22"
|
||||
}.toTable
|
||||
Root = Key.init("/queryitertest").tryGet()
|
||||
|
||||
for k, v in source:
|
||||
let key = (Root / k).tryGet()
|
||||
(await tds.put(key, v)).tryGet()
|
||||
|
||||
var
|
||||
disposed = false
|
||||
queryIter = (await query[string](tds, Query.init(Root))).tryGet()
|
||||
|
||||
let iterDispose: IterDispose = queryIter.dispose
|
||||
queryIter.dispose = () => (disposed = true; iterDispose())
|
||||
|
||||
let
|
||||
iter1 = (await toAsyncIter[string](queryIter)).tryGet()
|
||||
iter2 = await filterSuccess[string](iter1)
|
||||
|
||||
var items = initTable[string, string]()
|
||||
|
||||
for fut in iter2:
|
||||
let item = await fut
|
||||
|
||||
items[item.key.value] = item.value
|
||||
|
||||
check:
|
||||
items == source
|
||||
disposed == true
|
||||
queryIter.finished == true
|
||||
iter1.finished == true
|
||||
iter2.finished == true
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
import ./stores/testcachestore
|
||||
import ./stores/testrepostore
|
||||
import ./stores/testmaintenance
|
||||
import ./stores/testqueryiterhelper
|
||||
|
||||
{.warning[UnusedImport]: off.}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import ./utils/testoptions
|
||||
import ./utils/testkeyutils
|
||||
import ./utils/testasyncstatemachine
|
||||
import ./utils/testasynciter
|
||||
import ./utils/testtimer
|
||||
import ./utils/testthen
|
||||
import ./utils/testtrackedfutures
|
||||
|
|
|
|||
|
|
@ -0,0 +1,160 @@
|
|||
import std/sugar
|
||||
|
||||
import pkg/questionable
|
||||
import pkg/chronos
|
||||
import pkg/codex/utils/asynciter
|
||||
|
||||
import ../../asynctest
|
||||
import ../helpers
|
||||
|
||||
asyncchecksuite "Test AsyncIter":
|
||||
|
||||
test "Should be finished":
|
||||
let iter = AsyncIter[int].empty()
|
||||
|
||||
check:
|
||||
iter.finished == true
|
||||
|
||||
test "Should map each item using `map`":
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = map[int, string](iter1,
|
||||
proc (i: int): Future[string] {.async.} =
|
||||
$i
|
||||
)
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @["0", "1", "2", "3", "4"]
|
||||
|
||||
test "Should leave only odd items using `filter`":
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = await filter[int](iter1,
|
||||
proc (i: int): Future[bool] {.async.} =
|
||||
(i mod 2) == 1
|
||||
)
|
||||
|
||||
var collected: seq[int]
|
||||
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @[1, 3]
|
||||
|
||||
test "Should leave only odd items using `mapFilter`":
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = await mapFilter[int, string](iter1,
|
||||
proc (i: int): Future[?string] {.async.} =
|
||||
if (i mod 2) == 1:
|
||||
some($i)
|
||||
else:
|
||||
string.none
|
||||
)
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @["1", "3"]
|
||||
|
||||
test "Should yield all items before err using `map`":
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = map[int, string](iter1,
|
||||
proc (i: int): Future[string] {.async.} =
|
||||
if i < 3:
|
||||
return $i
|
||||
else:
|
||||
raise newException(CatchableError, "Some error")
|
||||
)
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
expect CatchableError:
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @["0", "1", "2"]
|
||||
iter2.finished
|
||||
|
||||
test "Should yield all items before err using `filter`":
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = await filter[int](iter1,
|
||||
proc (i: int): Future[bool] {.async.} =
|
||||
if i < 3:
|
||||
return true
|
||||
else:
|
||||
raise newException(CatchableError, "Some error")
|
||||
)
|
||||
|
||||
var collected: seq[int]
|
||||
|
||||
expect CatchableError:
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @[0, 1, 2]
|
||||
iter2.finished
|
||||
|
||||
test "Should yield all items before err using `mapFilter`":
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = await mapFilter[int, string](iter1,
|
||||
proc (i: int): Future[?string] {.async.} =
|
||||
if i < 3:
|
||||
return some($i)
|
||||
else:
|
||||
raise newException(CatchableError, "Some error")
|
||||
)
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
expect CatchableError:
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @["0", "1", "2"]
|
||||
iter2.finished
|
||||
|
||||
test "Should propagate cancellation error immediately":
|
||||
let
|
||||
fut = newFuture[?string]("testasynciter")
|
||||
|
||||
let
|
||||
iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
|
||||
iter2 = await mapFilter[int, string](iter1,
|
||||
proc (i: int): Future[?string] {.async.} =
|
||||
if i < 3:
|
||||
return some($i)
|
||||
else:
|
||||
return await fut
|
||||
)
|
||||
|
||||
proc cancelFut(): Future[void] {.async.} =
|
||||
await sleepAsync(100.millis)
|
||||
await fut.cancelAndWait()
|
||||
|
||||
asyncSpawn(cancelFut())
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
expect CancelledError:
|
||||
for fut in iter2:
|
||||
collected.add(await fut)
|
||||
|
||||
check:
|
||||
collected == @["0", "1"]
|
||||
iter2.finished
|
||||
|
|
@ -0,0 +1,129 @@
|
|||
import std/sugar
|
||||
|
||||
import pkg/questionable
|
||||
import pkg/chronos
|
||||
import pkg/codex/utils/iter
|
||||
|
||||
import ../../asynctest
|
||||
import ../helpers
|
||||
|
||||
checksuite "Test Iter":
|
||||
|
||||
test "Should be finished":
|
||||
let iter = Iter[int].empty()
|
||||
|
||||
check:
|
||||
iter.finished == true
|
||||
|
||||
test "Should be iterable with `items`":
|
||||
let iter = Iter.new(0..<5)
|
||||
|
||||
let items =
|
||||
collect:
|
||||
for v in iter:
|
||||
v
|
||||
|
||||
check:
|
||||
items == @[0, 1, 2, 3, 4]
|
||||
|
||||
test "Should be iterable with `pairs`":
|
||||
let iter = Iter.new(0..<5)
|
||||
|
||||
let pairs =
|
||||
collect:
|
||||
for i, v in iter:
|
||||
(i, v)
|
||||
|
||||
check:
|
||||
pairs == @[(0, 0), (1, 1), (2, 2), (3, 3), (4, 4)]
|
||||
|
||||
test "Should map each item using `map`":
|
||||
let iter = Iter.new(0..<5)
|
||||
.map((i: int) => $i)
|
||||
|
||||
check:
|
||||
iter.toSeq() == @["0", "1", "2", "3", "4"]
|
||||
|
||||
test "Should leave only odd items using `filter`":
|
||||
let iter = Iter.new(0..<5)
|
||||
.filter((i: int) => (i mod 2) == 1)
|
||||
|
||||
check:
|
||||
iter.toSeq() == @[1, 3]
|
||||
|
||||
test "Should leave only odd items using `mapFilter`":
|
||||
let
|
||||
iter1 = Iter.new(0..<5)
|
||||
iter2 = mapFilter[int, string](iter1,
|
||||
proc(i: int): ?string =
|
||||
if (i mod 2) == 1:
|
||||
some($i)
|
||||
else:
|
||||
string.none
|
||||
)
|
||||
|
||||
check:
|
||||
iter2.toSeq() == @["1", "3"]
|
||||
|
||||
test "Should yield all items before err using `map`":
|
||||
let
|
||||
iter = Iter.new(0..<5)
|
||||
.map(
|
||||
proc (i: int): string =
|
||||
if i < 3:
|
||||
return $i
|
||||
else:
|
||||
raise newException(CatchableError, "Some error")
|
||||
)
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
expect CatchableError:
|
||||
for i in iter:
|
||||
collected.add(i)
|
||||
|
||||
check:
|
||||
collected == @["0", "1", "2"]
|
||||
iter.finished
|
||||
|
||||
test "Should yield all items before err using `filter`":
|
||||
let
|
||||
iter = Iter.new(0..<5)
|
||||
.filter(
|
||||
proc (i: int): bool =
|
||||
if i < 3:
|
||||
return true
|
||||
else:
|
||||
raise newException(CatchableError, "Some error")
|
||||
)
|
||||
|
||||
var collected: seq[int]
|
||||
|
||||
expect CatchableError:
|
||||
for i in iter:
|
||||
collected.add(i)
|
||||
|
||||
check:
|
||||
collected == @[0, 1, 2]
|
||||
iter.finished
|
||||
|
||||
test "Should yield all items before err using `mapFilter`":
|
||||
let
|
||||
iter1 = Iter.new(0..<5)
|
||||
iter2 = mapFilter[int, string](iter1,
|
||||
proc (i: int): ?string =
|
||||
if i < 3:
|
||||
return some($i)
|
||||
else:
|
||||
raise newException(CatchableError, "Some error")
|
||||
)
|
||||
|
||||
var collected: seq[string]
|
||||
|
||||
expect CatchableError:
|
||||
for i in iter2:
|
||||
collected.add(i)
|
||||
|
||||
check:
|
||||
collected == @["0", "1", "2"]
|
||||
iter2.finished
|
||||
|
|
@ -1 +1 @@
|
|||
Subproject commit f4989fcce5d74a648e7e2598a72a7b21948f4a85
|
||||
Subproject commit 3ab6b84a634a7b2ee8c0144f050bf5893cd47c17
|
||||
Loading…
Reference in New Issue