Если наблюдаемое находится в процессе, подпишитесь на него, если нет, запустите его снова

Question

Если у меня есть наблюдаемое, выполнение которого занимает много времени. Мы назовем его longObservable, который имеет тип Observable, выполнение занимает 5 секунд, и каждый раз, когда он выполняется, он генерирует новую строку только один раз, а затем завершается.

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

И некоторые другие страницы называют это несколько раз. Если это происходит, я бы хотел продолжить этот процесс. Если он завершен, я бы хотел начать его снова.

longObservable.subscribe() // Immediate, start the timer

и это запускается через две секунды:

longObservable.subscribe() // Get the same string as
                            //  the above subscription in 3 seconds.

и это происходит через 20 секунд

longObservable.subscribe() // Start a new iteration and wait
                            // 5 seconds, get a new string.

Вторая подписка, я думаю, проста, она будет работать так, как я хочу. Это третий, с которым у меня проблемы. Он выдаст то же значение, что и два других, сразу после завершения longObservable.

Используется для определения местоположения на устройстве. Я хочу запросить новое местоположение, но если запрос уже выполняется, просто используйте этот результат.

Редактировать: Изменено наблюдаемое на предмет для многоадресной рассылки, удален дубль (1).

Edit2: https://stackblitz.com/edit/angular-venpk4 вот рабочий пример того, что я хочу. Я надеюсь сделать это без переменной timerRunning и с операторами RxJS. Он находится под компонентом hello и печатается на консоли.

Answer 1

Хитрая проблема. Вот мое решение в StackBlitz . Пара ключей к этому - оператор share(), который эффективно превращает наблюдаемое в субъект без необходимости явного объявления субъекта. Однако вам нужно, чтобы НОВАЯ тема создавалась с новой подпиской после завершения старой, поэтому я создал фабричную функцию, чтобы либо возвращать существующую разделяемую Наблюдаемую (если longObservable() все еще выполняется), либо генерировать новую.

Вот важные биты из 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)

Надеюсь, это поможет!

Answer 2

Я думаю, что вам нужна труба share(). Примерно так работает:

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

        constructor() {
            this._longObservable = timer(5000).pipe(
                // This will show us when timer emits a value which will prove that the
                // first two subscriptions below are sharing the same "execution"
                // of the observable.
                tap(() => console.log("Timer Triggered!")), 
                map(() => randomString()),
                share()
            );
        }

        ngOnInit() {
            // These two will share the observable,
            // since long observable hasn't 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 the 5 sec.
            // Since timer is a cold observable, this will trigger it to run again.
            setTimeout(() => this._longObservable.subscribe(console.log), 7000);
        }


    }

Выход:

Timer Triggered!
randomString1
randomString1
Timer Triggered!
randomString2

Вот статья о разнице между горячими и холодными наблюдаемыми, если вы не знакомы с этим различием: https://medium.com/@benlesh/hot-vs-cold-observables-f8094ed53339

Http-запросы в угловых и timer(5000) являются холодными наблюдаемыми.

Вот ссылка на некоторую информацию об общей трубе: https://www.learnrxjs.io/operators/multicasting/share.html

Answer 3

Как вы показываете в своем примере, ваш метод возвращает новый экземпляр Observable, который создается каждый раз, когда вы подписываетесь на него. Я думаю, что к вашим услугам вы можете создать свойство, которое будет хранить ваши наблюдаемые. Может быть, лучше сделать это свойство как BehaviorSubject. И где вы хотите, вы можете подписаться на эту недвижимость. Таким образом, каждый подписчик будет принадлежать одному и тому же наблюдаемому экземпляру.