Cancelling expand if other event occurs

Viewed 410

I'm new to ReactiveX and have been trying to build a "CS:Go bomb explosion notifier" program using RxJS.

I have the following so far:

// TimeSubject is the current heartbeat of the game (i.e. current client time),
// it is not emitted every 1 second (it will be emitted at least once every 30
// seconds and also whenever any other game event occurs)

// RoundSubject stores information about the current and previous round state
combineLatest([TimeSubject, RoundSubject])
    .pipe(
        // detect when the bomb is planted
        filter(([_, { current }]) => current.bomb === 'planted'),
        // create a timer object to keep track of the time until explosion
        map(([time, _]) => ({ plantTime: time, explosionTime: time + 40, currentTime: time })),
        // ignore other "bomb planted" events for the next 50 seconds
        throttleTime(50 * 1000),
        // count down until bomb is exploded
        // problem: RoundSubject can emit an event indicating the bomb got defused
        //          or the round ended before the bomb got to explode,
        //          but how do I catch that since I am throttling the events from that channel?
        expand(({ plantTime, explosionTime, currentTime }) =>
            explosionTime > currentTime
                ? of(({ plantTime, explosionTime, currentTime: currentTime + 1 }))
                    .pipe(delay(1000))
                : EMPTY)
    ).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`);
        }
    });

The problem here is that RoundSubject can emit an event like RoundEnded or Defused, which in either case should cancel the timer.

At the moment I don't know enough about the available operators to see how I can fix this in a good way. Additionally, I feel that my code is quite convoluted with the expand, so if you know a better approach, let me know :-).

Thanks.

3 Answers

When in doubt, name things!

By far the biggest pitfall in writing code with RxJS is writing observables with 5+ operators in a single pipe. It's very easy to lose the plot.

Don't be afraid to create multiple named streams; things will read much more naturally.

// This creates a stream that emits every time the bomb's status changes to the
// provided value.
const bombChangedStatusTo = (status) =>
  RoundSubject.pipe(
    pluck('current'),
    distinctUntilKeyChanged('bomb'),
    filter((bombStatus) => bombStatus === status)
  );

const bombPlanted$ = bombChangedStatusTo('planted');
const bombDefused$ = bombChangedStatusTo('defused');

The other answer is correct, expand is overkill here. Assuming we know the start time, the countdown can be as simple as mapping over the values emitted by some interval (see the last section for why we don't actually need plantTime here).

// we use share() since we'll subscribe to this more than once, it
// ensures that we're subscribing to the exact same interval each time
const clockInterval$ = interval(1000).pipe(
  startWith(null), // emit immediately instead of after 1s
  map(() => Math.floor(Date.now()/1000)),
  share()
);

const countDown = (startTime) =>
  clockInterval$.pipe(
    map((currentTime) => ({
      explosionTime: startTime + 40,
      currentTime
    })),
    takeWhile(
      ({ currentTime, explosionTime }) => currentTime < explosionTime,
      true // include the emission that triggered completion
    )
  );

Here we use exhaustMap to ensure that only one timer will be run per "bomb planted" event (see the docs). No need to use throttleTime, which would give us two timers counting to 40 instead of just one.

const bombClock$ = bombPlanted$.pipe(
  withLatestFrom(clockInterval$), // <-- reusing the shared clock
  exhaustMap(([_, plantTime]) =>
    countDown(plantTime).pipe(
      takeUntil(bombDefused$) // stop the timer if the bomb is defused
    )
  )
);

If we trigger the "bomb was planted" side effect using bombPlanted$, we no longer need to pass around plantTime as a property on the bombClock$ value

bombPlanted$.subscribe(() => {
  console.log('Bomb planted and will explode in 40 seconds!');
});

bombClock$.subscribe(({ explosionTime, currentTime }) => {
  if (explosionTime >= currentTime) {
    const secondsToExplode = explosionTime - currentTime;
    console.log(`.. explodes in: ${secondsToExplode} seconds`);
  } else {
    console.log('The bomb has exploded');
  }
});

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`);
  }
});

If you want to catch the RoundEnded or Defused events, a thing that you can do is to create a new stream that will be listening for those events and cancel the timer when needed.

let cancelStream$ = RoundedSubject.pipe(
  filter(({current}) => current.bomb === 'RoundEnded' || current.bomb === 'Defused'),
  
)

// TimeSubject is the current heartbeat of the game (i.e. current client time)
// RoundSubject stores information about the current and previous round state
combineLatest([TimeSubject, RoundSubject])
    .pipe(
        takeUntil(cancelStream$),
        // detect when the bomb is planted
        filter(([_, { current }]) => current.bomb === 'planted'),
        // create a timer object to keep track of the time until explosion
        map(([time, _]) => ({ plantTime: time, explosionTime: time + 40, currentTime: time })),
        // ignore other "bomb planted" events for the next 50 seconds
        throttleTime(50 * 1000),
        // count down until bomb is exploded
        // problem: RoundSubject can emit an event indicating the bomb got defused
        //          or the round ended before the bomb got to explode,
        //          but how do I catch that since I am throttling the events from that channel?
        expand(({ plantTime, explosionTime, currentTime }) =>
            explosionTime > currentTime
                ? of(({ plantTime, explosionTime, currentTime: currentTime + 1 }))
                    .pipe(delay(1000))
                : EMPTY)
    ).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`);
        }
    });

Related