Me quedé atascado, tratando de implementar la lógica de combine para una lista de iterables mixtos, es decir, tengo una lista de Iterable + Iterator + AsyncIterable + AsyncIterator , para lo cual estoy tratando de combinarlos, para obtener el mismo resultado como con combineLatestWith de RXJS .
Enlace a la fuente , más lo mismo a continuación ( mis documentos para el operador ):
(Ver enlace al parque infantil completo en la parte inferior)
function combineAsync<T>(iterable: AsyncIterable<T>, ...values: AnyIterable<T>[]): AsyncIterable<any[]> { return { [Symbol.asyncIterator](): AsyncIterator<T[]> { const list: AnyIterator<any>[] = [ iterable[Symbol.asyncIterator](), ...values.map((v: any) => typeof v[Symbol.iterator] === 'function' ? v[Symbol.iterator]() : (typeof v[Symbol.asyncIterator] === 'function' ? v[Symbol.asyncIterator]() : v)) ]; const pending = new Promise(() => { // forever-pending promise }); let start: Promise<IteratorResult<any[]>>, finished: boolean, latest: any[] = new Array(list.length), changed = false, finishedCount = 0, lastError: { err: any } | null; return { next(): Promise<IteratorResult<any>> { if (!start) { start = Promise.all(list.map(a => a.next())).then(all => { const value = []; for (let i = 0; i < all.length; i++) { const m = all[i]; if (m.done) { finished = true; return m; } value.push(m.value); } latest = [...value]; return {value, done: false}; }); return start; } if (!finished) { const getValues = () => list.map((a, index) => { if (!a) { return pending; } const p = a.next() as any; const it = typeof p.then === 'function' ? p : Promise.resolve(p); return it.then((v: any) => { if (v.done) { list[index] = null as any; // stop requesting values; if (++finishedCount === list.length) { return true; // the end; } return pending; } latest[index] = v.value; changed = true; }).catch((err: any) => { lastError = lastError || {err}; }); }); return start .then(() => { if (lastError) { const r = Promise.reject(lastError.err); lastError = null; return r; } if (changed) { changed = false; return {value: [...latest], done: false}; } return Promise.race(getValues()).then(end => { if (end) { finished = true; return {value: undefined, done: true}; } changed = false; return {value: [...latest], done: false}; }); }); } return Promise.resolve({value: undefined, done: true}); } }; } }; } Entonces, cuando paso 3 parámetros: p1, p2(8), p3(7) , definidos a continuación...
const p1 = [1, 2, 3, 4]; // converted to async iterable const p2 = async function* evenNumbers(maxEven: number): AsyncIterableIterator<number> { for (let i = 2; i <= maxEven; i += 2) { yield new Promise<number>(resolve => { setTimeout(() => resolve(i), 10); }); } }; const p3 = async function* oddNumbers(maxOdd: number): AsyncIterableIterator<number> { for (let i = 1; i <= maxOdd; i += 2) { yield new Promise<number>(resolve => { setTimeout(() => resolve(i), 5); }); } };... Esperaba obtener algo como esto:
[1, 2, 1] [2, 2, 1] [3, 2, 1] [4, 2, 1] [4, 2, 3] [4, 4, 3] [4, 4, 5] [4, 4, 7] [4, 6, 7] [4, 8, 7]pero en cambio, obtengo lo siguiente:
[1, 2, 1] [2, 2, 1] [3, 2, 1] [4, 2, 1] Pasé horas depurando este monstruo asíncrono, pero no pude entender cómo las actualizaciones de los iterables asíncronos no logran llegar a las llamadas de Promise.race que siguen.
¡Cualquier ayuda es muy apreciada!
Aquí está el patio de recreo completo .
ACTUALIZAR
Para demostrar que los valores correctos generalmente existen en el código, aquí está la versión con la consola principal comentada y, en su lugar, agregada en otros dos lugares en la función principal.
Vitaly hiciste un problema interesante. :) Es bastante complicado reutilizar las promesas ya lanzadas en Promise.race() pero es posible.
El error y el rechazo no se manejan aquí, pero si todo está bien, ese código se puede agregar más tarde.
class CachedIterator<T>{ protected lastValue: T | undefined; protected lastValueFetched: boolean = false; public _done = false; protected cachedIteration: Promise<() => IteratorResult<T>> | undefined; protected iterator: AsyncIterator<T>; constructor(iterable: AsyncIterable<T>, protected id?: string){ const v = iterable as any; this.iterator = (typeof v[Symbol.iterator] === 'function' ? v[Symbol.iterator]() : (typeof v[Symbol.asyncIterator] === 'function' ? v[Symbol.asyncIterator]() : v)) as AsyncIterator<T> } async next(): Promise<(() => IteratorResult<T>) | undefined>{ if(this._done) return undefined; if(!this.cachedIteration){ this.cachedIteration = this.iterator.next().then( (result)=> { return () => { this.fetch(result); return result; } } ) } return this.cachedIteration } async nextAndFetch() { const fetch = await this.next(); if(fetch) fetch(); } protected fetch(result: IteratorResult<T>){ this.cachedIteration = undefined; this.lastValueFetched = true; if(result.done){ this._done = true; if (result.value !== undefined){ this.lastValue = result.value; } return; } this.lastValue = result.value; //console.log("AWAITED next Value:", iteration ) } async last(): Promise<T> { if(!this.lastValueFetched){ //console.log("no first value, request Next"); await this.nextAndFetch(); } return this.lastValue!; } done(){ return this._done } } function combineAsync<T>(...values: AnyIterable<T>[]): AsyncIterable<any[]> { return { [Symbol.asyncIterator](): AsyncIterator<T[]> { let done = false; const list: CachedIterator<T>[] = values.map((v: any, id) => new CachedIterator<T>(v, 'id' + id)); //console.log("LIST", list); return { async next() { let skipDoneIteration = true; //FLAG for protection from ending iterations; while(skipDoneIteration){ if( list.every( f => f.done() ) ){ return { done: true, value: undefined } } /* RACE is the main problem here and it's obligatory we launch promises for race and one of them will be cached and other will end someday, maybe before we run next RACE so we need to separate getting async iteration results and fetching: drop iteration cache, converting iteratorResult to last() value so each iteration of combineAsync must have resulted with one fetch */ skipDoneIteration = false; const result = await Promise.race( list.filter(a => !a.done() ).map( a => a.next() ) ).then( fetch => { if(fetch){ return fetch(); } return undefined; }); if(result){ skipDoneIteration = !!result.done; /* another problem is final iterations with response { done: true, value: undefined } we must skip them; */ } } return Promise.all( list.map(a => a.last())).then(values => { return {value: values, done: false} }); } } } } }