如何在不中断可观察流订阅的情况下向主题发送不同的可观察结果？

Question

用一个简洁的例子编辑了这个问题。我想这会让人们更容易理解。注意这个例子是超级简化的，通常我在不同的组件之间分层，但是对于这个问题就足够了。

拿这个组件。它采用获取的对象的名称和获取下一个对象的按钮。为了获得下一个 REST 请求的值，我知道除了订阅答案之外别无他法，我想要的是像“combineLatest”这样的东西，但对于“未来”，这样我就可以组合最新的流.

import { Component, VERSION, OnInit } from '@angular/core';
import { BehaviorSubject, Observable } from 'rxjs';
import { HttpClient } from '@angular/common/http';

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
  
    private readonly PEOPLE_API_ENDPOINT = `https://swapi.dev/api/people/`;

   private characterSubject : BehaviorSubject<any> = new BehaviorSubject<any>({name: 'loading'});
  private currentCharacter: number = 1;
  
  character$ : Observable<any> = this.characterSubject.asObservable();
  
  constructor(
    private http: HttpClient
  ) {}
    
  ngOnInit() {
    this.updateCurrentCharacter();
  }

  nextCharacter() : void {
    this.currentCharacter ++;
    this.updateCurrentCharacter();
  }

  //I would want to avoid subscribing, instead
  //I would like some sort of operation to send the stream
  //emissions to the subject. As to not break the observable
  //chain up until presentation, like the best practices say.
  private updateCurrentCharacter() : void {
    this.fetchCharacter(this.currentCharacter)
      .subscribe( 
        character => this.characterSubject.next(character)
      );
  }

  private fetchCharacter (id: number) : Observable<any> {
        return this.http.get(this.PEOPLE_API_ENDPOINT + `${id}/`);
  }
}

<span>{{ character$ | async }} </span>

<button (click)="nextCharacter()">Next character</button>

Online demo

有没有办法做到这一点？做类似“emitIn（characterSubject）”的事情。我认为没有像这样动态地将源排放添加到源。

Answer 1

如果我理解正确，您的服务具有触发 http 调用的某种方法，例如 dataRepository.fetch(id)，并且您有不同的组件需要在调用响应到达时做出反应。

如果是这种情况，有很多方法可以处理这种要求，一种是使用作为服务的公共属性公开的主题，这是我理解你想要做的。

要实现此行为，您编写的代码就是您所需要的，换句话说，这是可以的

dataRepository.fetch(id)
  .subscribe(
    newData => currentData.next(newData)
  )

如果你想让它更完整并管理error和complete的情况，你可以这样写

dataRepository.fetch(id)
  .subscribe(currentData)

在最后一种形式中，您将currentData 作为subscribe 的观察者参数传递。 currentDatathough 也是一个 Observable，因此可以 next、error 和 complete。

如果您使用ReplaySubject，您可以添加存储最后结果并将它们呈现给在结果通知后创建的组件的可能性。

Answer 2

检索到的原始组件 observable 是动态更改的，但这真的很危险（任何管道都可能破坏任何 observable）。我需要另一个不叫管道的东西。

我不同意。您可以使用基于您的输入数据返回可观察对象的函数轻松创建动态可观察对象，并且不会中断。

function getUserFriendsById(userId: string): Observable<User[]>{
return service.getProfileById(userId).pipe(
  mergeMap(user=>{
    return getUsersByArrayOfIds(user.friends);
  })
 )
}

getUserFriendsById('exampleUserId').subscribe(friends=>{...});

但如果您认为如果内部出现问题，您可以简单地使用 CatchError。

function getUserFriendsById(userId: string): Observable<User[]>{
return service.getProfileById(userId).pipe(
  mergeMap(user=>{
    return getUsersByArrayOfIds(user.friends).pipe(
      catchError(err => {
      return of([]) // will return empty friend list if this APi call fails.
    }))
  }),
  catchError(err=>{
   return of([]); // will return empty friend list if this APi call fails.
  })
 )
}

getUserFriendsById('exampleUserId').subscribe(friends=>{...});

我们使用mergeMap只是因为我们需要第一个Observable的数据来发送第二个请求。

如果你正在寻找的东西是从基础中混合这些可观察的东西，那么这将使所有 rxjs 运算符无用，因为这就是它们的目的。

但是如果您不想将数据发送到您的 currentData Observable 并且您知道您的数据源，那么您甚至不需要为它创建单独的 Observable .只需使用管道获取数据就足够了。

currentData = dataRepository.fetch(id);
// at some point when you need currentData data you will subscribe to it.

// is same as 

dataRepository.fetch(id)
  .subscribe(
    newData => currentData.next(newData)
  )

甚至当您需要通过dataRepository.fetch(id) 响应调用另一个 Observable 时

dataRepository.fetch(id).pipe(
mergeMap(fetchedId=>{
    return getUserById(fetchedId);
  })
)

但是，如果您想更深入地创建流程而不传递数据，我认为您会感到失望，因为这就是 Observables 在一般情况下的工作方式。而且我想这种方法非常有限且难以维护，因为您最终会到达需要向多个 API 发送不同请求并收集数据的地步，这时流程将中断。 https://www.learnrxjs.io/learn-rxjs/operators/creation/create

// RxJS v6+
import { Observable } from 'rxjs';
/*
  Create an observable that emits 'Hello' and 'World' on  
  subscription.
*/
const hello = Observable.create(function(observer) {
  observer.next('Hello');
  observer.next('World');
  observer.complete();
});

希望对您有所帮助，但如果不是您要找的，请告诉我。