New feature: allow injectable SubscriptionStrategies for AsyncPipe

[WouterSpaak]

I'm submitting a...

[ ] Regression (a behavior that used to work and stopped working in a new release)
[ ] Bug report 
[ ] Performance issue
[x] Feature request
[ ] Documentation issue or request
[ ] Support request => Please do not submit support request here, instead see https://github.com/angular/angular/blob/master/CONTRIBUTING.md#question
[ ] Other... Please describe:

Current behavior

The AsyncPipe currently supports two types of asynchronous objects, Promise<T> and Observable<T>. It is impossible to add additional SubscriptionStrategies.

Expected behavior

There are other implementations of asynchronous delivery of values than just Observable and Promise. MobX comes to mind, and our team has been working extensively with a state management library we developed in-house called Sherlock. I'm sure there are other libraries out there, and there will be others coming along in the future as well.

I propose adding a multi provider for SUBSCRIPTION_STRATEGY_FACTORIES to CommonModule. When using a factory pattern, the respective strategies created by those factories could hold an internal reference to a SubscriptionLike, cleaning up some code in AsyncPipe#dispose.

By default, an ObservableStrategyFactory and a PromiseStrategyFactory should be provided, but consumers would be able to provide their own strategy factories to this provider.

A SubscriptionStrategyFactory should be able to determine whether it supports a passed in asynchronous object, and if so, create a new SubscriptionStrategy<T> for that type.

Proposed implementation:

This proposal requires quite a few changes. In order to let strategies keep some internal state on their subscription, a factory pattern is needed (not unlike the differ factories for ngForOf). These factories should declare their support for different flavours of asynchronous value sources, and create a strategy when needed.

An interface for SubscriptionStrategyFactory would look something like this:

interface SubscriptionStrategyFactory<T> {
    supports(async: T): boolean;
    create(): SubscriptionStrategy<T>;
}

An interface for SubscriptionStrategy would look like this:

interface SubscriptionStrategy<T> {
    private subscription?: SubscriptionLike;
    createSubscription(async: T, updateLatestValue: any): void;
    dispose(): void;
    onDestroy(): void;
}

The method AsyncPipe#_selectStrategy would have to be refactored to this:

private _selectStrategy(obj: any) {
    const factory = this.subscriptionStrategyFactories.find(factory => factory.supports(obj));
    if(!factory) {
        throw invalidPipeArgumentError(AsyncPipe, obj);
    }
    return factory.create();
}

The implementation for ObservableStrategy, for instance, would then look something like this:

export class ObservableStrategyFactory implements SubscriptionStrategyFactory<Observable<any>> {
    supports(async: Observable<any>) { return ɵisObservable(async); }
    create() { return new ObservableStrategy(); }
}

export class ObservableStrategy implements SubscriptionStrategy<Observable<any>> {
    private _subscription: Subscription;
    createSubscription(async: Observable<any>, updateLatestValue: any) {
        this._subscription = async.subscribe({next: updateLatestValue, error: (e: any) => { throw e; }});
    }
    dispose() { this.subscription && this.subscription.unsubscribe(); }
    onDestroy() { this.subscription && this.subscription.unsubscribe(); }
}

Implementing a new SubscriptionStrategy would then look something like this, in this case implementing a strategy for a Sherlock Derivable<T>.

export class DerivableStrategyFactory implements SubscriptionStrategyFactory<Derivable<any>> {
    supports(async: Derivable<any>) { return isDerivable(async); }
    create() { return new DerivableStrategy(); }
}

export class DerivableStrategy implements SubscriptionStrategy<Derivable<any>> {
    private readonly stopReaction$ = atom<boolean>(false);
    createSubscription(async: Derivable<any>, updateLatestValue: any) {
        async.react(v => updateLatestValue(v), { until: this.stopReaction$ });
    }
    dispose() { this.stopReaction$.set(true); }
    onDestroy() { this.stopReaction$.set(true); }
}

// in an NgModule...
@NgModule({
    providers: [{ provide: SUBSCRIPTION_STRATEGY_FACTORIES, useClass: DerivableStrategyFactory, multi: true }]
})
export class MyModule { }

