CODEWITHSUNDEEP
angularrxjs
srunner
srunner

Reputation: 31

How to combine two Observables into one Observable Array using RxJs

I am trying to combine two Observables into a single Array of Observables that my Angular App can subscribe to and receive values over time. The scenario is using a chat bot. The user provides input, the bot responds and the two are combined in an array for the chat flow.

See the below:

Chat input component to get the user input:

  onNewComment(value: string): void {
    this.chatService.addComment(value);
  }

This then speaks to chat.service.ts

import { Injectable, Output } from '@angular/core';
import { Observable, BehaviorSubject, Subject, combineLatest, from, merge, concat } from 'rxjs';
import { environment } from '../../environments/environment';
import { HttpClient, HttpHeaders } from '@angular/common/http';
import { UserInput } from '../interfaces/user-input';
import { map, tap, switchMap, mergeMap, toArray, scan, shareReplay, take } from 'rxjs/operators';
import { of } from 'rxjs';
import { text } from '@angular/core/src/render3';
import { Division } from '../interfaces/division';


const headers = new HttpHeaders();
headers.append(
    'Content-Type', 'application/json'
);

@Injectable({
  providedIn: 'root'
})
export class ChatService {
  apiUrl = environment.apiUrl;
  conversationSubject: BehaviorSubject<any[]> = new BehaviorSubject<any>([]);
  conversationArray$: Observable<any> =  this.conversationSubject.asObservable();
  private commentAddedSubject = new Subject<any>();
  commentAddedAction$ = this.commentAddedSubject.asObservable();


  sessionToken$ = this.http.get<any>(`${this.apiUrl}/api/session`).pipe(
    map(data => data.session_id),
    tap(data => this.addComment('')),
    shareReplay(1)
  );

  commentWithSessionToken$ = combineLatest([
    this.sessionToken$,
    this.commentAddedAction$
  ]).pipe(
    map(([sessionToken, comment]) => {
      return {
        session_id: sessionToken,
        input: {
          text: comment
        },
        };
      }
    ),
  );


  botResponse$ = this.commentWithSessionToken$
    .pipe(
      switchMap(comment =>
        of(comment)
          .pipe(
            mergeMap(data => this.http.post<any>(`${this.apiUrl}/api/message`,
              {
                userInput: data
              },
              {
                headers: headers
              })
              ),
            )
          )
      );

  addComment(newComment?: string) {
    this.commentAddedSubject.next(newComment);
  }

  constructor(
    private http: HttpClient,
  ) { }
}

I have tried a number of ways to combine commentWithSessionToken$ and botResponse$ into an array of user input and bot response but not having any luck. The end result I need is something like:

[commentWithSessionToken$, botResponse$, commentWithSessionToken$, etc, etc]

Any advice would be much appreciated. Thank you.

Upvotes: 0

Views: 3501

Answers (2)

MoxxiManagarm
MoxxiManagarm

Reputation: 9124

I think you should merge your observables and scan them

chat$ = merge(this.commentWithSessionToken$, this.botResponse$).pipe(
  scan((acc, curr) => ([...acc, curr]), []),
);

Upvotes: 0

Prashant
Prashant

Reputation: 383

Please use rxjs forkJoin

forkJoin(commentWithSessionToken$, botResponse$, commentWithSessionToken$, etc, etc)

Return type is observable<[commentWithSessionToken$, botResponse$,...]>

Upvotes: 0

Related Questions