If a task is currently ongoing, be sure to subscribe to it; otherwise, restart it

Question

If a task is currently ongoing, be sure to subscribe to it; otherwise, restart it

If I have a long-running observable called longObservable, which emits a new string once after 5 seconds and then completes.

longObservable(): Subject<string> {
    return timer(5000).pipe{
        map(() => randomString())
    }
}

Other pages call this observable multiple times. If it's already running, I want to continue that process. If it's completed, I want to start it again.

longObservable.subscribe() // Starts the timer immediately

This is followed by another subscription two seconds later:

longObservable.subscribe() // Will get the same string as
                            // the previous subscription in 3 seconds.

Finally, a third subscription runs 20 seconds later:

longObservable.subscribe() // Initiates a new iteration and waits
                            // 5 seconds before emitting a new string.

The second subscription works as intended, but I am facing difficulties with the third one. It immediately emits the same value because longObservable has already completed.

This setup is used for geolocation requests on a device. The goal is to request a new location, or use the existing result if a request is already in progress.

Edit: Changed the observable to a subject for multicasting and removed the take(1) operator.

Edit2: https://stackblitz.com/edit/angular-venpk4 contains a working example of what I'm trying to achieve. I hope to accomplish this without using the timerRunning variable and with the help of RxJS operators. This code can be found under the hello component and prints results to the console.

angular typescript rxjs

Answer 1

Answer №1

Dealing with a challenging problem led me to come up with a solution showcased on StackBlitz. One of the crucial components in solving this issue is the clever use of the share() operator, which efficiently converts the observable into a subject without explicitly declaring one. The key concept here is creating a new subject each time the previous one completes, achieved by implementing a factory function to either deliver the existing shareable Observable (if longObservable() is still running) or assemble a fresh one.

Highlighted below are the essential sections from the StackBlitz:

let inProgress: boolean = false;

function longObservable(): Observable<string> {
    return timer(5000).pipe(
        map(() => randomString()),
        tap(() => inProgress = false),
        share()
    )
}

let obs$: Observable<string>;

function getLongObs(): Observable<string> {
    if (inProgress) {
        return obs$
    } else {
        inProgress = true;
        obs$ = longObservable();
        return obs$;
    }
}

console.log('initiate first subscribe');
getLongObs().subscribe(
    rand => console.log(`First subscribe returned ${rand}`)
);

setTimeout(() => {
    console.log('initiate second subscribe');
    getLongObs().subscribe(
        rand => console.log(`Second subscribe returned ${rand}`)
    );
}, 2000);

setTimeout(() => {
    console.log('initiate third subscribe');
    getLongObs().subscribe(
        rand => console.log(`Third subscribe returned ${rand}`)
    );
}, 7000)

I trust this explanation aids in understanding the solution!

Answer 2

Dealing with a challenging problem led me to come up with a solution showcased on StackBlitz. One of the crucial components in solving this issue is the clever use of the share() operator, which efficiently converts the observable into a subject without explicitly declaring one. The key concept here is creating a new subject each time the previous one completes, achieved by implementing a factory function to either deliver the existing shareable Observable (if longObservable() is still running) or assemble a fresh one.

Highlighted below are the essential sections from the StackBlitz:

let inProgress: boolean = false;

function longObservable(): Observable<string> {
    return timer(5000).pipe(
        map(() => randomString()),
        tap(() => inProgress = false),
        share()
    )
}

let obs$: Observable<string>;

function getLongObs(): Observable<string> {
    if (inProgress) {
        return obs$
    } else {
        inProgress = true;
        obs$ = longObservable();
        return obs$;
    }
}

console.log('initiate first subscribe');
getLongObs().subscribe(
    rand => console.log(`First subscribe returned ${rand}`)
);

setTimeout(() => {
    console.log('initiate second subscribe');
    getLongObs().subscribe(
        rand => console.log(`Second subscribe returned ${rand}`)
    );
}, 2000);

setTimeout(() => {
    console.log('initiate third subscribe');
    getLongObs().subscribe(
        rand => console.log(`Third subscribe returned ${rand}`)
    );
}, 7000)

I trust this explanation aids in understanding the solution!

Answer 3

Answer №2

