Merge pull request #1030 from LukasKorba/1029-AsyncThrowingStream-update-so-it-does-not-use-Task

[#1029] AsyncThrowingStream update so it doesn't use Task
This commit is contained in:
Michal Fousek 2023-05-08 14:46:08 +02:00 committed by GitHub
commit b582e1e4e1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 35 additions and 46 deletions

View File

@ -164,18 +164,15 @@ extension LightWalletGRPCService: LightWalletService {
func blockRange(_ range: CompactBlockRange) -> AsyncThrowingStream<ZcashCompactBlock, Error> { func blockRange(_ range: CompactBlockRange) -> AsyncThrowingStream<ZcashCompactBlock, Error> {
let stream = compactTxStreamer.getBlockRange(range.blockRange()) let stream = compactTxStreamer.getBlockRange(range.blockRange())
var iterator = stream.makeAsyncIterator()
return AsyncThrowingStream { continuation in return AsyncThrowingStream() {
Task {
do { do {
for try await block in stream { guard let block = try await iterator.next() else { return nil }
continuation.yield(ZcashCompactBlock(compactBlock: block)) return ZcashCompactBlock(compactBlock: block)
}
continuation.finish()
} catch { } catch {
let serviceError = error.mapToServiceError() let serviceError = error.mapToServiceError()
continuation.finish(throwing: ZcashError.serviceBlockRangeFailed(serviceError)) throw ZcashError.serviceBlockRangeFailed(serviceError)
}
} }
} }
} }
@ -220,13 +217,12 @@ extension LightWalletGRPCService: LightWalletService {
utxoArgs.startHeight = UInt64(height) utxoArgs.startHeight = UInt64(height)
} }
let stream = compactTxStreamer.getAddressUtxosStream(args) let stream = compactTxStreamer.getAddressUtxosStream(args)
var iterator = stream.makeAsyncIterator()
return AsyncThrowingStream { continuation in return AsyncThrowingStream() {
Task {
do { do {
for try await reply in stream { guard let reply = try await iterator.next() else { return nil }
continuation.yield( return UTXO(
UTXO(
id: nil, id: nil,
address: reply.address, address: reply.address,
prevoutTxId: reply.txid, prevoutTxId: reply.txid,
@ -236,13 +232,9 @@ extension LightWalletGRPCService: LightWalletService {
height: Int(reply.height), height: Int(reply.height),
spentInTx: nil spentInTx: nil
) )
)
}
continuation.finish()
} catch { } catch {
let serviceError = error.mapToServiceError() let serviceError = error.mapToServiceError()
continuation.finish(throwing: ZcashError.serviceFetchUTXOsFailed(serviceError)) throw ZcashError.serviceFetchUTXOsFailed(serviceError)
}
} }
} }
} }
@ -258,18 +250,15 @@ extension LightWalletGRPCService: LightWalletService {
), ),
callOptions: Self.callOptions(timeLimit: self.streamingCallTimeout) callOptions: Self.callOptions(timeLimit: self.streamingCallTimeout)
) )
var iterator = stream.makeAsyncIterator()
return AsyncThrowingStream { continuation in return AsyncThrowingStream() {
Task {
do { do {
for try await compactBlock in stream { guard let compactBlock = try await iterator.next() else { return nil }
continuation.yield(ZcashCompactBlock(compactBlock: compactBlock)) return ZcashCompactBlock(compactBlock: compactBlock)
}
continuation.finish()
} catch { } catch {
let serviceError = error.mapToServiceError() let serviceError = error.mapToServiceError()
continuation.finish(throwing: ZcashError.serviceBlockStreamFailed(serviceError)) throw ZcashError.serviceBlockStreamFailed(serviceError)
}
} }
} }
} }