-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: scheduling with Rx-provided schedulers will no longer leak actio…
…n references Resolves #6561
- Loading branch information
Showing
12 changed files
with
139 additions
and
160 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,28 +1,31 @@ | ||
import { SchedulerLike } from '../types'; | ||
import { Observable } from '../Observable'; | ||
import { Subscription } from '../Subscription'; | ||
import { executeSchedule } from '../util/executeSchedule'; | ||
|
||
export function scheduleAsyncIterable<T>(input: AsyncIterable<T>, scheduler: SchedulerLike) { | ||
if (!input) { | ||
throw new Error('Iterable cannot be null'); | ||
} | ||
return new Observable<T>(subscriber => { | ||
const sub = new Subscription(); | ||
sub.add( | ||
scheduler.schedule(() => { | ||
const iterator = input[Symbol.asyncIterator](); | ||
sub.add(scheduler.schedule(function () { | ||
iterator.next().then(result => { | ||
return new Observable<T>((subscriber) => { | ||
executeSchedule(subscriber, scheduler, () => { | ||
const iterator = input[Symbol.asyncIterator](); | ||
executeSchedule( | ||
subscriber, | ||
scheduler, | ||
() => { | ||
iterator.next().then((result) => { | ||
if (result.done) { | ||
// This will remove the subscriptions from | ||
// the parent subscription. | ||
subscriber.complete(); | ||
} else { | ||
subscriber.next(result.value); | ||
this.schedule(); | ||
} | ||
}); | ||
})); | ||
}) | ||
); | ||
return sub; | ||
}, | ||
0, | ||
true | ||
); | ||
}); | ||
}); | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,19 +1,8 @@ | ||
import { Observable } from '../Observable'; | ||
import { Subscription } from '../Subscription'; | ||
import { observable as Symbol_observable } from '../symbol/observable'; | ||
import { InteropObservable, SchedulerLike, Subscribable } from '../types'; | ||
import { innerFrom } from '../observable/from'; | ||
import { observeOn } from '../operators/observeOn'; | ||
import { subscribeOn } from '../operators/subscribeOn'; | ||
import { InteropObservable, SchedulerLike } from '../types'; | ||
|
||
export function scheduleObservable<T>(input: InteropObservable<T>, scheduler: SchedulerLike) { | ||
return new Observable<T>(subscriber => { | ||
const sub = new Subscription(); | ||
sub.add(scheduler.schedule(() => { | ||
const observable: Subscribable<T> = (input as any)[Symbol_observable](); | ||
sub.add(observable.subscribe({ | ||
next(value) { sub.add(scheduler.schedule(() => subscriber.next(value))); }, | ||
error(err) { sub.add(scheduler.schedule(() => subscriber.error(err))); }, | ||
complete() { sub.add(scheduler.schedule(() => subscriber.complete())); }, | ||
})); | ||
})); | ||
return sub; | ||
}); | ||
return innerFrom(input).pipe(subscribeOn(scheduler), observeOn(scheduler)); | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,22 +1,8 @@ | ||
import { Observable } from '../Observable'; | ||
import { innerFrom } from '../observable/from'; | ||
import { observeOn } from '../operators/observeOn'; | ||
import { subscribeOn } from '../operators/subscribeOn'; | ||
import { SchedulerLike } from '../types'; | ||
|
||
export function schedulePromise<T>(input: PromiseLike<T>, scheduler: SchedulerLike) { | ||
return new Observable<T>((subscriber) => { | ||
return scheduler.schedule(() => | ||
input.then( | ||
(value) => { | ||
subscriber.add( | ||
scheduler.schedule(() => { | ||
subscriber.next(value); | ||
subscriber.add(scheduler.schedule(() => subscriber.complete())); | ||
}) | ||
); | ||
}, | ||
(err) => { | ||
subscriber.add(scheduler.schedule(() => subscriber.error(err))); | ||
} | ||
) | ||
); | ||
}); | ||
return innerFrom(input).pipe(subscribeOn(scheduler), observeOn(scheduler)); | ||
} |
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.