mirror of
https://github.com/denoland/deno.git
synced 2025-07-30 08:34:35 +00:00
BREAKING(std): reorganization (#5087)
* Prepend underscores to private modules * Remove collectUint8Arrays() It would be a misuse of Deno.iter()'s result. * Move std/_util/async.ts to std/async * Move std/util/sha*.ts to std/hash
This commit is contained in:
parent
2b02535028
commit
f184332c09
100 changed files with 306 additions and 365 deletions
58
std/async/mux_async_iterator.ts
Normal file
58
std/async/mux_async_iterator.ts
Normal file
|
@ -0,0 +1,58 @@
|
|||
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
|
||||
import { Deferred, deferred } from "./deferred.ts";
|
||||
|
||||
interface TaggedYieldedValue<T> {
|
||||
iterator: AsyncIterableIterator<T>;
|
||||
value: T;
|
||||
}
|
||||
|
||||
/** The MuxAsyncIterator class multiplexes multiple async iterators into a
|
||||
* single stream. It currently makes a few assumptions:
|
||||
* - The iterators do not throw.
|
||||
* - The final result (the value returned and not yielded from the iterator)
|
||||
* does not matter; if there is any, it is discarded.
|
||||
*/
|
||||
export class MuxAsyncIterator<T> implements AsyncIterable<T> {
|
||||
private iteratorCount = 0;
|
||||
private yields: Array<TaggedYieldedValue<T>> = [];
|
||||
private signal: Deferred<void> = deferred();
|
||||
|
||||
add(iterator: AsyncIterableIterator<T>): void {
|
||||
++this.iteratorCount;
|
||||
this.callIteratorNext(iterator);
|
||||
}
|
||||
|
||||
private async callIteratorNext(
|
||||
iterator: AsyncIterableIterator<T>
|
||||
): Promise<void> {
|
||||
const { value, done } = await iterator.next();
|
||||
if (done) {
|
||||
--this.iteratorCount;
|
||||
} else {
|
||||
this.yields.push({ iterator, value });
|
||||
}
|
||||
this.signal.resolve();
|
||||
}
|
||||
|
||||
async *iterate(): AsyncIterableIterator<T> {
|
||||
while (this.iteratorCount > 0) {
|
||||
// Sleep until any of the wrapped iterators yields.
|
||||
await this.signal;
|
||||
|
||||
// Note that while we're looping over `yields`, new items may be added.
|
||||
for (let i = 0; i < this.yields.length; i++) {
|
||||
const { iterator, value } = this.yields[i];
|
||||
yield value;
|
||||
this.callIteratorNext(iterator);
|
||||
}
|
||||
|
||||
// Clear the `yields` list and reset the `signal` promise.
|
||||
this.yields.length = 0;
|
||||
this.signal = deferred();
|
||||
}
|
||||
}
|
||||
|
||||
[Symbol.asyncIterator](): AsyncIterableIterator<T> {
|
||||
return this.iterate();
|
||||
}
|
||||
}
|
Loading…
Add table
Add a link
Reference in a new issue