[refactored] _reactiveNextOperator and _reactiveMap to use named args

Summary:
This is the first step towards #193.

A nice side effect of this change in conjunction with D3392 is that the types on `transform` don't have to be manually specified - TypeScript can correctly infer them.  It also makes calls to `reactiveMap` more readable, because the second argument is no longer an unnamed dictionary.

As part of this change, the signature of the operation in `_reactiveNextOperator` is now a higher-order function.  The root cause of D3383 was that there was no closure for an individual observer - state was either ephemeral for an individual emission or shared across all observers.  By passing `emit` and `values` separately, there is now a scope where an operator can store observer-level state.  In my next pass at #193, I will update `_nextOperator` to use this same signature.  In fact, I wonder if `_nextOperator` and `_reactiveNextOperator` can be merged entirely.

As an additional part of #193, I'm changing the name of `dispatch` to `emit` across the codebase.  Something that has been emitted can naturally be called an emission.  There's no good noun to refer to something that has been dispatched.  I expect having an easy way to refer to each item in a stream will make documentation more clear.

I typically abide by the one-small-change-per-diff policy, but since changing to named args is such a massive refactoring, it's a good time to make these smaller changes too.  It reduces the number of refactoring passes over the codebase.

Reviewers: O2 Material Motion, O3 Material JavaScript platform reviewers, #material_motion, featherless

Reviewed By: O2 Material Motion, #material_motion, featherless

Subscribers: featherless

Tags: #material_motion

Differential Revision: http://codereview.cc/D3395

packages/core/src/interactions/Tossable.ts

