First: Toward a solution
Here is a quick mock-up of what you might do:
/**********
* 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`);
}
});
Some Explanation
The 'I want to cancel an ongoing observable based on some upstream event' pattern should always point you toward switchMap
In the snippet above, any event that isn't throttled will either start a new bomb or do nothing. Either way, switchMap will cancel any ongoing bomb (if there is one).
You can probably see that there's quite a bit of room to change that behaviour, but I'm not sure what you'd want, so I leave that up to you.
priorityThrottleTime
priorityThrottleTime really is overkill for what you need but I wrote it a long while back and don't have the time to simplify. You could re-write this to take a predicate and only throttle when the predicate returns true. That way you would avoid the hassle of mapping into and out of that ({priority, payload}) object you see above.
Expand
Expand is a pretty bloated tool for a timer. That's not really a problem, but I might simplify it by just counting 40 seconds oof of the timeSubject that you already maintain.
That way you don't need to check each iteration.
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
}))
)
This does assume your timeSubject tics every second, otherwise you could alter is a bit like this:
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,
}))
))
)
Update
Okay, here's a new way to throttle, though I haven't really tested it much. It'll simplify your code a bunch though so maybe it's worth a try.
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();
}})
);
});
}
and here it is in use:
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`);
}
});