返回 observable 但等到另一个完成

Question

有人要求我向遗留代码添加功能。遗留代码处理带有一些附加功能的 get 请求，例如在 PWA 离线时避免重复调用和缓存 get 请求。

在这个 PWA 的上下文中，当有一些 POST/PUT/DELETE 操作时，我想在执行任何 get 之前提交。 我做了一个可观察的 $processingOfflineOperation ，它返回真或假，我想延迟任何获取操作，直到 $processingOfflineOperation 接近假。

这是get方法：

public get(endpoint: string, params?: any, avoidCache?: boolean, reqOpts?: any, ignoreHTTPStatusCode?: HttpStatusCode[]): Observable<any> {
    const pendingObs = this.checkIfPendingToAvoidDuplicateCalls('GET ' + endpoint, params);
    if (!!pendingObs) {
      // return duplicate call
      return pendingObs;
    }
    if (!reqOpts) {
      reqOpts = {
        params: new HttpParams()
      };
    }

    let header = new HttpHeaders({Authorization: this.getAuthorizationHeaderValue()});
    header = header.set('Content-type', 'application/json');
    reqOpts.headers = header;
    endpoint = (endpoint) ? endpoint : '';
    if (!this.network.online && !avoidCache) {
      // return from cache
      return of(this.cacheCalls['GET ' + endpoint + ' ' + JSON.stringify(params)]);
    } else if (!this.network.online) {
      return throwError(() => new Error('No internet'));
    }
    const $obs = this.http.get(this.url + endpoint + '?' + this.JSON_to_URLEncoded(params), reqOpts).pipe(
      catchError((err) => {
        this.clearPendingCall('GET ' + endpoint, params);
        if (!!ignoreHTTPStatusCode && !!ignoreHTTPStatusCode.find((status) => status === err.status)) {
          return of(null);
        }
        this.manageError(err);
        return throwError(err);
      }),
      map(res => {
        this.clearPendingCall('GET ' + endpoint, params);
        this.addCacheCall('GET ' + endpoint + ' ' + JSON.stringify(params), res);
        return res;
      }),
      share()
    );
    this.setPendingCall('GET ' + endpoint, params, $obs);
    return $obs;
  }

我不知道如何继续我所有的尝试都改变了 get 方法的行为 我最好的猜测是这样，但这并不像预期的那样工作：

public get(endpoint: string, params?: any, avoidCache?: boolean, reqOpts?: any, ignoreHTTPStatusCode?: HttpStatusCode[]): Observable<any> {


    const $subject = new Subject();
    this.$processingOfflineOperation.subscribe(isProcessing => {
      if (isProcessing) {
        return ;
      }

      const pendingObs = this.checkIfPendingToAvoidDuplicateCalls('GET ' + endpoint, params);
      if (!!pendingObs) {
        // return duplicate call
        $subject.next(pendingObs);
      }
      if (!reqOpts) {
        reqOpts = {
          params: new HttpParams()
        };
      }

      let header = new HttpHeaders({Authorization: this.getAuthorizationHeaderValue()});
      header = header.set('Content-type', 'application/json');
      reqOpts.headers = header;
      endpoint = (endpoint) ? endpoint : '';
      if (!this.network.online && !avoidCache) {
        // return from cache
        $subject.next(this.cacheCalls['GET ' + endpoint + ' ' + JSON.stringify(params)]);
      } else if (!this.network.online) {
        $subject.error(new Error('No internet'));
        return throwError(() => new Error('No internet'));
      }
      const $obs = this.http.get(this.url + endpoint + '?' + this.JSON_to_URLEncoded(params), reqOpts).pipe(
        catchError((err) => {
          this.clearPendingCall('GET ' + endpoint, params);
          if (!!ignoreHTTPStatusCode && !!ignoreHTTPStatusCode.find((status) => status === err.status)) {
            return of(null);
          }
          this.manageError(err);
          $subject.error(err);
          return throwError(err);
        }),
        map(res => {
          this.clearPendingCall('GET ' + endpoint, params);
          this.addCacheCall('GET ' + endpoint + ' ' + JSON.stringify(params), res);
          $subject.next(res);
        }),
        share()
      );
      this.setPendingCall('GET ' + endpoint, params, $obs);
      $obs.subscribe();
    });
    return $subject;
  }

在@dmance 评论之后，我尝试将第一个版本的 get 重命名为 getReal 并创建了一个新函数“get”，然后我这样做了：

  public get(endpoint: string, params?: any, avoidCache?: boolean, reqOpts?: any, ignoreHTTPStatusCode?: HttpStatusCode[]): Observable<any> {

    return this.getReal(endpoint, params, avoidCache, reqOpts, ignoreHTTPStatusCode).pipe(
  filter(() => !this.isProcessing), 
  take(1)
);
  } 

但仍然没有机会， 那么这个：

public get(endpoint: string, params?: any, avoidCache?: boolean, reqOpts?: any, ignoreHTTPStatusCode?: HttpStatusCode[]): Observable<any> {
    const $subject = new Subject();
    this.getReal(endpoint, params, avoidCache, reqOpts, ignoreHTTPStatusCode).pipe(
      skipUntil(this.$processingOfflineOperation.pipe(filter(isProcessing => {
        return isProcessing === false;
      })))
    ).subscribe({
      next: (v) => {
        $subject.next(v);
      },
      error: (err) => {
        $subject.error(err);
      }
    });
    return $subject;
  }

也不行。

不确定这是不是最优雅的方法，但这个方法有效：

 public get(endpoint: string, params?: any, avoidCache?: boolean, reqOpts?: any, ignoreHTTPStatusCode?: HttpStatusCode[]): Observable<any> {
    const $subject = new Subject();
    this.$processingOfflineOperation.subscribe((isProcessing) => {
      if (!isProcessing) {
        this.getReal(endpoint, params, avoidCache, reqOpts, ignoreHTTPStatusCode).subscribe({
          next: (v) => {
            $subject.next(v);
          },
          error: (err) => {
            $subject.error(err);
          }
        });
      }
    });
    return $subject;
  }

Answer 1

不确定这是最优雅的方法，但这个方法有效：

 public get(endpoint: string, params?: any, avoidCache?: boolean, reqOpts?: any, ignoreHTTPStatusCode?: HttpStatusCode[]): Observable<any> {
    const $subject = new Subject();
    this.$processingOfflineOperation.subscribe((isProcessing) => {
      if (!isProcessing) {
        this.getReal(endpoint, params, avoidCache, reqOpts, ignoreHTTPStatusCode).subscribe({
          next: (v) => {
            $subject.next(v);
          },
          error: (err) => {
            $subject.error(err);
          }
        });
      }
    });
    return $subject;
  }

优雅

您应该能够在没有中间主题的情况下重写它。我不能为你测试这个，但这样的东西应该可以工作：

public get(
  endpoint: string, 
  params?: any, 
  avoidCache?: boolean, 
  reqOpts?: any, 
  ignoreHTTPStatusCode?: HttpStatusCode[]
): Observable<any> {
  return this.$processingOfflineOperation.pipe(

    mergeMap(isProcessing => isProcessing ? EMPTY : 
      this.getReal(endpoint, params, avoidCache, reqOpts, ignoreHTTPStatusCode)
    )

  );
}

您会注意到，由于不管理自己的订阅，代码会变得更短/简洁。一般来说，嵌套的subscribe 块是代码异味。通常有更好的方法。

Answer 2

试试这个：

forkJoin(pendingOperations)
  .pipe(
    switchMap((_) => yourGetOperation())
  ).subscribe(getResults => console.log(getResults));