@@ -177,14 +177,14 @@ export class Tossable {
 
     const locationOnDown$ = this.location$._debounce(whenDragIsActive$);
 
-    this.draggedLocation$ = draggable.value$.addedBy<Point2D>(locationOnDown$, { onlyDispatchWithUpstream: true })._reactiveMap(
-      (
-        location: Point2D,
-        resistanceOrigin: Point2D,
-        radiusUntilResistance: number,
-        resistanceBasis: number,
-        resistanceFactor: number,
-      ) => {
+    this.draggedLocation$ = draggable.value$.addedBy<Point2D>(locationOnDown$, { onlyDispatchWithUpstream: true })._reactiveMap({
+      transform: ({
+        upstream: location,
+        resistanceOrigin,
+        radiusUntilResistance,
+        resistanceBasis,
+        resistanceFactor,
+      }) => {
         if (!resistanceFactor) {
           return location;
         }
@@ -213,16 +213,14 @@ export class Tossable {
           y: resistanceOrigin.y + radiusWithResistance * Math.sin(angle),
         };
       },
-      [
-        this.resistanceOrigin$,
-        this.radiusUntilResistance$,
-        this.resistanceBasis$,
-        this.resistanceFactor$,
-      ],
-      {
-        onlyDispatchWithUpstream: true,
+      inputs: {
+        resistanceOrigin: this.resistanceOrigin$,
+        radiusUntilResistance: this.radiusUntilResistance$,
+        resistanceBasis: this.resistanceBasis$,
+        resistanceFactor: this.resistanceFactor$,
       },
-    );
+      onlyDispatchWithUpstream: true,
+    });
 
     // maybe should be named velocityWhen?
     this.velocity$ = this.draggedLocation$.startWith({ x: 0, y: 0 }).velocity(whenDragIsAtRest$);
@@ -300,16 +298,18 @@ export function applyLinearResistanceToTossable({
 
   subscribe({
     sink: tossable.radiusUntilResistance$,
-    source: min$._reactiveMap(
-      (min: number, max: number) => Math.abs(max - min) / 2,
-      [ max$, ]
-    ),
+    source: min$._reactiveMap({
+      transform: ({ upstream: min, max }) => Math.abs(max - min) / 2,
+      inputs: {
+        max: max$,
+      }
+    }),
   });
 
   subscribe({
     sink: tossable.resistanceOrigin$,
-    source: axis$._reactiveMap(
-      (axis: Axis, min: number, max: number) => {
+    source: axis$._reactiveMap({
+      transform: ({ upstream: axis, min, max }) => {
         const linearCenter = min + (max - min) / 2;
 
         if (axis === Axis.X) {
@@ -326,7 +326,10 @@ export function applyLinearResistanceToTossable({
           console.warn(`Cannot apply linear resistance if axis isn't locked`);
         }
       },
-      [ min$, max$, ],
-    ),
+      inputs: {
+        min: min$,
+        max: max$,
+      },
+    }),
   });
 }

packages/core/src/operators/addedBy.ts

@@ -27,7 +27,7 @@ import {
 } from '../types';
 
 import {
-  ReactiveMappableOptions,
+  _ReactiveMapOptions,
 } from './foundation/_reactiveMap';
 
 export interface MotionAddable<T> {
@@ -40,12 +40,12 @@ export interface MotionAddable<T> {
   // use the type variable `U`.
   addedBy<U extends T & number>(
     amount$: U | Observable<U>,
-    options?: ReactiveMappableOptions,
+    options?: _ReactiveMapOptions,
   ): ObservableWithMotionOperators<number>;
 
   addedBy<U extends T & Point2D>(
     amount$: U | Observable<U>,
-    options?: ReactiveMappableOptions,
+    options?: _ReactiveMapOptions,
   ): ObservableWithMotionOperators<U>;
 }
 
@@ -54,7 +54,7 @@ export function withAddedBy<T, S extends Constructor<MotionMathOperable<T>>>(sup
     /**
      * Adds the amount to the incoming value and dispatches the result.
      */
-    addedBy<U extends T & (Point2D | number)>(amount$: U | Observable<U>, options?: ReactiveMappableOptions): ObservableWithMotionOperators<U> {
+    addedBy<U extends T & (Point2D | number)>(amount$: U | Observable<U>, options?: _ReactiveMapOptions): ObservableWithMotionOperators<U> {
       return this._mathOperator(
         (value, amount) => value + amount,
         amount$,

packages/core/src/operators/dividedBy.ts

@@ -27,7 +27,7 @@ import {
 } from '../types';
 
 import {
-  ReactiveMappableOptions,
+  _ReactiveMapOptions,
 } from './foundation/_reactiveMap';
 
 export interface MotionDivisible<T> {
@@ -41,12 +41,12 @@ export interface MotionDivisible<T> {
   // use the type variable `U`.
   dividedBy<U extends T & number>(
     denominator$: U | Observable<U>,
-    options?: ReactiveMappableOptions,
+    options?: _ReactiveMapOptions,
   ): ObservableWithMotionOperators<number>;
 
   dividedBy<U extends T & Point2D>(
     denominator$: U | Observable<U>,
-    options?: ReactiveMappableOptions,
+    options?: _ReactiveMapOptions,
   ): ObservableWithMotionOperators<U>;
 }
 
@@ -55,7 +55,7 @@ export function withDividedBy<T, S extends Constructor<MotionMathOperable<T>>>(s
     /**
      * Divides the incoming value by the denominator and dispatches the result.
      */
-    dividedBy<U extends T & (Point2D | number)>(denominator$: U | Observable<U>, options?: ReactiveMappableOptions): ObservableWithMotionOperators<U> {
+    dividedBy<U extends T & (Point2D | number)>(denominator$: U | Observable<U>, options?: _ReactiveMapOptions): ObservableWithMotionOperators<U> {
       return this._mathOperator(
         (value, denominator) => value / denominator,
         denominator$,

packages/core/src/operators/foundation/__tests__/_reactiveMap.test.ts

@@ -53,22 +53,27 @@ describe('motionObservable._reactiveMap',
 
     it('should call the transform function with each upstream value',
       () => {
-        subject._reactiveMap(listener, []).subscribe(() => {});
+        subject._reactiveMap({
+          transform: listener,
+          inputs: {}
+        }).subscribe(() => {});
 
         subject.next(2);
-        expect(listener).to.have.been.calledWith(2);
+        expect(listener).to.have.been.calledWithMatch({ upstream: 2 });
 
         subject.next(3);
-        expect(listener).to.have.been.calledWith(3);
+        expect(listener).to.have.been.calledWithMatch({ upstream: 3 });
       }
     );
 
     it('should call the transform function with each sideloaded value',
       () => {
-        subject._reactiveMap(
-          (upstream, sideloaded) => upstream + sideloaded,
-          [ argSubject ],
-        ).subscribe(listener);
+        subject._reactiveMap({
+          transform: ({ upstream, sideloaded }) => upstream + sideloaded,
+          inputs: {
+            sideloaded: argSubject,
+          },
+        }).subscribe(listener);
 
         subject.next(40);
         argSubject.next(2);
@@ -81,10 +86,12 @@ describe('motionObservable._reactiveMap',
 
     it('should call the transform function when any stream dispatches a new value',
       () => {
-        subject._reactiveMap(
-          (upstream, sideloaded) => upstream + sideloaded,
-          [ argSubject ],
-        ).subscribe(listener);
+        subject._reactiveMap({
+          transform: ({ upstream, sideloaded }) => upstream + sideloaded,
+          inputs: {
+            sideloaded: argSubject,
+          },
+        }).subscribe(listener);
 
         subject.next(40);
         argSubject.next(2);
@@ -100,10 +107,13 @@ describe('motionObservable._reactiveMap',
 
     it('should passthrough constants',
       () => {
-        subject._reactiveMap(
-          (upstream, constant, sideloaded) => upstream + constant + sideloaded,
-          [ 100, argSubject ],
-        ).subscribe(listener);
+        subject._reactiveMap({
+          transform: ({ upstream, constant, sideloaded }) => upstream + constant + sideloaded,
+          inputs: {
+            constant: 100,
+            sideloaded: argSubject,
+          },
+        }).subscribe(listener);
 
         subject.next(40);
         argSubject.next(2);

packages/core/src/operators/foundation/_mathOperator.ts

@@ -27,14 +27,14 @@ import {
 } from '../../types';
 
 import {
-  ReactiveMappableOptions,
+  _ReactiveMapOptions,
 } from './_reactiveMap';
 
 export interface MotionMathOperable<T> {
   _mathOperator<U extends T & (number | Point2D)>(
     operation: (upstream: number, parameter: number) => number,
     amount$: U | Observable<U>,
-    options?: ReactiveMappableOptions,
+    options?: _ReactiveMapOptions,
   ): ObservableWithMotionOperators<U>;
 }
 
@@ -43,21 +43,23 @@ export function withMathOperator<T, S extends Constructor<MotionReactiveMappable
     /**
      * Applies the operation to each dimension and dispatches the result.
      */
-    _mathOperator<U extends T & (number | Point2D)>(operation: (upstream: number, parameter: number) => number, amount$: U | Observable<U>, options?: ReactiveMappableOptions): ObservableWithMotionOperators<U> {
-      return (this as any as MotionReactiveMappable<U>)._reactiveMap(
-        (value: U, amount: U) => {
-          if (isPoint2D(value)) {
+    _mathOperator<U extends T & (number | Point2D)>(operation: (upstream: number, parameter: number) => number, amount$: U | Observable<U>, options?: _ReactiveMapOptions): ObservableWithMotionOperators<U> {
+      return (this as any as MotionReactiveMappable<U>)._reactiveMap({
+        transform: ({ upstream, amount }) => {
+          if (isPoint2D(upstream)) {
             return {
-              x: operation(value.x, (amount as Point2D).x),
-              y: operation(value.y, (amount as Point2D).y),
+              x: operation(upstream.x, (amount as Point2D).x),
+              y: operation(upstream.y, (amount as Point2D).y),
             } as U;
           } else {
-            return operation(value as number, amount as number) as U;
+            return operation(upstream as number, amount as number) as U;
           }
         },
-        [ amount$, ],
-        options
-      );
+        inputs: {
+          amount: amount$,
+        },
+        ...options
+      });
     }
   };
 }

packages/core/src/operators/foundation/_reactiveMap.ts

@@ -16,21 +16,23 @@
 
 import {
   Constructor,
+  MaybeReactive,
   MotionReactiveNextOperable,
   MotionTappable,
   NextChannel,
   ObservableWithMotionOperators,
 } from '../../types';
 
-export type ReactiveMappableOptions = {
+export type _ReactiveMapOptions = {
   onlyDispatchWithUpstream?: boolean,
 };
+
+export type _ReactiveMapArgs<D, T, U> = _ReactiveMapOptions & {
+  transform: (kwargs: { upstream: T } & D) => U,
+  inputs: MaybeReactive<D>,
+};
 export interface MotionReactiveMappable<T> {
-  _reactiveMap<U>(
-    transform: (upstreamValue: T, ...args: Array<any>) => U,
-    args: Array<any>,
-    options?: ReactiveMappableOptions,
-  ): ObservableWithMotionOperators<U>;
+  _reactiveMap<U, D>(kwargs: _ReactiveMapArgs<D, T, U>): ObservableWithMotionOperators<U>;
 }
 
 export function withReactiveMap<T, S extends Constructor<MotionReactiveNextOperable<T> & MotionTappable<T>>>(superclass: S): S & Constructor<MotionReactiveMappable<T>> {
@@ -43,22 +45,22 @@ export function withReactiveMap<T, S extends Constructor<MotionReactiveNextOpera
      * If the `onlyDispatchWithUpstream` option is set, `transform` is only
      * called when the upstream value changes.
      */
-    _reactiveMap<U>(transform: (upstreamValue: T, ...args: Array<any>) => U, args: Array<any>, { onlyDispatchWithUpstream = false }: ReactiveMappableOptions = {}): ObservableWithMotionOperators<U> {
+    _reactiveMap<U, D>({ transform, inputs, onlyDispatchWithUpstream = false }: _ReactiveMapArgs<D, T, U>): ObservableWithMotionOperators<U> {
       let upstreamChanged = false;
 
       return this._tap(
         () => {
           upstreamChanged = true;
         }
-      )._reactiveNextOperator(
-        (dispatch: NextChannel<U>, upstreamValue: T, ...argValues: Array<any>) => {
+      )._reactiveNextOperator<U, D>({
+        operation: ({ emit }) => (values) => {
           if (upstreamChanged || !onlyDispatchWithUpstream) {
-            dispatch(transform(upstreamValue, ...argValues));
+            emit(transform(values));
           }
           upstreamChanged = false;
         },
-        args
-      );
+        inputs
+      });
     }
   };
 }

packages/core/src/operators/foundation/_reactiveNextOperator.ts

@@ -29,16 +29,22 @@ import {
 
 import {
   Constructor,
+  EmittingOperation,
+  MaybeReactive,
   NextChannel,
   Observable,
   ObservableWithMotionOperators,
   Observer,
-  ReactiveNextOperation,
   Subscription,
 } from '../../types';
 
+export type _ReactiveNextOperatorArgs<D, T, U> = CombineLatestOptions & {
+  operation: EmittingOperation<D, T, U>,
+  inputs: MaybeReactive<D>,
+}
+
 export interface MotionReactiveNextOperable<T> extends Observable<T> {
-  _reactiveNextOperator<U>(operation: ReactiveNextOperation<T, U>, args: Array<any>, options?: CombineLatestOptions): ObservableWithMotionOperators<U>;
+  _reactiveNextOperator<U, D>(kwargs: _ReactiveNextOperatorArgs<D, T, U>): ObservableWithMotionOperators<U>;
 }
 
 export function withReactiveNextOperator<T, S extends Constructor<Observable<T>>>(superclass: S): S & Constructor<MotionReactiveNextOperable<T>> {
@@ -55,12 +61,22 @@ export function withReactiveNextOperator<T, S extends Constructor<Observable<T>>
      * `_reactiveNextOperator` will not call `operation` until it has received
      * a value from every argument it is subscribed to.
      */
-    _reactiveNextOperator<U>(operation: ReactiveNextOperation<T, U>, args: Array<any>, options?: CombineLatestOptions): ObservableWithMotionOperators<U> {
+    _reactiveNextOperator<U, D>({ operation, inputs, ...combineLatestOptions }: _ReactiveNextOperatorArgs<D, T, U>): ObservableWithMotionOperators<U> {
       return new MotionObservable(
         (observer: Observer<U>) => {
-          const boundNext: NextChannel<U> = observer.next.bind(observer);
-          return combineLatest([ this, ...args ], options).subscribe(
-            (values) => operation(boundNext, ...values)
+          const innerOperation = operation({
+            emit: observer.next.bind(observer),
+          });
+
+          return combineLatest<D, MaybeReactive<D>>(
+            // TypeScript doesn't like ...inputs, so we use the longhand version
+            Object.assign(
+              { upstream: this },
+              inputs,
+            ),
+            combineLatestOptions
+          ).subscribe(
+            innerOperation
           ).unsubscribe;
         }
       );

packages/core/src/operators/lowerBound.ts

@@ -28,10 +28,12 @@ export interface MotionLowerBoundable {
 export function withLowerBound<T, S extends Constructor<MotionReactiveMappable<T>>>(superclass: S): S & Constructor<MotionLowerBoundable> {
   return class extends superclass implements MotionLowerBoundable {
     lowerBound(limit$: number | Observable<number>): ObservableWithMotionOperators<number> {
-      return (this as any as MotionReactiveMappable<number>)._reactiveMap(
-        (value: number, limit) => Math.max(value, limit),
-        [ limit$ ],
-      );
+      return (this as any as MotionReactiveMappable<number>)._reactiveMap({
+        transform: ({ upstream, limit }) => Math.max(upstream, limit),
+        inputs: {
+          limit: limit$,
+        },
+      });
     }
   };
 }