Angular14+RxJS7中pipe无值传递问题排查及解决

问题场景 #

使用Angular 14.2.0 + RxJS 7.5.0开发时，遇到如下问题：

• 直接订阅from(detectEthereumProvider())能正常打印出provider值
• 对该Observable调用pipe()并搭配tap、switchMap操作符后，控制台无任何输出，操作符完全不触发

问题代码 #

import { Injectable } from '@angular/core';
import { BehaviorSubject, tap, from, switchMap, Observable } from 'rxjs';
import { PreviewModel } from '../models/preview.model';
import { FirebaseService } from './firebase.service';
import Web3 from 'web3';
import detectEthereumProvider from '@metamask/detect-provider';

@Injectable({
  providedIn: 'root'
})
export class CheckoutService {

  private isLoadingSubject = new BehaviorSubject<boolean>(false);

  private bookPreviewSubject = new BehaviorSubject<PreviewModel | null>(null);
  private web3: any;

  constructor(private firebaseService: FirebaseService) {
    this.web3 = (<any>Web3);
  }

  getBookPreviewSubject(): Observable<PreviewModel | null> {
    return this.bookPreviewSubject
  }

  getLoadingSubject(): Observable<boolean> {
    return this.isLoadingSubject
  }

  fetchBookPreview(bookHash: string) {
    this.isLoadingSubject.next(true);
    return this.firebaseService
      .getPreview(bookHash)
      .then(response => this.bookPreviewSubject.next(response))
      .finally(() => this.isLoadingSubject.next(false));
  }

  public createAndSendTX(bookPreview: PreviewModel, priceInEth = 0.001) {

    const providerSource = from(detectEthereumProvider());

    // 直接订阅能正常打印：[object Object]
    // providerSource.subscribe(val => console.log("subscribe", val));

    // 此处pipe无任何日志输出
    return providerSource.pipe(
      tap((provider: any) => console.log("tap", provider)),
      switchMap(async (provider) => {
        console.log("switchMap", provider);
        if (!provider) {
          throw new Error('Please install MetaMask');
        }
        return await provider.request({ method: 'eth_requestAccounts' });
      }),
    );
  }
}

原因分析 #

RxJS中的冷Observable（from()创建的Observable属于此类）具有惰性：只有当存在订阅者（调用subscribe()方法）时，才会启动数据流的执行逻辑。

问题代码中仅通过pipe()转换得到了一个新的Observable，但并未对其进行订阅，因此整个数据流处于未激活状态，tap、switchMap等操作符自然不会执行。而直接订阅原始的providerSource时，因为存在订阅者，所以数据流正常触发并输出值。

解决代码 #

只需对pipe()返回的Observable调用subscribe()即可激活数据流：

return providerSource.pipe(
      tap((provider: any) => console.log("tap", provider)),
      switchMap(async (provider) => {
        console.log("switchMap", provider);
        if (!provider) {
          throw new Error('Please install MetaMask');
        }
        return await provider.request({ method: 'eth_requestAccounts' });
      }),
    ).subscribe(val => console.log("createAndSendTX", val));

内容的提问来源于stack exchange，提问作者Asalvatore