JS 迭代协议高级应用 - 实现异步迭代器与可观察序列的交互模式

async function* observableToAsyncIterator(observable) {
    let valueQueue = [];
    let resolveNext = null; // 用于解决等待下一个值的Promise
    let isDone = false;
    let error = null;

    // 订阅Observable
    const subscription = observable.subscribe({
        next(value) {
            valueQueue.push(value);
            if (resolveNext) {
                resolveNext(); // 通知等待的next()可以获取值了
                resolveNext = null;
            }
        },
        error(err) {
            error = err;
            isDone = true;
            if (resolveNext) {
                resolveNext();
                resolveNext = null;
            }
        },
        complete() {
            isDone = true;
            if (resolveNext) {
                resolveNext();
                resolveNext = null;
            }
        }
    });

    try {
        while (!isDone || valueQueue.length > 0) {
            if (valueQueue.length > 0) {
                yield valueQueue.shift(); // 弹出并返回队列中的值
            } else if (isDone) {
                // 如果已经完成且队列为空，就退出循环
                break;
            } else {
                // 队列为空，且Observable未完成，等待新值
                await new Promise(resolve => {
                    resolveNext = resolve;
                });
            }
            if (error) {
                throw error; // 如果有错误，抛出
            }
        }
    } finally {
        // 确保在迭代器完成或中断时取消订阅
        subscription.unsubscribe();
    }
}

// 示例用法：
// import { interval } from 'rxjs';
// const source$ = interval(1000).pipe(take(5)); // 每秒发出一个值，共5个

// async function main() {
//     console.log("开始消费Observable作为异步迭代器...");
//     for await (const value of observableToAsyncIterator(source$)) {
//         console.log(`收到值: ${value}`);
//     }
//     console.log("异步迭代器消费完成。");
// }

// main();

import { Observable, Subscription } from 'rxjs';

/**
 * 将RxJS Observable转换为异步迭代器。
 * 允许使用 for await...of 语法消费Observable流。
 * @param {Observable<T>} observable 要转换的RxJS Observable。
 * @returns {AsyncIterable<T>} 一个异步迭代器。
 */
function observableToAsyncIterable<T>(observable: Observable<T>): AsyncIterable<T> {
    const queue: T[] = [];
    let error: any = null;
    let isComplete = false;
    let subscription: Subscription | null = null;

    // 用于解决等待下一个值的Promise，或者在完成/错误时通知
    let nextResolve: (() => void) | null = null;
    let nextReject: ((reason?: any) => void) | null = null;

    const pullNext = () => {
        if (nextResolve) {
            nextResolve();
            nextResolve = null;
            nextReject = null;
        }
    };

    const subscribeToObservable = () => {
        if (subscription) return; // 避免重复订阅

        subscription = observable.subscribe({
            next(value) {
                queue.push(value);
                pullNext(); // 有新值了，通知等待的迭代器
            },
            error(err) {
                error = err;
                isComplete = true;
                pullNext(); // 有错误了，通知迭代器抛出
            },
            complete() {
                isComplete = true;
                pullNext(); // 完成了，通知迭代器结束
            }
        });
    };

    // 返回符合异步迭代器协议的对象
    return {
        async next(): Promise<IteratorResult<T>> {
            subscribeToObservable(); // 第一次调用next时订阅

            // 如果队列中有值，直接返回
            if (queue.length > 0) {
                return { value: queue.shift()!, done: false };
            }

            // 如果已经完成且队列为空，则迭代结束
            if (isComplete) {
                if (error) {
                    throw error; // 如果有错误，抛出
                }
                return { value: undefined, done: true };
            }

            // 队列为空且未完成，则等待新值
            await new Promise<void>((resolve, reject) => {
                nextResolve = resolve;
                nextReject = reject;
            });

            // 再次检查队列或完成状态
            if (error) {
                throw error;
            }
            if (queue.length > 0) {
                return { value: queue.shift()!, done: false };
            }
            // 如果走到这里，说明是complete了，且队列为空
            return { value: undefined, done: true };
        },

        // 实现 [Symbol.asyncIterator] 方法，返回自身
        [Symbol.asyncIterator]() {
            return this;
        },

        // 可选：实现 return 和 throw 方法，用于迭代器提前结束或处理错误
        async return(value?: any): Promise<IteratorResult<T>> {
            if (subscription) {
                subscription.unsubscribe(); // 清理资源
                subscription = null;
            }
            return { value: value, done: true };
        },
        async throw(err?: any): Promise<IteratorResult<T>> {
            if (subscription) {
                subscription.unsubscribe(); // 清理资源
                subscription = null;
            }
            throw err;
        }
    };
}

// 示例用法 (假设你已经安装了 rxjs)
// import { interval, take } from 'rxjs';

// async function runExample() {
//     console.log("--- 开始使用 Observable 转换的异步迭代器 ---");
//     const source$ = interval(500).pipe(take(7)); // 每500ms发出一个值，共7个

//     try {
//         for await (const val of observableToAsyncIterable(source$)) {
//             console.log(`收到值: ${val}`);
//             if (val === 3) {
//                 console.log("手动中断迭代器在值 3。");
//                 break; // 模拟提前退出循环
//             }
//         }
//     } catch (e) {
//         console.error("迭代器中发生错误:", e);
//     } finally {
//         console.log("--- 异步迭代器消费结束 ---");
//     }
// }

// runExample();