Suppose I have a subscription like this:
mySubject$.subscribe(async inp => await complexFunction(inp))
Suppose that mySubject$ is a Subject that I own. I need a way to do this somehow:
await mySubject$.next(inp)
so that I can wait for the complexFunction to complete.
I tried to create a custom Subject implementation for this, but it does not seem to be easy. Maybe I'm missing somethink and it's actually simple.
What are my options here?
Update:
Based on the idea that rxjs APIs should accept Promises, I tried the code below. It does not work, so again, I'm out of ideas.
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');
});
});
I found a solution. It's not very simple, but it's not hard to understand. Below is not a complete implementation, you should add cleanup and checks, but it works.
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();
}
}
Example usage:
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');
}));
I'm not exactly sure of your intent, but maybe something like this would work:
const source$ = mySubject$.pipe(
mergeMap(inp => complexFunction(inp))
);
source$.subscribe(
value => console.log('received value from complexFunction', value)
);
source$ is an observable that will emit the result of complexFunction() each time mySubject$ emits.