RxJS execution order after subject emission

Question

I want to understand the code execution order following a 'next' call on a Subject.

Background: I have 3 classes (call them HaveSubject, HaveSubscription1, HaveSubscription2). HaveSubject needs to tell HS1 and HS2 to do something through a Subject that HS1 and HS2 are subscribed to. Their tasks must be completed before HaveSubject goes on to execute method somethingVeryImportant.

Pseudocode:

class HaveSubject {
    // Angular service
    public mySubject$: Subject<string> = new Subject<string>();

    public codeImExecuting() {
        this.mySubject$.next('input for tasks')
        this.somethingVeryImportant();
    }

    private somethingVeryImportant() {
        // stuff
    }
}

class HaveSubscription1 {
    // Angular service
    constructor(private hs: HaveSubject) {
        this.hs.mySubject$.subscribe(inputStr => {
            // did I do this before somethingVeryImportant is called?
        });
    }
}

class HaveSubscription2 {
    // Angular service
    constructor(private hs: HaveSubject) {
        this.hs.mySubject$.subscribe(inputStr => {
            // did I do this before somethingVeryImportant is called?
        });
    }
}

My question is: what is the best way to ensure that HS1 and HS2 have executed the code attached to their subscriptions before going on to execute method somethingVeryImportant? If the order of operations is: HaveSubject calls 'next' on subject -> HS1 and HS2 do their tasks -> HaveSubject goes on to its next line of code, which is somethingVeryImportant, then I have no issues. I'm just not sure that subscriptions are executed immediately after they receive the 'next' item in the subscription.

NOTE: There are a few things I can't do that I would normally, such as have HaveSubject inject into the other two, because the other two are created dynamically (i.e. I may have none, one, or both of HaveSubscriptionX, not clear how many will be created, and these are Angular services that are provided by a component, not in the root...).

Thoughts?

Answer 1

so this looks like you've made some suspect architecture decisions, since the flow is

1. service 1 function executes and emits from observable
2. service 2/3/etc subscribers to service 1 observable execute code that produces some output or side effect
3. service 1 function needs to execute dependent on some output or side effect from service 2/3/etc functions

this requires the subject to be aware of it's observers, which is the opposite of the rxjs philosophy. The ideal solution here would be to fix these architectural concerns. It's tough to say how to accomplish this without knowing more about how or why things occur this way or what the overall goal is.

However, you can accomplish this in a pretty reliable way, you need to add some subject on your first function that can signal completion by the dependent services:

class HaveSubject {
    // Angular service
    public mySubject$: Subject<string> = new Subject<string>();
    private workDone$: Subject<void> = new Subject<void>();

    imDone() {
        this.workDone$.next();
    }

    public codeImExecuting() {
        if (this.mySubject$.observers.length) { // check for observers
            // if observers, listen for that many emissions from workDone$, only reacting to the last one
            this.workDone$.pipe(take(this.mySubject$.observers.length), last())
                          .subscribe(v => this.somethingVeryImportant());
        } else { // or just run the function, doesn't matter
            this.somethingVeryImportant();
        }
        this.mySubject$.next('input for tasks');

    }

    private somethingVeryImportant() {
        // stuff
    }
}

and have the observers of mySubject$ call imDone() when they're done.

Answer 2

Simplest call events on finishing side work in HaveSubscription# (not ideal a lot duplicated run check second option)

class HaveSubject {
    // Angular service
    public mySubject$: Subject<string> = new Subject<string>();
    public mySubjectDone$: Subject<void> = new Subject<void>();

    public constructor() {
        this.mySubjectDone$.subscribe(this.somethingVeryImportant.bind(this));
    }

    public codeImExecuting() {
        this.mySubject$.next('input for tasks')
    }

    private somethingVeryImportant() {
        // stuff
    }
}

class HaveSubscription1 {
    // Angular service
    constructor(private hs: HaveSubject) {
        this.hs.mySubject$.subscribe(inputStr => {
            // did I do this before somethingVeryImportant is called?
            hs.mySubjectDone$.next()
        });
    }
}

class HaveSubscription2 {
    // Angular service
    constructor(private hs: HaveSubject) {
        this.hs.mySubject$.subscribe(inputStr => {
            // did I do this before somethingVeryImportant is called?
            hs.mySubjectDone$.next()
        });
    }
}

or if you dont want to require any actions from HaveSubscription# trigger delayed action

class HaveSubject {
    // Angular service
    public mySubject$: Subject<string> = new Subject<string>();

    public constructor() {
        this.mySubject$.pipe(
            debounceTime(16) // delay or auditTime, debounceTime, ...
        ).subscribe(this.somethingVeryImportant.bind(this));
    }

    public codeImExecuting() {
        this.mySubject$.next('input for tasks')
    }

    private somethingVeryImportant() {
        // stuff
    }
}

class HaveSubscription1 {
    // Angular service
    constructor(private hs: HaveSubject) {
        this.hs.mySubject$.subscribe(inputStr => {
            // did I do this before somethingVeryImportant is called?
        });
    }
}

class HaveSubscription2 {
    // Angular service
    constructor(private hs: HaveSubject) {
        this.hs.mySubject$.subscribe(inputStr => {
            // did I do this before somethingVeryImportant is called?
        });
    }
}

If you have some versioning mechanism that will react to changes done in HaveSubscription# you can do this:

this.mySubject$.pipe(
  map(this.calculateVersion),
  distinctUntilChanged(),
).subscribe(this.somethingVeryImportant.bind(this));