I'm not too familiar with how MobX handles reactions, but I presume @observable reactions could be easily implemented with this pattern.

Minimal reproduction of the problem with instructions

n/a

What is the motivation / use case for changing the behavior?

Benefits are:

1. Making the AsyncPipe a generic link between the ChangeDetectorRef and any asynchronous emitter
2. Allowing pluggability of the AsyncPipe will 'future-proof' its implementation
3. Internalising logic for subscription/unsupscription inside strategies, and letting the AsyncPipe handle the change detection

I've been working on implementing this for a couple of days, but it got a bit larger than I expected. I have it working quite nicely, but it touches upon some core architecture principles in AsyncPipe and I'd like to test the waters and see what people think of this proposal before I clean up a PR and write extensive tests.

Environment

Not applicable.

[dawidgarus]

Wouldn't it be better to provide service that converts object to Observable, instead of SubscriptionStrategy?
Example:

class DerivableToObservable<T=any> implements ToObservable<Derivable<T>, T> {
  supports(obj: any): obj is Derivable<T> {
    return isDerivable(obj);
  }
  createObservable(obj: Derivable<T>): Observable<T> {
    const stop$ = atom<boolean>(false);
    return new Observable<T>(sub => {
      obj.react(v => sub.next(v), { until: stop$ });
      return () => stop$.set(true);
    });
  }
}

It's simpler and could be reused for other purposes than in async pipe.

[WouterSpaak]

Rxjs is not a panacea for asynchrony. While observables solve a lot of problems, they're not a catch-all for all asynchronous operations. I argue that the AsyncPipe ought to support every type of asynchrony. We've been working with asynchronous proxies for example, which would not work with your example. I think the AsyncPipe is a very powerful tool that is currently hindered by its own implementation. I'd like to free up the code.

[trotyl]

While I agree to this proposal, Observable (not rxjs) is indeed the universal asynchronous pattern.

A little problem in JavaScript could be which of AsyncIterator and Observable is more fundamental, there're arguments at tc39/proposal-async-iteration#47 and tc39/proposal-observable#111.

Eventually all types providing asynchronous-compatible functionality should implement Symbol.asyncIterator or Symbol.observable itself, rather than waiting for every tool to treat them separately.

[robwormald]

I'm with @trotyl on this one - it would be reasonably straightforward to ship pipes for MobX or whatever, without adding to the complexity/code-size of core. It's likely we'll roll the async pipe's facility into the core of the templating language in the near future, and I'd rather not introduce an API that we won't be able to support in the future.

Better for this to live as a third-party OSS library, I think?

[angular-robot]

Just a heads up that we kicked off a community voting process for your feature request. There are 20 days until the voting process ends.

Find more details about Angular's feature request process in our documentation.

[angular-robot]

Thank you for submitting your feature request! Looks like during the polling process it didn't collect a sufficient number of votes to move to the next stage.

We want to keep Angular rich and ergonomic and at the same time be mindful about its scope and learning journey. If you think your request could live outside Angular's scope, we'd encourage you to collaborate with the community on publishing it as an open source package.

You can find more details about the feature request process in our documentation.

[pkozlowski-opensource]

There are 2 related points discussed here:

• making the async pipe working with any Subscribable and not just Observable (in other words - making the async pipe more generic) - this was recently implemented in feat(common): allow any Subscribable in async pipe #39627 and after this change I can personally use the async pipe with non-RxJS subscribable (ex. ones from https://github.com/AmadeusITGroup/ngx-tansu)
• working with any reactive platforms - I believe that we should a separate / dedicated pipe of MobX and other approach instead of trying to evolve the async pipe into a "handle it all" but bloated class.

Going to close this one as the issue is partially resolved and we don't want to put too many responsibilities into the async pipe.

[angular-automatic-lock-bot]

This issue has been automatically locked due to inactivity.
Please file a new issue if you are encountering a similar or related problem.

Read more about our automatic conversation locking policy.

_{This action has been performed automatically by a bot.}