首先：寻求解决方案
这里是你可能做的一个快速模型：
/**********
* Custom Operator to throttle only specific emissions
*********/
function priorityThrottleTime<T>(
thrTime: number,
priorityStr = "priority"
): MonoTypeOperatorFunction<T> {
return s => defer(() => {
const priorityTimeStamp = new Map<string, number>();
return s.pipe(
filter(v => Date.now() - (
priorityTimeStamp.get(v[priorityStr]) ||
0) >= thrTime
),
tap(v => {
if(v[priorityStr] != null){
priorityTimeStamp.set(
v[priorityStr],
Date.now()
)
}
})
);
});
}
// TimeSubject is the current heartbeat of the game (i.e. current client time)
// RoundSubject stores information about the current and previous round state
roundSubject.pipe(
// detect when the bomb is planted, map to priority: 1,
// otherwise map without priority
map(round =>
round.current.bomb === 'planted' ?
({priority: 1, payload: round}) :
({payload: round})
),
// same prioroty events ("bomb planted" events)
// ignored for the next 50 seconds
priorityThrottleTime(50 * 1000),
// Throttling is done, get our payload back
map(({payload}) => payload),
// create a new observable depending on what the round is doing
switchMap(({current}) =>
current.bomb !== 'planted' ?
EMPTY :
timeSubject.pipe(
// Grab the next heartbeat
take(1),
// create a timer object to keep track of the time until explosion
map(time => ({
plantTime: time,
explosionTime: time + 40,
currentTime: time
})),
// count down until bomb is exploded
expand(({ plantTime, explosionTime, currentTime }) =>
currentTime > explosionTime ?
EMPTY :
of(({
plantTime,
explosionTime,
currentTime: currentTime + 1
})).pipe(delay(1000))
)
)
)
).subscribe(({ plantTime, explosionTime, currentTime }) => {
if (plantTime === currentTime)
console.log('Bomb planted and will explode in 40 seconds!');
else if (explosionTime >= currentTime) {
const secondsToExplode = explosionTime - currentTime;
console.log(`.. explodes in: ${secondsToExplode} seconds`);
}
});
一些解释
'我想取消基于某些上游事件的正在进行的 observable' 模式应该始终将您指向switchMap
在上面的 sn-p 中，任何没有被限制的事件要么启动一个新的炸弹，要么什么都不做。无论哪种方式，switchMap 都会取消任何正在进行的炸弹（如果有的话）。
您可能会发现有很大的空间可以改变这种行为，但我不确定您想要什么，所以我把它留给您。
priorityThrottleTime
priorityThrottleTime 对于您需要的东西来说确实有点过头了，但我很久以前就写过它，没有时间简化。您可以重新编写它以获取谓词，并且仅在谓词返回 true 时进行节流。这样您就可以避免映射进出上面看到的 ({priority, payload}) 对象的麻烦。
展开
Expand 是一个相当臃肿的计时器工具。这不是一个真正的问题，但我可以通过计算您已经维护的 timeSubject 的 40 秒来简化它。
这样您就不需要检查每次迭代。
timeSubject.pipe(
// first emissions is time 0,
// then take 40 seconds, then stop
take(41),
// create the timer object
map((currentTime, timePassed) => ({
plantTime: currentTime - timePassed,
explosionTime: 40 - timePassed,
currentTime
}))
)
这确实假设你的 timeSubject 每秒抽动一次，否则你可以改变的是有点像这样：
timeSubject.pipe(
take(1),
// create a timer that tics every second
switchMap(plantTime => timer(0, 1000).pipe(
// take 40 seconds, then stop
take(41),
// create a timer object
map(timePassed => ({
plantTime,
explosionTime: plantTime + 40,
currentTime: plantTime + timePassed,
}))
))
)
更新
好的，这是一种新的节流方法，尽管我还没有真正测试过它。虽然它会简化你的代码，所以也许值得一试。
function throttleTimeOn<T>(
thrTime: number,
pred: (x:T) => boolean
): MonoTypeOperatorFunction<T> {
return s => defer(() => {
let throttleTimeStamp = 0;
return s.pipe(
filter(v => {
const isThrot = pred(v);
if(!isThrot) return true;
else return Date.now() -
throttleTimeStamp >= thrTime;
}),
tap(v => { if(pred(v)) {
throttleTimeStamp = Date.now();
}})
);
});
}
这里正在使用：
roundSubject.pipe(
// events that meet predicate ("bomb planted" events)
// ignored for the next 50 seconds
throttleTimeOn(
50 * 1000,
({current}) => current.bomb === 'planted'
),
// create a new observable depending on what the round is doing
switchMap(({current}) =>
current.bomb !== 'planted' ?
EMPTY :
timeSubject.pipe(
// first emissions is time 0,
// then take 40 seconds, then stop
take(41),
// create the timer object
map((currentTime, timePassed) => ({
plantTime: currentTime - timePassed,
explosionTime: 40 - timePassed,
currentTime
}))
)
)
).subscribe(({ plantTime, explosionTime, currentTime }) => {
if (plantTime === currentTime)
console.log('Bomb planted and will explode in 40 seconds!');
else if (explosionTime >= currentTime) {
const secondsToExplode = explosionTime - currentTime;
console.log(`.. explodes in: ${secondsToExplode} seconds`);
}
});