How to send to a Subject different Observable results without having to break the observable flow subscribing?

Question

Edited the question with a concise example. I think it'll make it easier for people to understand. Take note this example is super simplified, normally I layer things between different components, but for the question it will suffice.

Take this component. It takes the name of a fetched object, and a button to fetch the next object. To get the value of the next REST request, I know of no other way than to subscribe to the answer, and what I would like is to have something like "combineLatest" but for "the future", so I can combineLatest of later streams.

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

Is there any way to do that? Doing something like "emitIn(characterSubject)". I think there is nothing like this, like dynamically add source emissions to a source.

Answer 1

If I understand right, you have a service which has some method that triggers an http call, like dataRepository.fetch(id), and you have different components that need to react when the response of the call arrives.

If this is the case, there are many ways to deal with such requirement, one being to use Subjects exposed as public properties of the service, which is what I understand you want to do.

To achieve this behavior the code you have written is what you need, in other words this is OK

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

If you want to make it more complete and manage also the error and complete case, you may write it like this

dataRepository.fetch(id)
  .subscribe(currentData)

In this last form, you are passing currentData as the Observer parameter of subscribe. currentDatathough is also an Observable and therefore can next, error and complete.

If you use ReplaySubject you can add the possibility to store the last results and present them to components which are created after the notification of the result.

Answer 2

Original component retrieved observable is dynamically changed, but that would be really dangerous (any pipe could break any observable). I would need another thing not called pipe.

I disagree. you can easily create dynamic observables using functions that return an observable based on your Input data and It will no break.

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

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

but If you think If something breaks inside you can simply use 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=>{...});

We use mergeMap simply because we need first Observable's Data to send second request.

If what you are looking is something to mix these observables from base then It would make all of rxjs operators useless because that's what they are here for.

But If you don't want to emit data to your currentData Observable and you know your data source then you don't even need to create separate Observable for it. Just using pipe to get the data is enough.

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)
  )

Or even when you need to call another Observable by dataRepository.fetch(id) response

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

But still If you want to go deeper and create a flow without passing data I think you will get disappointed because It's How Observables in general work. and I guess this approach is very limited and hard to maintain because you finally reach a point where you need to send different requests to multiple API and gather data and that's when flow will break. 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();
});

I hope it helps but let me know If it's not what you are looking for.