In rxjs, how do I get the last value from a switchMap that is ended by takeUntil?

user_793454 0
Viewed 498

Here is a typical rxjs example of drag-and-drop:

const drag$ = mousedown$.pipe(
  switchMap(
    (start) => {
      return mousemove$.pipe(map(move => {
        move.preventDefault();
        return {
          left: move.clientX - start.offsetX,
          top: move.clientY - start.offsetY
        }
      }),
        takeUntil(mouseup$));
    }));

drag$.subscribe(pos => {
  box.style.top = `${pos.top}px`
  box.style.left = `${pos.left}px`
});

The takeUntil will unsubscribe the inner observable of mouseMove$. But what if I need the last value in the inner observable in order to do something with it upon the mouseUp? I know I can probably assign a value to a global temporary variable to get at it that way but I'm trying to avoid that.

The more involved story of what I'm trying to do is this:

I'm working on a 3D game where I want to implement grab-and-throw. I've set it up similar to the rxjs conanical drag-and-drop scenario where I'm subscribing to a controller 'squeeze' event, then switchMaping to another observable of controller movement where I 'drag' and reposition the object where ever my hand position is, and finally I takeUntil the 'squeeze' is released, which lets go of the object.

But there is one more piece I need to implement for 'throwing' the object, which is I need the velocity vector of my hand movement at the moment the squeeze is released to apply an impulse force to the object so it flies away when I let go. That vector can be calculated on every frame inside the switchMap's pipe using scan.

The problem is takeUntil just stops the data stream and I don't have access the last value that was in the accumulator of scan.

rxjs
2 Answers

You can use the withLatestFrom() operator to get the most recent emission from another observable. The result is an array of the stream value, followed by the other observable's last emission:

sample shape

mouseup$.pipe(
  withLatestFrom(drag$)
).subscribe(
  ([mouseup, drag]) => console.log({drag, mouseup})
);

Here's a StackBlitz example.

user_18583570

This is a very interesting problem, I'm guessing you also need the values as the drag event takes place, but also its last emitted value before the mouseup event occurs.

One approach I think would be this:

const drag$ = mousedown$.pipe(
  switchMap(
    (start) => {
      mousemove$ = mousemove$.pipe(
        map(move => {
          move.preventDefault();
          return {
            left: move.clientX - start.offsetX,
            top: move.clientY - start.offsetY
          }
        }),
        takeUntil(mouseup$),
        share(),
      );

      return concat(
        // get the `mousemove` events as they take place
        mousemove$,

        mousemove$.pipe(
          reduce((acc, crt) => { /* ... */ }),
        ),
      )
}));

The reduce operator works the same as scan, with the exception that the former will emit the accumulated value when the source completes.

If you want to use the same scan logic in both observables, then the switchMap's inner observable could look like this:

mousemove$ = mousemove$.pipe(
  map(move => {
    move.preventDefault();
    return {
      left: move.clientX - start.offsetX,
      top: move.clientY - start.offsetY
    }
  }),
  scan((acc, crt) => { /* ... */ }),
  takeUntil(mouseup$),
  share(),
);

return concat(
  mousemove$,

  mousemove$.pipe(
    // it will emit the last emitted value when the source completes
    takeLast(1),
  ),
)
user_96326210
Related