logos-storage/logos-storage-nim
1
0
Fork 0
mirror of https://github.com/logos-storage/logos-storage-nim.git synced 2026-01-05 23:13:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

renaming: AsyncResultIterator => AsyncResultIter to have it consistent with other iterators

Browse Source
This commit is contained in:
Marcin Czenko 2025-06-24 15:24:55 +02:00
parent c5c1f979c0
commit 28ccfb1744
No known key found for this signature in database
GPG Key ID: 33DEA0C8E30937C0
16 changed files with 113 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
codex/erasure/erasure.nim
View File

@ -121,13 +121,13 @@ func indexToPos(steps, idx, step: int): int {.inline.} =
proc getPendingBlocks(
self: Erasure, manifest: Manifest, indices: seq[int]
): AsyncResultIterator[(?!bt.Block, int)] =
): AsyncResultIter[(?!bt.Block, int)] =
## Get pending blocks iterator
##
if indices.len == 0:
trace "No indices to fetch blocks for", treeCid = manifest.treeCid
return AsyncResultIterator[(?!bt.Block, int)].empty()
return AsyncResultIter[(?!bt.Block, int)].empty()
var pendingBlocks: seq[Future[(?!bt.Block, int)].Raising([CancelledError])] = @[]
@ -162,8 +162,8 @@ proc getPendingBlocks(
# but we check for that at the very beginning -
# thus, if this happens, we raise an assert
raiseAssert("fatal: pendingBlocks is empty - this should never happen")
AsyncResultIterator[(?!bt.Block, int)].new(genNext, isFinished)
AsyncResultIter[(?!bt.Block, int)].new(genNext, isFinished)
proc prepareEncodingData(
self: Erasure,

4
codex/stores/blockstore.nim
View File

@ -154,9 +154,7 @@ method hasBlock*(
method listBlocks*(
self: BlockStore, blockType = BlockType.Manifest
): Future[?!AsyncResultIterator[Cid]] {.
base, async: (raises: [CancelledError]), gcsafe
.} =
): Future[?!AsyncResultIter[Cid]] {.base, async: (raises: [CancelledError]), gcsafe.} =
## Get the list of blocks in the BlockStore. This is an intensive operation
##

4
codex/stores/cachestore.nim
View File

@ -139,7 +139,7 @@ func cids(self: CacheStore): (iterator (): Cid {.gcsafe, raises: [].}) =
method listBlocks*(
self: CacheStore, blockType = BlockType.Manifest
): Future[?!AsyncResultIterator[Cid]] {.async: (raises: [CancelledError]).} =
): Future[?!AsyncResultIter[Cid]] {.async: (raises: [CancelledError]).} =
## Get the list of blocks in the BlockStore. This is an intensive operation
##
@ -152,7 +152,7 @@ method listBlocks*(
success(cids())
let iter = await (
AsyncResultIterator[Cid].new(genNext, isFinished).filter(
AsyncResultIter[Cid].new(genNext, isFinished).filter(
proc(cid: ?!Cid): Future[bool] {.async: (raises: [CancelledError]).} =
without cid =? cid, err:
trace "Cannot get Cid from the iterator", err = err.msg

2
codex/stores/maintenance.nim
View File

@ -18,7 +18,7 @@ import pkg/questionable/results
import ./repostore
import ../utils/timer
import ../utils/asyncresultiterator
import ../utils/asyncresultiter
import ../clock
import ../logutils
import ../systemclock

4
codex/stores/networkstore.nim
View File

@ -19,7 +19,7 @@ import ../blockexchange
import ../logutils
import ../merkletree
import ../utils/asyncheapqueue
import ../utils/asyncresultiterator
import ../utils/asyncresultiter
import ./blockstore
export blockstore, blockexchange, asyncheapqueue
@ -127,7 +127,7 @@ method ensureExpiry*(
method listBlocks*(
self: NetworkStore, blockType = BlockType.Manifest
): Future[?!AsyncResultIterator[Cid]] {.async: (raw: true, raises: [CancelledError]).} =
): Future[?!AsyncResultIter[Cid]] {.async: (raw: true, raises: [CancelledError]).} =
self.localStore.listBlocks(blockType)
method delBlock*(

16
codex/stores/queryiterhelper.nim
View File

@ -4,16 +4,16 @@ import pkg/chronos
import pkg/chronicles
import pkg/datastore/typedds
import ../utils/asyncresultiterator
import ../utils/asyncresultiter
{.push raises: [].}
type KeyVal*[T] = tuple[key: Key, value: T]
proc toAsyncResultIterator*[T](
proc toAsyncResultIter*[T](
queryIter: QueryIter[T], finishOnErr: bool = true
): Future[?!AsyncResultIterator[QueryResponse[T]]] {.async: (raises: [CancelledError]).} =
## Converts `QueryIter[T]` to `AsyncResultIterator[QueryResponse[T]]` and automatically
): Future[?!AsyncResultIter[QueryResponse[T]]] {.async: (raises: [CancelledError]).} =
## Converts `QueryIter[T]` to `AsyncResultIter[QueryResponse[T]]` and automatically
## runs dispose whenever `QueryIter` finishes or whenever an error occurs (only
## if the flag finishOnErr is set to true)
##
@ -22,7 +22,7 @@ proc toAsyncResultIterator*[T](
trace "Disposing iterator"
if error =? (await queryIter.dispose()).errorOption:
return failure(error)
return success(AsyncResultIterator[QueryResponse[T]].empty())
return success(AsyncResultIter[QueryResponse[T]].empty())
var errOccurred = false
@ -42,11 +42,11 @@ proc toAsyncResultIterator*[T](
proc isFinished(): bool =
queryIter.finished
AsyncResultIterator[QueryResponse[T]].new(genNext, isFinished).success
AsyncResultIter[QueryResponse[T]].new(genNext, isFinished).success
proc filterSuccess*[T](
iter: AsyncResultIterator[QueryResponse[T]]
): Future[AsyncResultIterator[tuple[key: Key, value: T]]] {.
iter: AsyncResultIter[QueryResponse[T]]
): Future[AsyncResultIter[tuple[key: Key, value: T]]] {.
async: (raises: [CancelledError])
.} =
## Filters out any items that are not success

10
codex/stores/repostore/store.nim
View File

@ -295,12 +295,12 @@ method hasBlock*(
method listBlocks*(
self: RepoStore, blockType = BlockType.Manifest
): Future[?!AsyncResultIterator[Cid]] {.async: (raises: [CancelledError]).} =
): Future[?!AsyncResultIter[Cid]] {.async: (raises: [CancelledError]).} =
## Get the list of blocks in the RepoStore.
## This is an intensive operation
##
var iter = AsyncResultIterator[Cid]()
var iter = AsyncResultIter[Cid]()
let key =
case blockType
@ -346,7 +346,7 @@ proc blockRefCount*(self: RepoStore, cid: Cid): Future[?!Natural] {.async.} =
method getBlockExpirations*(
self: RepoStore, maxNumber: int, offset: int
): Future[?!AsyncResultIterator[BlockExpiration]] {.
): Future[?!AsyncResultIter[BlockExpiration]] {.
async: (raises: [CancelledError]), base, gcsafe
.} =
## Get iterator with block expirations
@ -360,11 +360,11 @@ method getBlockExpirations*(
error "Unable to execute block expirations query", err = err.msg
return failure(err)
without asyncQueryIter =? (await queryIter.toAsyncResultIterator()), err:
without asyncQueryIter =? (await queryIter.toAsyncResultIter()), err:
error "Unable to convert QueryIter to AsyncIter", err = err.msg
return failure(err)
let filteredIter: AsyncResultIterator[KeyVal[BlockMetadata]] =
let filteredIter: AsyncResultIter[KeyVal[BlockMetadata]] =
await asyncQueryIter.filterSuccess()
proc mapping(

4
codex/utils.nim
View File

@ -19,9 +19,9 @@ import pkg/chronos
import ./utils/asyncheapqueue
import ./utils/fileutils
import ./utils/iter
import ./utils/asyncresultiterator
import ./utils/asyncresultiter
export asyncheapqueue, fileutils, iter, asyncresultiterator, chronos
export asyncheapqueue, fileutils, iter, asyncresultiter, chronos
when defined(posix):
import os, posix

8
codex/utils/asynciter.nim
View File

@ -40,11 +40,9 @@ import ./iter
## - empty - to create an empty async iterator (AsyncIter)
type
AsyncIterFunc[T, U] =
proc(fut: T): Future[U] {.async.}
AsyncIterFunc[T, U] = proc(fut: T): Future[U] {.async.}
AsyncIterIsFinished = proc(): bool {.raises: [], gcsafe.}
AsyncIterGenNext[T] =
proc(): Future[T] {.async.}
AsyncIterGenNext[T] = proc(): Future[T] {.async.}
AsyncIter*[T] = ref object
finished: bool
@ -218,4 +216,4 @@ proc empty*[T](_: type AsyncIter[T]): AsyncIter[T] =
proc isFinished(): bool =
true
AsyncIter[T].new(genNext, isFinished)
AsyncIter[T].new(genNext, isFinished)

111
codex/utils/asyncresultiterator.nim → codex/utils/asyncresultiter.nim
View File

@ -17,11 +17,11 @@ import pkg/chronos
import ./iter
## AsyncResultIterator[T] is similar to `AsyncIterator[Future[T]]`
## AsyncResultIter[T] is similar to `AsyncIterator[Future[T]]`
## but does not throw exceptions others than CancelledError.
##
## Instead of throwing exception, it uses Result to communicate errors (
## thus the name AsyncResultIterator).
## thus the name AsyncResultIter).
##
## Public interface:
##
@ -29,58 +29,57 @@ import ./iter
## - next - allows to set a custom function to be called when the next item is requested
##
## Operations:
## - new - to create a new async iterator (AsyncResultIterator)
## - new - to create a new async iterator (AsyncResultIter)
## - finish - to finish the async iterator
## - finished - to check if the async iterator is finished
## - next - to get the next item from the async iterator
## - items - to iterate over the async iterator
## - pairs - to iterate over the async iterator and return the index of each item
## - mapFuture - to convert a (raising) Future[T] to a (raising) Future[U] using a function fn: auto -> Future[U] - we use auto to handle both raising and non-raising futures
## - mapAsync - to convert a regular sync iterator (Iter) to an async iter (AsyncResultIterator)
## - map - to convert one async iterator (AsyncResultIterator) to another async iter (AsyncResultIterator)
## - mapFilter - to convert one async iterator (AsyncResultIterator) to another async iter (AsyncResultIterator) and apply filtering at the same time
## - filter - to filter an async iterator (AsyncResultIterator) returning another async iterator (AsyncResultIterator)
## - delayBy - to delay each item returned by async iter by a given duration
## - empty - to create an empty async iterator (AsyncResultIterator)
## - mapAsync - to convert a regular sync iterator (Iter) to an async iterator (AsyncResultIter)
## - map - to convert one async iterator (AsyncResultIter) to another async iterator (AsyncResultIter)
## - mapFilter - to convert one async iterator (AsyncResultIter) to another async iterator (AsyncResultIter) and apply filtering at the same time
## - filter - to filter an async iterator (AsyncResultIter) and return another async iterator (AsyncResultIter)
## - delayBy - to delay each item returned by async iterator by a given duration
## - empty - to create an empty async iterator (AsyncResultIter)
type
AsyncResultIteratorFunc[T, U] =
proc(fut: T): Future[U] {.async: (raises: [CancelledError]), gcsafe, closure.}
AsyncResultIteratorIsFinished = proc(): bool {.raises: [], gcsafe, closure.}
AsyncResultIteratorGenNext[T] =
proc(): Future[T] {.async: (raises: [CancelledError]), gcsafe.}
AsyncResultIterFunc[T, U] =
proc(fut: T): Future[U] {.async: (raises: [CancelledError]).}
AsyncResultIterIsFinished = proc(): bool {.raises: [], gcsafe.}
AsyncResultIterGenNext[T] = proc(): Future[T] {.async: (raises: [CancelledError]).}
AsyncResultIterator*[T] = ref object
AsyncResultIter*[T] = ref object
finished: bool
next*: AsyncResultIteratorGenNext[?!T]
next*: AsyncResultIterGenNext[?!T]
proc flatMap[T, U](
fut: auto, fn: AsyncResultIteratorFunc[?!T, ?!U]
fut: auto, fn: AsyncResultIterFunc[?!T, ?!U]
): Future[?!U] {.async: (raises: [CancelledError]).} =
let t = await fut
await fn(t)
proc flatMap[T, U](
fut: auto, fn: AsyncResultIteratorFunc[?!T, Option[?!U]]
fut: auto, fn: AsyncResultIterFunc[?!T, Option[?!U]]
): Future[Option[?!U]] {.async: (raises: [CancelledError]).} =
let t = await fut
await fn(t)
########################################################################
## AsyncResultIterator public interface methods
## AsyncResultIter public interface methods
########################################################################
proc new*[T](
_: type AsyncResultIterator[T],
genNext: AsyncResultIteratorGenNext[?!T],
isFinished: IsFinished,
_: type AsyncResultIter[T],
genNext: AsyncResultIterGenNext[?!T],
isFinished: AsyncResultIterIsFinished,
finishOnErr: bool = true,
): AsyncResultIterator[T] =
): AsyncResultIter[T] =
## Creates a new Iter using elements returned by supplier function `genNext`.
## Iter is finished whenever `isFinished` returns true.
##
var iter = AsyncResultIterator[T]()
var iter = AsyncResultIter[T]()
proc next(): Future[?!T] {.async: (raises: [CancelledError]).} =
try:
@ -93,7 +92,7 @@ proc new*[T](
iter.finished = true
return item
else:
return failure("AsyncResultIterator is finished but next item was requested")
return failure("AsyncResultIter is finished but next item was requested")
except CancelledError as err:
iter.finished = true
raise err
@ -106,12 +105,12 @@ proc new*[T](
# forward declaration
proc mapAsync*[T, U](
iter: Iter[T], fn: AsyncResultIteratorFunc[T, ?!U], finishOnErr: bool = true
): AsyncResultIterator[U]
iter: Iter[T], fn: AsyncResultIterFunc[T, ?!U], finishOnErr: bool = true
): AsyncResultIter[U]
proc new*[U, V: Ordinal](
_: type AsyncResultIterator[U], slice: HSlice[U, V], finishOnErr: bool = true
): AsyncResultIterator[U] =
_: type AsyncResultIter[U], slice: HSlice[U, V], finishOnErr: bool = true
): AsyncResultIter[U] =
## Creates new Iter from a slice
##
@ -124,8 +123,8 @@ proc new*[U, V: Ordinal](
)
proc new*[U, V, S: Ordinal](
_: type AsyncResultIterator[U], a: U, b: V, step: S = 1, finishOnErr: bool = true
): AsyncResultIterator[U] =
_: type AsyncResultIter[U], a: U, b: V, step: S = 1, finishOnErr: bool = true
): AsyncResultIter[U] =
## Creates new Iter in range a..b with specified step (default 1)
##
@ -137,53 +136,53 @@ proc new*[U, V, S: Ordinal](
finishOnErr = finishOnErr,
)
proc finish*[T](self: AsyncResultIterator[T]): void =
proc finish*[T](self: AsyncResultIter[T]): void =
self.finished = true
proc finished*[T](self: AsyncResultIterator[T]): bool =
proc finished*[T](self: AsyncResultIter[T]): bool =
self.finished
iterator items*[T](self: AsyncResultIterator[T]): auto {.inline.} =
iterator items*[T](self: AsyncResultIter[T]): auto {.inline.} =
while not self.finished:
yield self.next()
iterator pairs*[T](self: AsyncResultIterator[T]): auto {.inline.} =
iterator pairs*[T](self: AsyncResultIter[T]): auto {.inline.} =
var i = 0
while not self.finished:
yield (i, self.next())
inc(i)
proc mapFuture*[T, U](
fut: auto, fn: AsyncResultIteratorFunc[T, U]
fut: auto, fn: AsyncResultIterFunc[T, U]
): Future[U] {.async: (raises: [CancelledError]).} =
let t = await fut
await fn(t)
proc mapAsync*[T, U](
iter: Iter[T], fn: AsyncResultIteratorFunc[T, ?!U], finishOnErr: bool = true
): AsyncResultIterator[U] =
AsyncResultIterator[U].new(
iter: Iter[T], fn: AsyncResultIterFunc[T, ?!U], finishOnErr: bool = true
): AsyncResultIter[U] =
AsyncResultIter[U].new(
genNext = () => fn(iter.next()),
isFinished = () => iter.finished(),
finishOnErr = finishOnErr,
)
proc map*[T, U](
iter: AsyncResultIterator[T],
fn: AsyncResultIteratorFunc[?!T, ?!U],
iter: AsyncResultIter[T],
fn: AsyncResultIterFunc[?!T, ?!U],
finishOnErr: bool = true,
): AsyncResultIterator[U] =
AsyncResultIterator[U].new(
): AsyncResultIter[U] =
AsyncResultIter[U].new(
genNext = () => iter.next().flatMap(fn),
isFinished = () => iter.finished,
finishOnErr = finishOnErr,
)
proc mapFilter*[T, U](
iter: AsyncResultIterator[T],
mapPredicate: AsyncResultIteratorFunc[?!T, Option[?!U]],
iter: AsyncResultIter[T],
mapPredicate: AsyncResultIterFunc[?!T, Option[?!U]],
finishOnErr: bool = true,
): Future[AsyncResultIterator[U]] {.async: (raises: [CancelledError]).} =
): Future[AsyncResultIter[U]] {.async: (raises: [CancelledError]).} =
var nextU: Option[?!U]
proc filter(): Future[void] {.async: (raises: [CancelledError]).} =
@ -203,13 +202,13 @@ proc mapFilter*[T, U](
nextU.isNone
await filter()
AsyncResultIterator[U].new(genNext, isFinished, finishOnErr = finishOnErr)
AsyncResultIter[U].new(genNext, isFinished, finishOnErr = finishOnErr)
proc filter*[T](
iter: AsyncResultIterator[T],
predicate: AsyncResultIteratorFunc[?!T, bool],
iter: AsyncResultIter[T],
predicate: AsyncResultIterFunc[?!T, bool],
finishOnErr: bool = true,
): Future[AsyncResultIterator[T]] {.async: (raises: [CancelledError]).} =
): Future[AsyncResultIter[T]] {.async: (raises: [CancelledError]).} =
proc wrappedPredicate(
t: ?!T
): Future[Option[?!T]] {.async: (raises: [CancelledError]).} =
@ -221,8 +220,8 @@ proc filter*[T](
await mapFilter[T, T](iter, wrappedPredicate, finishOnErr = finishOnErr)
proc delayBy*[T](
iter: AsyncResultIterator[T], d: Duration, finishOnErr: bool = true
): AsyncResultIterator[T] =
iter: AsyncResultIter[T], d: Duration, finishOnErr: bool = true
): AsyncResultIter[T] =
## Delays emitting each item by given duration
##
@ -234,14 +233,14 @@ proc delayBy*[T](
finishOnErr = finishOnErr,
)
proc empty*[T](_: type AsyncResultIterator[T]): AsyncResultIterator[T] =
## Creates an empty AsyncResultIterator
proc empty*[T](_: type AsyncResultIter[T]): AsyncResultIter[T] =
## Creates an empty AsyncResultIter
##
proc genNext(): Future[?!T] {.async: (raises: [CancelledError]).} =
T.failure("Next item requested from an empty AsyncResultIterator")
T.failure("Next item requested from an empty AsyncResultIter")
proc isFinished(): bool =
true
AsyncResultIterator[T].new(genNext, isFinished)
AsyncResultIter[T].new(genNext, isFinished)

6
codex/utils/iter.nim
View File

@ -157,7 +157,9 @@ iterator pairs*[T](self: Iter[T]): tuple[key: int, val: T] {.inline.} =
proc map*[T, U](iter: Iter[T], fn: IterFunction[T, U]): Iter[U] =
Iter[U].new(genNext = () => fn(iter.next()), isFinished = () => iter.finished)
proc mapFilter*[T, U](iter: Iter[T], mapPredicate: IterFunction[T, Option[U]]): Iter[U] =
proc mapFilter*[T, U](
iter: Iter[T], mapPredicate: IterFunction[T, Option[U]]
): Iter[U] =
var nextUOrErr: Option[?!U]
proc tryFetch(): void =
@ -207,4 +209,4 @@ proc empty*[T](_: type Iter[T]): Iter[T] =
proc isFinished(): bool =
true
Iter[T].new(genNext, isFinished)
Iter[T].new(genNext, isFinished)

6
tests/codex/helpers/mockrepostore.nim
View File

@ -14,7 +14,7 @@ import pkg/questionable
import pkg/questionable/results
import pkg/codex/stores/repostore
import pkg/codex/utils/asyncresultiterator
import pkg/codex/utils/asyncresultiter
type MockRepoStore* = ref object of RepoStore
delBlockCids*: seq[Cid]
@ -32,7 +32,7 @@ method delBlock*(
method getBlockExpirations*(
self: MockRepoStore, maxNumber: int, offset: int
): Future[?!AsyncResultIterator[BlockExpiration]] {.async: (raises: [CancelledError]).} =
): Future[?!AsyncResultIter[BlockExpiration]] {.async: (raises: [CancelledError]).} =
self.getBeMaxNumber = maxNumber
self.getBeOffset = offset
@ -41,7 +41,7 @@ method getBlockExpirations*(
limit = min(offset + maxNumber, len(testBlockExpirationsCpy))
let
iter1 = AsyncResultIterator[int].new(offset ..< limit)
iter1 = AsyncResultIter[int].new(offset ..< limit)
iter2 = map[int, BlockExpiration](
iter1,
proc(i: ?!int): Future[?!BlockExpiration] {.async: (raises: [CancelledError]).} =

4
tests/codex/stores/testqueryiterhelper.nim
View File

@ -6,7 +6,7 @@ import pkg/chronos
import pkg/datastore/typedds
import pkg/datastore/sql/sqliteds
import pkg/codex/stores/queryiterhelper
import pkg/codex/utils/asyncresultiterator
import pkg/codex/utils/asyncresultiter
import ../../asynctest
import ../helpers
@ -43,7 +43,7 @@ asyncchecksuite "Test QueryIter helper":
queryIter.dispose = () => (disposed = true; iterDispose())
let
iter1 = (await toAsyncResultIterator[string](queryIter)).tryGet()
iter1 = (await toAsyncResultIter[string](queryIter)).tryGet()
iter2 = await filterSuccess[string](iter1)
var items = initTable[string, string]()

4
tests/codex/stores/testrepostore.nim
View File

@ -15,7 +15,7 @@ import pkg/codex/stores
import pkg/codex/stores/repostore/operations
import pkg/codex/blocktype as bt
import pkg/codex/clock
import pkg/codex/utils/asyncresultiterator
import pkg/codex/utils/asyncresultiter
import pkg/codex/merkletree/codex
import ../../asynctest
@ -293,7 +293,7 @@ asyncchecksuite "RepoStore":
test "Should retrieve block expiration information":
proc unpack(
beIter: Future[?!AsyncResultIterator[BlockExpiration]].Raising([CancelledError])
beIter: Future[?!AsyncResultIter[BlockExpiration]].Raising([CancelledError])
): Future[seq[BlockExpiration]] {.async: (raises: [CancelledError]).} =
var expirations = newSeq[BlockExpiration](0)
without iter =? (await beIter), err:

2
tests/codex/testutils.nim
View File

@ -1,7 +1,7 @@
import ./utils/testoptions
import ./utils/testkeyutils
import ./utils/testasyncstatemachine
import ./utils/testasyncresultiterator
import ./utils/testasyncresultiter
import ./utils/testtimer
import ./utils/testtrackedfutures

37
tests/codex/utils/testasyncresultiterator.nim → tests/codex/utils/testasyncresultiter.nim
View File

@ -2,14 +2,14 @@ import std/sugar
import pkg/questionable
import pkg/chronos
import pkg/codex/utils/iter
import pkg/codex/utils/asyncresultiterator
import pkg/codex/utils/asyncresultiter
import ../../asynctest
import ../helpers
asyncchecksuite "Test AsyncResultIterator":
asyncchecksuite "Test AsyncResultIter":
test "Should be finished":
let iter = AsyncResultIterator[int].empty()
let iter = AsyncResultIter[int].empty()
check:
iter.finished == true
@ -24,7 +24,7 @@ asyncchecksuite "Test AsyncResultIterator":
fut.complete(success(intIter.next()))
return fut
let iter = AsyncResultIterator[int].new(asyncGen, () => intIter.finished)
let iter = AsyncResultIter[int].new(asyncGen, () => intIter.finished)
var collected: seq[int]
for iFut in iter:
@ -37,11 +37,10 @@ asyncchecksuite "Test AsyncResultIterator":
check collected == expectedSeq
let nextRes = await iter.next()
assert nextRes.isFailure
check nextRes.error.msg ==
"AsyncResultIterator is finished but next item was requested"
check nextRes.error.msg == "AsyncResultIter is finished but next item was requested"
test "getting async iter for simple sync range iterator":
let iter1 = AsyncResultIterator[int].new(0 ..< 5)
let iter1 = AsyncResultIter[int].new(0 ..< 5)
var collected: seq[int]
for iFut in iter1:
@ -54,7 +53,7 @@ asyncchecksuite "Test AsyncResultIterator":
collected == @[0, 1, 2, 3, 4]
test "Should map each item using `map`":
let iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
let iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
let iter2 = map[int, string](
iter1,
@ -78,7 +77,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Should leave only odd items using `filter`":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await filter[int](
iter1,
proc(i: ?!int): Future[bool] {.async: (raises: [CancelledError]).} =
@ -101,7 +100,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Should leave only odd items using `mapFilter`":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =
@ -124,7 +123,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `map` when finish on error is true":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -152,7 +151,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `map` when finish on error is false":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -181,7 +180,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `map` when errors are mixed with successes":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -210,7 +209,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `mapFilter` when finish on error is true":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =
@ -240,7 +239,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `mapFilter` when finish on error is false":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =
@ -271,7 +270,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `filter` when finish on error is false":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5)
iter1 = AsyncResultIter[int].new(0 ..< 5)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -314,7 +313,7 @@ asyncchecksuite "Test AsyncResultIterator":
test "Collecting errors on `filter` when finish on error is true":
let
iter1 = AsyncResultIterator[int].new(0 ..< 5)
iter1 = AsyncResultIter[int].new(0 ..< 5)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -383,9 +382,9 @@ asyncchecksuite "Test AsyncResultIterator":
# cancellation of the async predicate function.
let fut: Future[Option[?!string]].Raising([CancelledError]) =
Future[Option[?!string]].Raising([CancelledError]).init("testasyncresultiterator")
Future[Option[?!string]].Raising([CancelledError]).init("testasyncresultiter")
let iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
let iter1 = AsyncResultIter[int].new(0 ..< 5).delayBy(10.millis)
let iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =