RxJS Observable merge angular

Question

I have a sandbox which subscribes to a stream of messages and I want to filter that stream to find messages that have been sent to or received from a specific user using route params specified in another component.

messages.sandbox.ts:

messages$: Observable<Array<Message>> = this.store.select(state => state.data.messages);    

fetchReceived(id: string): Observable<Array<Message>> {
    return this.messages$.map((messages: any) => {
        return messages.filter((message: Message) => {
            return message.recipientId == id;
        });

    });
}

fetchSent(id: string): Observable<Array<Message>> {
    return this.messages$.map((messages: any) => {
        return messages.filter((message: Message) => {
            return message.userId == id;
        })
    })
}

messages.detail.container.ts

sentMessages$ = new Observable<Array<Message>>();
receivedMessages$ = new Observable<Array<Message>>();
matchingMessages$ = new Observable<Array<Message>>();



ngOnInit() {

    this.route.params.subscribe((params: Params) => {
        this.sentMessages$ = this.sb.fetchReceived(params['id']);
        this.receivedMessages$ = this.sb.fetchSent(params['id']);

        this.matchingMessages$ = Observable.merge(this.sentMessages$, this.receivedMessages$);

    });
}

this.matchingMessages$ seems to only include this.receivedMessages$ however I know that this.sentMessages$ is not null as I can use it in my template without a problem.

Am I missing something with merging Observables? Would it be better to create a single fetchMessages method which filters for either the userId or recipientId equalling the route param id? If so how would I go about that?

Thanks!!

Brandon Brandon · Accepted Answer · 2017-04-07T15:54:09

You have the right general idea. Just a few flaws.

Never use new Observable<T>(). It does not do what you think it does. It pretty much does not do anything useful. Always construct observables from factory methods or other observables
You need to transform the params observable into a new observable using an operator. Your problem is you subscribe to the params observable, and then construct new observables each time. But other code will have already subscribed to the initial observables so they will never see the changes.

So you want to do something like this:

sentMessages$ : Observable<Array<Message>>;
receivedMessages$ : Observable<Array<Message>>;
matchingMessages$ : Observable<Array<Message>>;

ngOnInit() {
    const params$ = this.route.params;

    // use switchMap to map the new params to a new sent observable
    // each time params change, unsubscribe from the old fetch and subscribe
    // to the new fetch.  Anyone subscribed to "sentMessages" will see the
    // change transparently
    this.sentMessages$ = params$.switchMap((params: Params) => this.sb.fetchReceived(params['id']));
    // same for received
    this.receivedMessages$ = params$.switchMap((params: Params) => this.sb.fetchSent(params['id'])));

    // merge the 2 streams together
    this.matchingMessages$ = Observable.merge(this.sentMessages$, this.receivedMessages$);
}

Edit:

to answer your other question: is it better to create a single observable that matches senders and receivers: depends upon your use case. But here is how you could go about it:

messages.sandbox.ts:

fetchEither(id: string): Observable<Array<Message>> {
    return this.messages$.map((messages: any) => {
        return messages.filter((message: Message) => {
            return message.recipientId == id || message.userId === id;
        });

    });
}

container:

matchingMessages$ : Observable<Array<Message>>;

ngOnInit() {
    const params$ = this.route.params;

    // use switchMap to map the new params to a new either observable
    // each time params change, unsubscribe from the old and subscribe
    // to the new fetch.  Anyone subscribed to "matchingMessages" will see the
    // change transparently
    this.matchingMessages$ = params$.switchMap((params: Params) => this.sb.fetchEither(params['id']));
}

RxJS Observable merge angular

1 Answers