Best way to ensure subscription happened before emitting next value

Question

I'm using OAuth 2.0 to retrieve a Token for my User in my Angular Application. I use the popular lib angular-oauth2-oidc for handling the OAuth Flow. So in my AppComponent I set up the oAuthService from angular-oauth2-oidc and then initiate the oAuth Flow in my Login Component by calling initCodeFlow(). The User is redirected to enter his credentials and afterwards he get's redirected to me and I recieve the token in the AppComponent and put it as next token into my DataService TokenSubject.

But now I want to make an API call with that token in the header in my MainPageComponent. So I subscribe in my MainPage to the TokenSubject and when I recieve the Token the code in the subscribe block get's executed.

But what if the token get's send to me quicker than the main page builds up? Then I'm not yet subscribed to the TokenSubject in the Main Page when the next Token value is emitted by the AppComponent. I'm not sure, but my main page sometimes did't open up and I think this is probably the reason.

What's the best way to make sure that the Main Page is subscribed to the TokenSubject, before I emit the TokenSubject?

My AppComponent:

import { OAuthService } from 'angular-oauth2-oidc';
...

export class AppComponent implements OnInit {
    token: Token = {
      access_token: null,
      refresh_token: null
};

constructor(private oauthService: OAuthService,
                private adapterService: AdapterService,
                private dataService: DataService,
                private logger: NGXLogger) {
}

ngOnInit() {
        this.configureOAuthService();
        this.subscribeToOAuthEvents();
    }

configureOAuthService() {
        const authConfig: AuthConfig = {
            issuer: environment.oauthLoginUrl,
            redirectUri: environment.oauthRedirectUrl,
            clientId: environment.apiKey,
            scope: 'openid',
            responseType: 'code',
            showDebugInformation: true,
            disablePKCE: false,
            oidc: true,
            disableAtHashCheck: false,
        };

        this.oauthService.configure(authConfig);
        this.oauthService.tokenValidationHandler = new JwksValidationHandler();
        this.oauthService.loadDiscoveryDocumentAndTryLogin();
}

subscribeToOAuthEvents() {
        this.oauthService.events.subscribe(e => {
            if (e.type === 'token_received' || e.type === 'token_refreshed' ) {
                this.storeTokenInDataService();

                const self = this;
                // this will wait for 2 hours minus 30 sec before it refreshes the access token;
                setTimeout(() => { self.refreshToken(self); }, 7_170_000);
            } else if (e.type === 'token_expires') {
                this.logger.log('token_expires, going to refresh');
                this.refreshToken(this);
            } else if (e instanceof OAuthErrorEvent) {
                this.logger.error('OAuth error:', e);
            } else {
                this.logger.log('OAuthEvent:', e);
            }
        });
    }

    storeTokenInDataService() {
        this.token.access_token = this.oauthService.getAccessToken();
        this.token.refresh_token = this.oauthService.getRefreshToken();
        this.dataService.nextToken(this.token);
    }

In my Main Page I subscribe to the sharedToken but I#m not sure if that always happens before the Token gets emitted:

ngOnInit(): void {
  console.log('going to get the token from dataService');
  this.dataService.sharedToken
    .pipe(
      takeUntil(this.destroy$),
      tap(token => {
        this.token = token;
        console.log('got Token');
      }),
      switchMap(token => {
        this.headers = this.createHeaders(token.access_token);

        return this.adapterService.getUserInfo(this.headers);
      }))
    .subscribe(UserInfo => {
      this.logger.log('received userInfo : ', userInfo);
      ... 
  });
 }
}

In my DataService it looks like this:

  private token = new Subject<Token>();
  sharedPToken = this.token ;

  nextToken(token: Token) {
    this.token.next(token);
  }

Update: The question was solved, my final Main Component looks like this:

ngOnInit(): void {
  this.dataService.sharedTokenReplaySubject
    .pipe(
      takeLast(1),
      tap(token => {
        this.token = token;
        console.log('got Token');
      }),
      switchMap(token => {
        this.headers = this.createHeaders(token.access_token);

        return this.adapterService.getUserInfo(this.headers);
      }))
    .subscribe(UserInfo => {
      this.logger.log('received userInfo : ', userInfo);
      ... 
  });
 }
}

Answer 1

In short: you can't ensure it with Subject.

You could use the RxJS ReplaySubject instead. It's a variant of the more general Subject you're using. However it can "hold/buffer" specific number of previous emissions and emit to future subscribers immediately upon subscription. In your case you could use buffer 1. It'll hold and emit the last pushed value pushed to it.

private token = new ReplaySubject<Token>(1);  // <-- buffer 1
sharedToken = this.token.asObservable();

nextToken(token: Token) {
  this.token.next(token);
}

This way you don't need to worry if the value is pushed to the token before or after the subscription to sharedToken. The subscription this.dataService.sharedToken will get the token even if the value got pushed to token before the subscription.