RxJS: Correct way to manually emit an Observable

Question

Working with RxJS in Angular 4.x, I'm seeing two very different patterns for generating Observables from streams of user initiated actions. One stream is the direct result of a user clicking an 'add item' button that generates a new object. The other is a series of events issued by some third party code I'm using.

I want to be able to combine these two streams using something like 'combineLatest' to generate a single Observable.

With my button, I've followed the following pattern:

const signal = Observable.create(
            (observer) => {
                this.additem= (item) => observer.next(item);
            }
        );

this.item$ = signal.map((item) => [item])
                            .scan((accumulator, value) => {
                                return accumulator.concat(value);
                            });

However, I'm seeing a lot of information saying I should be using Subjects instead - which I am trying to use with my event callback like so:

sort$ = new Subject();

sortChange(sort){
        sort$.next(sort);
}

Then I'm attempting to combine these like this:

combine$ = Observable.combineLatest(sort$, item$,
                  (sort, items) => {
                      return "something that does stuff with these";}
        );

My questions are - what is the preferred pattern for 'manually' generating streams? Can/should observables and subjects be meshed together into a single observable like I'm trying to do here?

I think that the best source of information for this is the answer to this question: https://stackoverflow.com/questions/39494058/angular-2-behavior-subject-vs-observable Even if your question is not exactly a duplicate to this one to me, it feels like this answer might be what you need to make your choice. — Supamiu, Jun 27 '17 at 14:41
If you want combine two Observable you can use the operator merge — alehn96, Jun 27 '17 at 14:42
I'm sorry i've been looking at your code for 2 hours now... and i am not able to understand what is ```this.item``` actually doing here?? How are you calling that function? from what i can see, you're probably trying to pass the item you create on button click to the observable? — Samarth Saxena, Aug 05 '22 at 22:13

martin · Accepted Answer · 2017-06-27T16:11:57.110

Of course you can combine Observables and Subjects into one stream.

I think the question here is what makes more sense in you usecase. From your description when implementing something like "add item" functionality I'd prefer Subject over Observable.create.

This is because every time you subscribe to your signal you're reassigning this.additem. The callback to Observable.create is called for every observer. Note that more correct usage of Observable.create would look like this:

const signal = Observable.create((observer) => {
   this.additem = (item) => observer.next(item);
   return () => this.additem = null;
});

The returned callback () => this.additem = null is called when you unsubscribe from this Observable and that's the place where you should handle all cleanup.

However, if you make two subscriptions to signal then you'll override this.additem twice and then if you chose to unsubscribe one of the observers you would this.additem = null and it would probably lead to an unexpected behavior.

So in this case it makes more sense to use Subject. For example like this:

const subject = new Subject();
this.additem = (item) => subject.next(item);

If you want to see more real life example of Observable.create have a look at for example this: Subscribe to a stream with RxJS and twitter-stream-api module

Edit: Also have a look at these articles from the lead developer of RxJS 5:

Thanks! This is the kind of answer I was looking for. Pitfalls like this weren't apparent to me looking at the documentation, and I knew there had to be reasons to choose one approach over the other since both seemed to work. — Potatoes, Jun 27 '17 at 16:09
@Potatoes I added a few links to articles I consider very helpful — martin, Jun 27 '17 at 16:12

RxJS: Correct way to manually emit an Observable

1 Answers1