如果观察不到，则共享订阅

问题描述

我有一个角度应用程序，在某些情况下（软件中的某些触发器或在用户请求时）应与服务器同步某些数据。所以我有这样的功能：

    ...

    public createSyncObservable(): Observable<any> {
        return this.retriveDataFromStorage().pipe(
            switchMap(
                (data) => forkJoin(this.api.sendData1(data.data1),this.api.sendData2(data.data2),this.api.sendData3(data.data3))
            ),switchMap(
                (data) => this.api.getDataFromServer()
            ),switchMap(
                (data) => this.updateLocal(data)
            )
        )
    }

我想要的行为是：

• 如果用户（或某些触发器）请求同步并且已经在进行同步，则我不应该再次执行同步，只需等待当前同步结束并返回相同的可观察（共享）即可。
• 如果上一次同步已完成，则应重新开始（创建一个新的可观察对象）。

我目前最好的解决方案是执行以下操作（未经测试的代码）：

    ...
    public syncData(): Observable<any> {
        if (this.observable_complete) {
            this.observable_complete = false;
            this.syncObservable$ = this.createSyncObservable().pipe(share())
            this.syncObservable$.subscribe(
                (data) => {this.observable_complete = true}
            ) 
        }
        return this.syncObservable$;
    }

这是要走的路吗？也许我缺少一些RxJS运算符，在这种情况下对我有帮助？这个解决方案似乎有点hacky ...

解决方法

如果调用this.createSyncObservable()并没有做任何实际的工作，而仅订阅它返回的可观察对象，则只需调用一次该函数。然后，您可以简单地执行以下操作：

public syncData$ = this.createSyncObservable().pipe(share());

如果没有订阅者（例如，share完成时），

this.createSyncObservable()将从其来源退订。因此，this.syncData$的订阅者将完成对this.createSyncObservable()返回的可观察对象的订阅。

// The first subscribe will trigger a subscribe to createSyncObservable()
syncData$.subscribe()
// A second subscribe while the first hasn't completed won't trigger a subscribe
// to createSyncObservable() but instead just wait for its response
syncData$.subscribe()

// After some time ...
// Another subscribe after createSyncObservable() completed will trigger another 
// subscribe to createSyncObservable()
syncData$.subscribe()

https://stackblitz.com/edit/rxjs-44qzaj?file=index.ts

,

您应该尝试使用takeWhile / skipWhile运算符：

这只是一个例子，但我希望您觉得它有用。

private isResourceFree = true;

public createSyncObservable(): Observable<any> {

        return of(isResourceFree).pipe(
            takeWhile(val => val),tap(_ => this.isResourceFree = false),switchMap(_ => this.retriveDataFromStorage()),switchMap(
                (data) => forkJoin(this.api.sendData1(data.data1),this.api.sendData2(data.data2),this.api.sendData3(data.data3))
            ),switchMap(
                (data) => this.api.getDataFromServer()
            ),switchMap(
                (data) => this.updateLocal(data)
            ),tap(_ => this.isResourceFree = true)
        )
    }