Angular rxjs private subject + public observable setup

Question

I'm new to rxjs/observables. They're a bit over my head so this may be a simple question.

I want to have a subject in a service which can be frequently triggered to get updates. The update could then be immediately returned to the caller (if they wait), but also emitted to any subscribers. I'd like the subject to be private so only the service can .next() it.

The last caveat is, when something subscribes to the observable I'd like to get the current value ONCE, and then be alerted to any updates after that. My rookie attempt is below. It seems to mostly work, but the page subscription gets the current value once per call of next() on initial subscribe. So if getNewThings() was called 5 times, the page subscription would immediately get the current subject value 5 times. How can I fix that?

Service

@Injectable()
export class PretendService{
    private mySubject: BehaviorSubject<Thing[]> = new BehaviorSubject<Thing[]>(null);
    public readonly mySubjectObservable: Observable<Thing[]> = this.mySubject.asObservable();
    ...
    public getNewThings(): Promise<Thing[]>{
        let p = new Promise<Thing[]>((resolve, reject) => {

            //Do the work to get Thing[]s
            let theThings: Thing[] = [];
            mySubject.next(theThings);
            resolve(theThings);
        }

        return p;
    }
}

Page

...
thingSubscription: Subscription;
constructor(private pretendService: PretendService){
    this.thingSubscription = this.pretendService.mySubjectObservable.subscribe((things)=>{
        //ISSUE HERE: 
        //This fires once per .next() with the current value on initial subscription
        console.log(`got some things [{things}]`);
    });
}
...
//Unsubscribe in destructor...etc...

UPDATE

Here is a stackblitz that shows some of this. Clicking the main button will trigger the refresh method multiple times. Then click the "Other Page" link to trigger the subscription. Notice on the Page2 component there is an OnDestroy implementation that can be commented/uncommented. That was my main issue - I wasn't destroying properly, so it was gathering subscriptions. https://stackblitz.com/edit/angular-umk8qm?embed=1&file=src/app/page2.component.ts

I'd like to get the current value ONCE, and then be alerted to any updates after that: isn't that exactly what you get: you get the current value, and are also notified every time a new event is emitted. Unclear what you get and what you expect to get instead. Give a concrete example. — JB Nizet
Not quite. Let's say other components called getNewThings() 5 times, and the current list has 2 items in it. When this page sets up a new subscription the subscribe code would fire 5 times with the current list of 2 items. I'd like it to just fire one time with the 2 items on initial setup (and then be told of new changes after that). — BRass
You'll have to reproduce that in a StackBlitz. A BehaviorSubject only emits its current value, not the last 5 values. — JB Nizet
As @JBNizet has noted, a BehaviorSubject emits the current value for the initial subscription and then new values emitted after that. You can read more about it here: reactivex.io/rxjs/manual/overview.html#behaviorsubject — Daniel W Strimpel
Thanks guys - glad I'm not crazy. I'll see if I can recreate this in a StackBlitz. In general, is the pattern above "normal"? I'm trying to find a good baseline to use for similar activity. — BRass

BRass BRass · Accepted Answer · 2018-11-02T19:39:39

See the StackBlitz on the update. In my case I was not implementing OnDestroy properly, which caused my app to accumulate subscriptions. This looked like it was getting 1 update per next() call on the BehaviorSubject. However, it was just getting 1, but since there were lots of subscriptions building up it appeared like 1 subscription was getting multiple updates. A more complete implementation is below.

Page

export class Page2Component implements OnDestroy
...
thingSubscription: Subscription;
constructor(private pretendService: PretendService){
    this.thingSubscription = this.pretendService.mySubjectObservable.subscribe((things)=>{
        console.log(`got some things [{things}]`);
    });
}
...
ngOnDestroy() {    
    //This should fire if things are destroying properly
    console.log('destroying!');
    this.thingSubscription && this.thingSubscription.unsubscribe();
}
....

Angular rxjs private subject + public observable setup

1 Answers