RxJs - 将数据推送到 observable

Question

我想在 Angular 中实现一个无限表，所以当我向下滚动到现有表时，我想从 API 获取下一组数据。我正在尝试使用BehaviorSubject 来执行此操作，如下所示：

page: number = 0;
size: number = 10;
clientsData$: BehaviorSubject<any> = new BehaviorSubject<any>([]);

onTableScroll(e): void {
    // get the next set of data
    this.getData();
}

getData(): void {
    this.clientsService.getAllClients().pipe(
      tap((data) => {
        this.clientsData$.next(data);
        console.log('this.clientsData$: ', this.clientsData$);    // it never reaches this line
      })
    );
}

ngOnInit(): void {
    this.getData();
}

但问题是，即使我在模板中使用了async 管道，它似乎也没有订阅BehaviorSubject。所以在初始化时，这根本不起作用，我什至没有得到初始数据）。 每次调用getData() 并更新表格时，我都需要获取一组新数据。 我不想使用subscribe，而是在t 模板中使用async 管道。 请指教，谢谢！

Answer 1

它从来没有达到这个的原因是因为你还没有订阅它。只有当您订阅时，observable 才会“起飞”。

destructionSubject: Subject<any> = new Subject<any>();
getData(): void {
    this.clientsService.getAllClients().pipe(
      tap((data) => {
        this.clientsData$.next(data);
        console.log('this.clientsData$: ', this.clientsData$);    // it never reaches this line
      }),
      takeUntil(this.destructionSubject), // take until this destructionSubject emits
    ).subscribe(data => {
      console.log(data);
    });
}
...

ngOnDestroy(): void {
  this.destructionSubject.next(); // unsubscribe/destroy the subscription
}

Answer 2

您通过 Angular 的 aynsc 管道订阅 clientData$。这意味着clientData$ 发出的新值会被自动拾取。但是您永远不会向您的 behaviorSubject 发送新值 (next(value))，因此您当然不会看到它的任何数据。

在您的getData() 函数中，您创建一个可观察对象（您不运行它）然后返回。所以getData() 什么都不做。您可以将其重写为getData():void{}，这应该运行大致相同。那么应该很清楚，对clientData$.next(data) 的调用永远不会被执行。

那么你如何“解决”这个问题？好吧，创建一个 observable 就像定义一个函数，调用一个函数就像订阅一个 observable --- 你需要“运行”你在 getData() 中创建的 observable。 Subjects 既是 observables 又是观察者，你可以在 getData()getData() 订阅你的 subject 到你的 observable

page: number = 0;
size: number = 10;
clientsData$: BehaviorSubject<any> = new BehaviorSubject<any>([]);

onTableScroll(e): void {
    // get the next set of data
    this.getData();
}

getData(): void {
    this.clientsService.getAllClients().pipe(
      tap(data => 
        console.log('this.clientsData$: ', this.clientsData$)
      )
    ).subscribe(this.clientsData$);
}

ngOnInit(): void {
    this.getData();
}

这至少现在应该正在运行。它没有考虑错误处理。我还假设 clientsService.getAllClients() 返回一个将完成（或错误）的可观察对象 - 它不会永远保持运行 - 否则你需要一种方法来取消订阅或冒内存泄漏的风险。

Answer 3

您的方法可行，但通过以下方式实现可观察对象要容易得多：

• 想想什么是事件（这里是 onTableScroll）
• 为您的活动编写一个函数，然后将value 或void/nothing 放入主题中
• 在可观察对象中使用您的主题并将其与您的数据源结合起来
• 仅订阅 UI 中需要的数据或与某些服务（此处为 clientData$）进行通信的数据

const { Subject, Observable, merge } = rxjs;
const { first, switchMapTo } = rxjs.operators;

const tableScroll$$ = new Subject();
const clientData$ = getClientData$(tableScroll$$)

function onTableScroll(e) {
  tableScroll$$.next()
}

function getClientData$(scroll$) {
  // Observable for the ngOnInit getData$() call
  const initialData$ = getData$().pipe(first())

  // Observable for the scroll$ getData$() calls
  const scrollData$ = scroll$.pipe(
    switchMapTo(getData$())
  )

  return merge(
    initialData$,
    scrollData$
  )
}

function getData$() {
  return new Observable(obs => obs.next('foo'))
}

clientData$.subscribe(v => console.log('clientData$: ', v))

onTableScroll()

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>