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

renaming: SafeAsyncIter => AsyncResultIterator

Browse Source
This commit is contained in:
Marcin Czenko 2025-06-24 12:48:05 +02:00
parent ba04d59192
commit bbadf63f6b
No known key found for this signature in database
GPG Key ID: 33DEA0C8E30937C0
11 changed files with 78 additions and 78 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]
): SafeAsyncIter[(?!bt.Block, int)] =
): AsyncResultIterator[(?!bt.Block, int)] =
## Get pending blocks iterator
##
if indices.len == 0:
trace "No indicies to fetch blocks for", treeCid = manifest.treeCid
return SafeAsyncIter[(?!bt.Block, int)].empty()
trace "No indices to fetch blocks for", treeCid = manifest.treeCid
return AsyncResultIterator[(?!bt.Block, int)].empty()
var pendingBlocks: seq[Future[(?!bt.Block, int)].Raising([CancelledError])] = @[]
@ -163,7 +163,7 @@ proc getPendingBlocks(
# thus, if this happens, we raise an assert
raiseAssert("fatal: pendingBlocks is empty - this should never happen")
SafeAsyncIter[(?!bt.Block, int)].new(genNext, isFinished)
AsyncResultIterator[(?!bt.Block, int)].new(genNext, isFinished)
proc prepareEncodingData(
self: Erasure,

2
codex/stores/blockstore.nim
View File

@ -154,7 +154,7 @@ method hasBlock*(
method listBlocks*(
self: BlockStore, blockType = BlockType.Manifest
): Future[?!SafeAsyncIter[Cid]] {.base, async: (raises: [CancelledError]), gcsafe.} =
): Future[?!AsyncResultIterator[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[?!SafeAsyncIter[Cid]] {.async: (raises: [CancelledError]).} =
): Future[?!AsyncResultIterator[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 (
SafeAsyncIter[Cid].new(genNext, isFinished).filter(
AsyncResultIterator[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/networkstore.nim
View File

@ -127,7 +127,7 @@ method ensureExpiry*(
method listBlocks*(
self: NetworkStore, blockType = BlockType.Manifest
): Future[?!SafeAsyncIter[Cid]] {.async: (raw: true, raises: [CancelledError]).} =
): Future[?!AsyncResultIterator[Cid]] {.async: (raw: true, raises: [CancelledError]).} =
self.localStore.listBlocks(blockType)
method delBlock*(

14
codex/stores/queryiterhelper.nim
View File

@ -10,10 +10,10 @@ import ../utils/safeasynciter
type KeyVal*[T] = tuple[key: Key, value: T]
proc toSafeAsyncIter*[T](
proc toAsyncResultIterator*[T](
queryIter: QueryIter[T], finishOnErr: bool = true
): Future[?!SafeAsyncIter[QueryResponse[T]]] {.async: (raises: [CancelledError]).} =
## Converts `QueryIter[T]` to `SafeAsyncIter[QueryResponse[T]]` and automatically
): Future[?!AsyncResultIterator[QueryResponse[T]]] {.async: (raises: [CancelledError]).} =
## Converts `QueryIter[T]` to `AsyncResultIterator[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 toSafeAsyncIter*[T](
trace "Disposing iterator"
if error =? (await queryIter.dispose()).errorOption:
return failure(error)
return success(SafeAsyncIter[QueryResponse[T]].empty())
return success(AsyncResultIterator[QueryResponse[T]].empty())
var errOccurred = false
@ -42,11 +42,11 @@ proc toSafeAsyncIter*[T](
proc isFinished(): bool =
queryIter.finished
SafeAsyncIter[QueryResponse[T]].new(genNext, isFinished).success
AsyncResultIterator[QueryResponse[T]].new(genNext, isFinished).success
proc filterSuccess*[T](
iter: SafeAsyncIter[QueryResponse[T]]
): Future[SafeAsyncIter[tuple[key: Key, value: T]]] {.
iter: AsyncResultIterator[QueryResponse[T]]
): Future[AsyncResultIterator[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[?!SafeAsyncIter[Cid]] {.async: (raises: [CancelledError]).} =
): Future[?!AsyncResultIterator[Cid]] {.async: (raises: [CancelledError]).} =
## Get the list of blocks in the RepoStore.
## This is an intensive operation
##
var iter = SafeAsyncIter[Cid]()
var iter = AsyncResultIterator[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[?!SafeAsyncIter[BlockExpiration]] {.
): Future[?!AsyncResultIterator[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.toSafeAsyncIter()), err:
without asyncQueryIter =? (await queryIter.toAsyncResultIterator()), err:
error "Unable to convert QueryIter to AsyncIter", err = err.msg
return failure(err)
let filteredIter: SafeAsyncIter[KeyVal[BlockMetadata]] =
let filteredIter: AsyncResultIterator[KeyVal[BlockMetadata]] =
await asyncQueryIter.filterSuccess()
proc mapping(

76
codex/utils/safeasynciter.nim
View File

@ -17,7 +17,7 @@ import pkg/chronos
import ./iter
## SafeAsyncIter[T] is similar to `AsyncIter[Future[T]]`
## AsyncResultIterator[T] is similar to `AsyncIter[Future[T]]`
## but does not throw exceptions others than CancelledError.
## It is thus way easier to use with checked exceptions
##
@ -28,19 +28,19 @@ 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 (SafeAsyncIter)
## - new - to create a new async iterator (AsyncResultIterator)
## - 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 (SafeAsyncIter)
## - map - to convert one async iterator (SafeAsyncIter) to another async iter (SafeAsyncIter)
## - mapFilter - to convert one async iterator (SafeAsyncIter) to another async iter (SafeAsyncIter) and apply filtering at the same time
## - filter - to filter an async iterator (SafeAsyncIter) returning another async iterator (SafeAsyncIter)
## - 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 (SafeAsyncIter)
## - empty - to create an empty async iterator (AsyncResultIterator)
type
SafeFunction[T, U] =
@ -48,7 +48,7 @@ type
SafeIsFinished = proc(): bool {.raises: [], gcsafe, closure.}
SafeGenNext[T] = proc(): Future[T] {.async: (raises: [CancelledError]), gcsafe.}
SafeAsyncIter*[T] = ref object
AsyncResultIterator*[T] = ref object
finished: bool
next*: SafeGenNext[?!T]
@ -65,20 +65,20 @@ proc flatMap[T, U](
await fn(t)
########################################################################
## SafeAsyncIter public interface methods
## AsyncResultIterator public interface methods
########################################################################
proc new*[T](
_: type SafeAsyncIter[T],
_: type AsyncResultIterator[T],
genNext: SafeGenNext[?!T],
isFinished: IsFinished,
finishOnErr: bool = true,
): SafeAsyncIter[T] =
): AsyncResultIterator[T] =
## Creates a new Iter using elements returned by supplier function `genNext`.
## Iter is finished whenever `isFinished` returns true.
##
var iter = SafeAsyncIter[T]()
var iter = AsyncResultIterator[T]()
proc next(): Future[?!T] {.async: (raises: [CancelledError]).} =
try:
@ -91,7 +91,7 @@ proc new*[T](
iter.finished = true
return item
else:
return failure("SafeAsyncIter is finished but next item was requested")
return failure("AsyncResultIterator is finished but next item was requested")
except CancelledError as err:
iter.finished = true
raise err
@ -105,11 +105,11 @@ proc new*[T](
# forward declaration
proc mapAsync*[T, U](
iter: Iter[T], fn: SafeFunction[T, ?!U], finishOnErr: bool = true
): SafeAsyncIter[U]
): AsyncResultIterator[U]
proc new*[U, V: Ordinal](
_: type SafeAsyncIter[U], slice: HSlice[U, V], finishOnErr: bool = true
): SafeAsyncIter[U] =
_: type AsyncResultIterator[U], slice: HSlice[U, V], finishOnErr: bool = true
): AsyncResultIterator[U] =
## Creates new Iter from a slice
##
@ -122,8 +122,8 @@ proc new*[U, V: Ordinal](
)
proc new*[U, V, S: Ordinal](
_: type SafeAsyncIter[U], a: U, b: V, step: S = 1, finishOnErr: bool = true
): SafeAsyncIter[U] =
_: type AsyncResultIterator[U], a: U, b: V, step: S = 1, finishOnErr: bool = true
): AsyncResultIterator[U] =
## Creates new Iter in range a..b with specified step (default 1)
##
@ -135,17 +135,17 @@ proc new*[U, V, S: Ordinal](
finishOnErr = finishOnErr,
)
proc finish*[T](self: SafeAsyncIter[T]): void =
proc finish*[T](self: AsyncResultIterator[T]): void =
self.finished = true
proc finished*[T](self: SafeAsyncIter[T]): bool =
proc finished*[T](self: AsyncResultIterator[T]): bool =
self.finished
iterator items*[T](self: SafeAsyncIter[T]): auto {.inline.} =
iterator items*[T](self: AsyncResultIterator[T]): auto {.inline.} =
while not self.finished:
yield self.next()
iterator pairs*[T](self: SafeAsyncIter[T]): auto {.inline.} =
iterator pairs*[T](self: AsyncResultIterator[T]): auto {.inline.} =
var i = 0
while not self.finished:
yield (i, self.next())
@ -159,27 +159,27 @@ proc mapFuture*[T, U](
proc mapAsync*[T, U](
iter: Iter[T], fn: SafeFunction[T, ?!U], finishOnErr: bool = true
): SafeAsyncIter[U] =
SafeAsyncIter[U].new(
): AsyncResultIterator[U] =
AsyncResultIterator[U].new(
genNext = () => fn(iter.next()),
isFinished = () => iter.finished(),
finishOnErr = finishOnErr,
)
proc map*[T, U](
iter: SafeAsyncIter[T], fn: SafeFunction[?!T, ?!U], finishOnErr: bool = true
): SafeAsyncIter[U] =
SafeAsyncIter[U].new(
iter: AsyncResultIterator[T], fn: SafeFunction[?!T, ?!U], finishOnErr: bool = true
): AsyncResultIterator[U] =
AsyncResultIterator[U].new(
genNext = () => iter.next().flatMap(fn),
isFinished = () => iter.finished,
finishOnErr = finishOnErr,
)
proc mapFilter*[T, U](
iter: SafeAsyncIter[T],
iter: AsyncResultIterator[T],
mapPredicate: SafeFunction[?!T, Option[?!U]],
finishOnErr: bool = true,
): Future[SafeAsyncIter[U]] {.async: (raises: [CancelledError]).} =
): Future[AsyncResultIterator[U]] {.async: (raises: [CancelledError]).} =
var nextU: Option[?!U]
proc filter(): Future[void] {.async: (raises: [CancelledError]).} =
@ -199,11 +199,11 @@ proc mapFilter*[T, U](
nextU.isNone
await filter()
SafeAsyncIter[U].new(genNext, isFinished, finishOnErr = finishOnErr)
AsyncResultIterator[U].new(genNext, isFinished, finishOnErr = finishOnErr)
proc filter*[T](
iter: SafeAsyncIter[T], predicate: SafeFunction[?!T, bool], finishOnErr: bool = true
): Future[SafeAsyncIter[T]] {.async: (raises: [CancelledError]).} =
iter: AsyncResultIterator[T], predicate: SafeFunction[?!T, bool], finishOnErr: bool = true
): Future[AsyncResultIterator[T]] {.async: (raises: [CancelledError]).} =
proc wrappedPredicate(
t: ?!T
): Future[Option[?!T]] {.async: (raises: [CancelledError]).} =
@ -215,8 +215,8 @@ proc filter*[T](
await mapFilter[T, T](iter, wrappedPredicate, finishOnErr = finishOnErr)
proc delayBy*[T](
iter: SafeAsyncIter[T], d: Duration, finishOnErr: bool = true
): SafeAsyncIter[T] =
iter: AsyncResultIterator[T], d: Duration, finishOnErr: bool = true
): AsyncResultIterator[T] =
## Delays emitting each item by given duration
##
@ -228,14 +228,14 @@ proc delayBy*[T](
finishOnErr = finishOnErr,
)
proc empty*[T](_: type SafeAsyncIter[T]): SafeAsyncIter[T] =
## Creates an empty SafeAsyncIter
proc empty*[T](_: type AsyncResultIterator[T]): AsyncResultIterator[T] =
## Creates an empty AsyncResultIterator
##
proc genNext(): Future[?!T] {.async: (raises: [CancelledError]).} =
T.failure("Next item requested from an empty SafeAsyncIter")
T.failure("Next item requested from an empty AsyncResultIterator")
proc isFinished(): bool =
true
SafeAsyncIter[T].new(genNext, isFinished)
AsyncResultIterator[T].new(genNext, isFinished)

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

@ -32,7 +32,7 @@ method delBlock*(
method getBlockExpirations*(
self: MockRepoStore, maxNumber: int, offset: int
): Future[?!SafeAsyncIter[BlockExpiration]] {.async: (raises: [CancelledError]).} =
): Future[?!AsyncResultIterator[BlockExpiration]] {.async: (raises: [CancelledError]).} =
self.getBeMaxNumber = maxNumber
self.getBeOffset = offset
@ -41,7 +41,7 @@ method getBlockExpirations*(
limit = min(offset + maxNumber, len(testBlockExpirationsCpy))
let
iter1 = SafeAsyncIter[int].new(offset ..< limit)
iter1 = AsyncResultIterator[int].new(offset ..< limit)
iter2 = map[int, BlockExpiration](
iter1,
proc(i: ?!int): Future[?!BlockExpiration] {.async: (raises: [CancelledError]).} =

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

@ -43,7 +43,7 @@ asyncchecksuite "Test QueryIter helper":
queryIter.dispose = () => (disposed = true; iterDispose())
let
iter1 = (await toSafeAsyncIter[string](queryIter)).tryGet()
iter1 = (await toAsyncResultIterator[string](queryIter)).tryGet()
iter2 = await filterSuccess[string](iter1)
var items = initTable[string, string]()

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

@ -293,7 +293,7 @@ asyncchecksuite "RepoStore":
test "Should retrieve block expiration information":
proc unpack(
beIter: Future[?!SafeAsyncIter[BlockExpiration]].Raising([CancelledError])
beIter: Future[?!AsyncResultIterator[BlockExpiration]].Raising([CancelledError])
): Future[seq[BlockExpiration]] {.async: (raises: [CancelledError]).} =
var expirations = newSeq[BlockExpiration](0)
without iter =? (await beIter), err:

32
tests/codex/utils/testsafeasynciter.nim
View File

@ -7,9 +7,9 @@ import pkg/codex/utils/safeasynciter
import ../../asynctest
import ../helpers
asyncchecksuite "Test SafeAsyncIter":
asyncchecksuite "Test AsyncResultIterator":
test "Should be finished":
let iter = SafeAsyncIter[int].empty()
let iter = AsyncResultIterator[int].empty()
check:
iter.finished == true
@ -24,7 +24,7 @@ asyncchecksuite "Test SafeAsyncIter":
fut.complete(success(intIter.next()))
return fut
let iter = SafeAsyncIter[int].new(asyncGen, () => intIter.finished)
let iter = AsyncResultIterator[int].new(asyncGen, () => intIter.finished)
var collected: seq[int]
for iFut in iter:
@ -37,10 +37,10 @@ asyncchecksuite "Test SafeAsyncIter":
check collected == expectedSeq
let nextRes = await iter.next()
assert nextRes.isFailure
check nextRes.error.msg == "SafeAsyncIter is finished but next item was requested"
check nextRes.error.msg == "AsyncResultIterator is finished but next item was requested"
test "getting async iter for simple sync range iterator":
let iter1 = SafeAsyncIter[int].new(0 ..< 5)
let iter1 = AsyncResultIterator[int].new(0 ..< 5)
var collected: seq[int]
for iFut in iter1:
@ -53,7 +53,7 @@ asyncchecksuite "Test SafeAsyncIter":
collected == @[0, 1, 2, 3, 4]
test "Should map each item using `map`":
let iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
let iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
let iter2 = map[int, string](
iter1,
@ -77,7 +77,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Should leave only odd items using `filter`":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await filter[int](
iter1,
proc(i: ?!int): Future[bool] {.async: (raises: [CancelledError]).} =
@ -100,7 +100,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Should leave only odd items using `mapFilter`":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =
@ -123,7 +123,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `map` when finish on error is true":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -151,7 +151,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `map` when finish on error is false":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -180,7 +180,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `map` when errors are mixed with successes":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -209,7 +209,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `mapFilter` when finish on error is true":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =
@ -239,7 +239,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `mapFilter` when finish on error is false":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =
@ -270,7 +270,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `filter` when finish on error is false":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5)
iter1 = AsyncResultIterator[int].new(0 ..< 5)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -313,7 +313,7 @@ asyncchecksuite "Test SafeAsyncIter":
test "Collecting errors on `filter` when finish on error is true":
let
iter1 = SafeAsyncIter[int].new(0 ..< 5)
iter1 = AsyncResultIterator[int].new(0 ..< 5)
iter2 = map[int, string](
iter1,
proc(i: ?!int): Future[?!string] {.async: (raises: [CancelledError]).} =
@ -384,7 +384,7 @@ asyncchecksuite "Test SafeAsyncIter":
let fut: Future[Option[?!string]].Raising([CancelledError]) =
Future[Option[?!string]].Raising([CancelledError]).init("testsafeasynciter")
let iter1 = SafeAsyncIter[int].new(0 ..< 5).delayBy(10.millis)
let iter1 = AsyncResultIterator[int].new(0 ..< 5).delayBy(10.millis)
let iter2 = await mapFilter[int, string](
iter1,
proc(i: ?!int): Future[Option[?!string]] {.async: (raises: [CancelledError]).} =