RxJS如何暂停所有计时相关的功能？

Creator 2.x
#1
  • Creator 版本：2.4.9

在试用rxjs的时候，碰到了一个暂停相关的问题。
假如在系统中有很多observable，用到了timer/interval/delayWhen之类时间相关的函数，那么当暂停游戏的时候，如何同时暂停这些observable里面的定时器呢？
R3有各种预制的TimeProvider代替scheduler来控制时间速率，但是rxjs好像没有类似的直接可用的方法。感觉上rxjs应该是要利用自定义的scheduler，但是没有搜到相关的内容。

例子：R3的TimeProvider

Observable.Interval(TimeSpan.FromSeconds(5), UnityTimeProvider.UpdateIgnoreTimeScale);
#2

经过一些对rxjs的研究，搞出了一个简单但是还不完整的解决方法。
可以让使用这个scheduler的timer使用ccc的deltaTime来计时，这样就可以动态暂停回复以及变速了。

// GameplayScheduler.ts

import { SchedulerAction, SchedulerLike, Subscription } from "rxjs";

const {ccclass, property} = cc._decorator;

class MAction<T> extends Subscription implements SchedulerAction<T>{
    constructor(  
        public sche: GameplayScheduler,
        public work: (this: SchedulerAction<T>, state: T) => void, 
        public delay: number, 
        public state: T
    )
    {
        super();
    }

    schedule(state?: T, delay?: number): Subscription{
        this.delay = delay + this.sche.now();
        this.state = state;
        this.sche.reschedule(this);
        return this;
    }
}

// use scaled deltatime to fire scheduled works
export class GameplayScheduler implements SchedulerLike{
    
    private _now : number = 0;
    private _actions : MAction<any>[] = [];

    now(): number { return this._now; }
    
    schedule<T>(work: (this: SchedulerAction<T>, state: T) => void, delay: number, state: T): Subscription{
        const exec_time = delay + this.now();
        const ix = this._find_insert_idx(exec_time);
        if ( !( 0 <= ix && ix <= this._actions.length ) ) cc.error("wrong ix: ", ix);
        const sub = new MAction(this, work, exec_time, state);
        this._actions.splice( ix, 0, sub );
        return sub;
    }

    reschedule<T>(act: MAction<T>){
        const ix = this._find_insert_idx(act.delay);
        this._actions.splice(ix, 0, act);
    }

    update(dt:number){
        this._now += dt * 1000;
        const { _actions } = this;
        while( _actions.length > 0 && _actions[0].delay <= this._now ){
            const act = _actions.shift();
            act.work(act.state);
        }
    }

    private _find_insert_idx(delay:number): number{
        let [L, R] = [0, this._actions.length];
        while (L < R){
            let m = Math.floor((L+R) / 2);
            const elem = this._actions[m];
            if ( elem.delay <= delay ){ L = m+1; }
            else { R = m; }
        }
        return L;
    }

}

// SchedulerCtrl.ts

import { GameplayScheduler } from "./GameplayScheduler";
import { timer, Subscription } from 'rxjs';

const { ccclass, property } = cc._decorator;

@ccclass
export class SchedulerCtrl extends cc.Component{
    @property(cc.Node)
    btn_start : cc.Node = null;
    @property(cc.Node)
    btn_stop : cc.Node = null;

    private _sche : GameplayScheduler = null;
    private _ob : Subscription = null;

    protected onLoad(): void {
        this._sche = new GameplayScheduler();
    }

    start(){
        this.btn_start.on('click', () => {
            // this._ob = timer(1000, 1000).subscribe( x => cc.log(x)); // default scheduler
            this._ob = timer(1000, 1000, this._sche).subscribe( x => cc.log(x)); //gameplay scheduler
        });

        this.btn_stop.on('click', () => {
            if (this._ob == null) return;
            this._ob.unsubscribe();
            this._ob = null;
        });
    }

    update(dt: number){
        this._sche.update(dt);
    }
}