How can I "monkey patch" an Observable for Zone.js?

Question

I'm creating an Angular 2 component, and Angular's change detection isn't working for me when using a certain Observable pattern. It looks something like this:

    let getResult$ = this.http.get('/api/identity-settings');

    let manager$ = getResult$
        .map((response) => /* -- create manager object -- */);

    let signinResponse$ = manager$
        .flatMap(manager => manager.processSigninResponse());

    this.readyToLogin$ = manager$.map(() => true).startWith(false);
    this.isLoggedIn$ = signinResponse$.map(() => true).startWith(false);

Then in my template:

<h1>Ready to Log In: {{readyToLogin$ | async}}</h1>
<h1>Logged In: {{isLoggedIn$ | async}}</h1>

Since the readyToLogin$ Observable is based on a synchronous set of operations that happen in response to the http.get() (which Angular "monkey patches" to ensure it knows when it needs to detect changes), the "Ready to Log In" message switches to true at the appropriate time.

However, since processSignInResponse() produces a Promise<>, anything subscribing to the result of flatMap is occurring asynchronously to the http request's completion event. Therefore, it requires manual intervention to notify my component's zone that it needs to check for changes.

How can I wrap the signInResponse$ observable in a way that NgZone knows to detect changes after any subscriptions have been resolved?

Update

Brandon's answer worked until I updated to RC5, at which point things stopped working again. Turns out the 3rd-party library I was using borked Zone.js. Once that was resolved, there was no need to use a workaround at all--the built-in monkey patching just worked!

Answer 1

For RxJs 6 with the pipable operator:

private runInZone(zone) {
  return function mySimpleOperatorImplementation(source) {
    return Observable.create(observer => {
      const onNext = (value) => zone.run(() => observer.next(value));
      const onError = (e) => zone.run(() => observer.error(e));
      const onComplete = () => zone.run(() => observer.complete());
      return source.subscribe(onNext, onError, onComplete);
    });
  };
}

Usage:

$someObservable.pipe(runInZone(zone));

Answer 2

You can make a new observeOnZone operator that can be used to "monkey patch" any observable. Something like:

Rx.Observable.prototype.observeOnZone = function (zone) {
    return Observable.create(observer => {
        var onNext = (value) => zone.run(() => observer.next(value));
        var onError = (e) => zone.run(() => observer.error(e));
        var onComplete = () => zone.run(() => observer.complete());
        return this.subscribe(onNext, onError, onComplete);
    });
};

And use it like so:

this.isLoggedIn$ = signinResponse$.map(() => true).startWith(false).observeOnZone(zone);