To achieve what you're looking for, consider using the share() pipe. An example implementation would look like this:


    export class AppComponent {
        private _longObservable: Observable<string> = null

        constructor() {
            this._longObservable = timer(5000).pipe(
                // This will indicate when the timer emits a value to demonstrate that both subscriptions below are sharing the same "execution"
                tap(() => console.log("Timer Triggered!")), 
                map(() => randomString()),
                share()
            );
        }

        ngOnInit() {
            // These two subscriptions will share the observable,
            // as the long observable has not completed by the time the second
            // subscription is triggered.
            this._longObservable.subscribe(console.log);
            setTimeout(() => this._longObservable.subscribe(console.log), 2000);

            // This subscription occurs after 5 seconds.
            // Since timer is a cold observable, this will trigger it to run again.
            setTimeout(() => this._longObservable.subscribe(console.log), 7000);
        }
    }

Output:


Timer Triggered!
randomString1
randomString1
Timer Triggered!
randomString2

If you're unfamiliar with the distinction between hot and cold observables, check out this article: https://medium.com/@benlesh/hot-vs-cold-observables-f8094ed53339

In Angular, HTTP requests and timer(5000) are examples of cold observables.

For more information on the share pipe, visit: https://www.learnrxjs.io/operators/multicasting/share.html

Answer 4

To achieve what you're looking for, consider using the share() pipe. An example implementation would look like this:


    export class AppComponent {
        private _longObservable: Observable<string> = null

        constructor() {
            this._longObservable = timer(5000).pipe(
                // This will indicate when the timer emits a value to demonstrate that both subscriptions below are sharing the same "execution"
                tap(() => console.log("Timer Triggered!")), 
                map(() => randomString()),
                share()
            );
        }

        ngOnInit() {
            // These two subscriptions will share the observable,
            // as the long observable has not completed by the time the second
            // subscription is triggered.
            this._longObservable.subscribe(console.log);
            setTimeout(() => this._longObservable.subscribe(console.log), 2000);

            // This subscription occurs after 5 seconds.
            // Since timer is a cold observable, this will trigger it to run again.
            setTimeout(() => this._longObservable.subscribe(console.log), 7000);
        }
    }

Output:


Timer Triggered!
randomString1
randomString1
Timer Triggered!
randomString2

If you're unfamiliar with the distinction between hot and cold observables, check out this article: https://medium.com/@benlesh/hot-vs-cold-observables-f8094ed53339

In Angular, HTTP requests and timer(5000) are examples of cold observables.

For more information on the share pipe, visit: https://www.learnrxjs.io/operators/multicasting/share.html

Answer 5

Answer №3

In the demonstration you provided, it is evident that your function generates a fresh Observable each time it is subscribed to. One suggestion for your service could be to establish a variable to retain this Observable. Perhaps using a BehaviorSubject would offer an improved approach. By implementing this variable as a BehaviorSubject, you can subscribe to it anywhere in your codebase, ensuring consistency by always accessing the same Observable instance.

Answer 6

In the demonstration you provided, it is evident that your function generates a fresh Observable each time it is subscribed to. One suggestion for your service could be to establish a variable to retain this Observable. Perhaps using a BehaviorSubject would offer an improved approach. By implementing this variable as a BehaviorSubject, you can subscribe to it anywhere in your codebase, ensuring consistency by always accessing the same Observable instance.

If a task is currently ongoing, be sure to subscribe to it; otherwise, restart it

Answer №1

Answer №2

Answer №3

Similar questions

Angular 9 does not include the mat-radio-group component within its known modules

AWS lambda not properly displaying static content for an Angular application when running Node Express server

You need to provide 1 type argument(s) for the Generic type ModuleWithProviders<T> in Angular 10 Swagger Codegen

TypeScript failing to infer type from return value of class method

The combination of Observable streams in combineLatest will persist even if one encounters a

What is the best way to retrieve the final entry from a JSON file while using json server with Angular?

What is the best way to ensure that a div containing lengthy text wraps to the next line as if it were only text overflow?

Angular 2 testing error: Unable to connect to 'ngModel' as it is not recognized as a valid property of 'input'

Exploring the world of publishing Angular 2 applications

Changing dot notation to bracket notation in Angular

How can I dispatch multiple actions simultaneously within a single epic using redux-observable?

What could be the reason behind my Ionic app receiving a 401 error from an API call while Postman is not encountering the

Angular 4 - Automatically scroll to specific list item based on search query input

Tips for configuring VS Code to automatically change a callable property to an arrow function instead of a standard function

Leveraging services in Angular: accessing directly in view or via component

How can I effectively build an Angular Library containing various components?

Ways to identify modifications from a BehaviorSubject and automatically trigger a response according to the updated value

Execute --runTestsByPath on two or more distinct paths

Angular version 5 and above introduces a new feature called "openFromComponent" within the Snackbar component, facilitating seamless communication

Implement the properties encapsulation design pattern