| 98 | } |
| 99 | |
| 100 | export function replicateAsyncIterator<T, TReturn, TNext>( |
| 101 | source: AsyncIterator<T, TReturn, TNext>, |
| 102 | count: number, |
| 103 | ): (AsyncIteratorClass<T, TReturn, TNext>)[] { |
| 104 | const queue = new AsyncIdQueue< |
| 105 | { next: IteratorResult<T, TReturn> } | { next?: never, error: unknown } |
| 106 | >() |
| 107 | |
| 108 | const ids = Array.from({ length: count }, (_, i) => i.toString()) |
| 109 | let isSourceFinished = false |
| 110 | |
| 111 | const start = once(async () => { |
| 112 | try { |
| 113 | while (true) { |
| 114 | const item = await source.next() |
| 115 | |
| 116 | ids.forEach((id) => { |
| 117 | if (queue.isOpen(id)) { |
| 118 | queue.push(id, { next: item }) |
| 119 | } |
| 120 | }) |
| 121 | |
| 122 | if (item.done) { |
| 123 | break |
| 124 | } |
| 125 | } |
| 126 | } |
| 127 | catch (error) { |
| 128 | ids.forEach((id) => { |
| 129 | if (queue.isOpen(id)) { |
| 130 | queue.push(id, { error }) |
| 131 | } |
| 132 | }) |
| 133 | } |
| 134 | finally { |
| 135 | isSourceFinished = true |
| 136 | } |
| 137 | }) |
| 138 | |
| 139 | const replicated: AsyncIteratorClass<T, TReturn, TNext>[] = ids.map((id) => { |
| 140 | queue.open(id) |
| 141 | |
| 142 | return new AsyncIteratorClass( |
| 143 | async () => { |
| 144 | start() |
| 145 | |
| 146 | const item = await queue.pull(id) |
| 147 | |
| 148 | if (item.next) { |
| 149 | return item.next |
| 150 | } |
| 151 | |
| 152 | throw item.error |
| 153 | }, |
| 154 | async (reason) => { |
| 155 | queue.close({ id }) |
| 156 | |
| 157 | if (reason !== 'next' && !queue.length && !isSourceFinished) { |