如何等待 Subject.next答案

【问题标题】：How to await Subject.next如何等待 Subject.next
【发布时间】：2021-09-30 11:32:09
【问题描述】：

假设我有这样的订阅：

mySubject$.subscribe(async inp => await complexFunction(inp))

假设mySubject$ 是我拥有的主题。我需要一种方法来做到这一点：

await mySubject$.next(inp)

这样我就可以等待complexFunction 完成。

我尝试为此创建一个自定义主题实现，但这似乎并不容易。也许我错过了一些想法，它实际上很简单。

我有什么选择？

更新：

基于 rxjs API 应该接受 Promises 的想法，我尝试了下面的代码。它行不通，所以我又没办法了。

function rigObservable<T>(observable: Observable<T>) {
    let resolve;
    return {
        awaiter: new Promise<void>(res => resolve = res),
        observable: observable.pipe(tap(() => resolve()))
    };
}

fdescribe('test', () => {
    it('should work', async () => {
        const subject = new Subject<string>();
        const wrapper = rigObservable(subject);
        let testValue;
        wrapper.observable.subscribe(inp => of(inp).pipe(delay(1000), tap(t => testValue = t)).toPromise());
        subject.next('text');
        await wrapper.awaiter;
        expect(testValue).toEqual('text');
    });
});

【问题讨论】：

标签： javascript rxjs

【解决方案1】：

我不确定你的意图，但也许这样的事情会起作用：

const source$ = mySubject$.pipe(
    mergeMap(inp => complexFunction(inp))
);

source$.subscribe(
    value => console.log('received value from complexFunction', value)
);

source$ 是一个 observable，每次 mySubject$ 发出时都会发出 complexFunction() 的结果。

【讨论】：

不起作用，在这种情况下，日志会立即发生，而不是在 complexFunction 完成后发生。（假设 complexFunction 是异步的。在我的问题中，我写道：mySubject$.subscribe(async inp => await complexFunction(inp))。订阅者函数是异步的。）
你试过了吗？它只会在 complexFunction 发出后才会发出！
我试试
我应该使用mergeMap 而不是switchMap。这是一个有效的 StackBlitz 演示。
我对此进行了更深入的检查。我认为这个概念不适用于我的用例。我无法控制 complexFunction，我不能将它放在 mergeMap 中。我只能控制主题。我想这将基于 rxjs API 接受 Promises 的想法起作用。基于此，我在我的问题中添加了一个示例测试代码。原则上应该可行，但实际上不行。

【解决方案2】：

我找到了解决方案。这不是很简单，但也不难理解。下面不是一个完整的实现，你应该添加清理和检查，但它可以工作。

function awaitNext(testFunc) {
    return async function() {
        Subject.prototype['saved_subscribe'] = Subject.prototype.subscribe;
        Subject.prototype['saved_next'] = Subject.prototype.next;
        Subject.prototype['promiseArray'] = [];
        Subject.prototype['subscribe'] = function (n, e, c) {
            let wrappedN, wrappedE, wrappedC;
            wrappedN = v => {
                const p = n(v);
                if (p instanceof Promise) this.promiseArray.push(p);
            }
            wrappedE = v => {
                const p = e(v);
                if (p instanceof Promise) this.promiseArray.push(p);
            }
            wrappedC = v => {
                const p = c(v);
                if (p instanceof Promise) this.promiseArray.push(p);
            }
            return this.saved_subscribe(wrappedN, wrappedE, wrappedC);
        } as any;
        Subject.prototype['next'] = function (v) {
            this.saved_next(v);
            return Promise.all(this.promiseArray).then(() => this.promiseArray = []);
        }
        return await testFunc();
    }
}

示例用法：

it('works', awaitNext(async () => {
        const subject = new Subject<string>();
        let actualValue;
        subject.subscribe(v => of(v).pipe(delay(2000), tap(t => actualValue = t)).toPromise());
        await subject.next('foo');
        expect(actualValue).toEqual('foo');
}));

【讨论】：