0
\$\begingroup\$

The code is written in TypeScript.

export function waitFor<T, R> (
  reducer: (acc: R, value: T) => R,
  initial: R,
  condition: (accumulated: R) => boolean,
): OperatorFunction<T, R> {
  return (source$: Observable<T>) => {
    return new Observable<R>((subscriber) => {
      let accumulated: R = initial
      return source$.subscribe({
        next (value) {
          accumulated = reducer(accumulated, value)
          if (condition(accumulated)) {
            subscriber.next(accumulated)
            subscriber.complete()
          }
        },
        error (error) {
          subscriber.error(error)
        },
        complete () {
          subscriber.complete()
        },
      })
    })
  }
}

Here's a passing marable test which should make it more obvious what the operator is supposed to do.

import { marbles } from 'rxjs-marbles/jest'
import { waitFor } from './utils'

describe(`waitFor`, () => {

  it(`waits for sum to be over 12`, marbles(m => {
    const source = m.cold('  ----a----b----c----d----e----f----g------|', { a: 1, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7 })
    const expected = m.cold('------------------------(x|)', { x: 15 })
    const actual = source.pipe(waitFor((a, b) => a + b, 0, sum => sum > 12))
    m.expect(actual).toBeObservable(expected)
  }))

})

I'm mostly interested in the best practices of writing the operator. Did I miss unsubscribing from something and introduced a memory leak? Did I reinvent the wheel, i.e. could I have achieved the same versatility of the operator by combining the existing standard ones?

I realize now this is just a scan followed by a skipUntil and first. Ignoring this, I'd still like to know if my logic was sound and if I missed something important in my custom operator.

\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Yes, this implementation looks sound to me.

Also you should first try to implement new operators like this by combining existing operators, since that leaves less room for mistakes, oversights, and let you piggyback on all the years of improvements the library wend trough.

You already mentioned you re-implemented scan + something extra. Myself i would use find.

The resulting operator below already has 2 minor improvements:

export function waitFor<T, R> (
  reducer: (acc: R | S, value: T, index: number) => R,
  seed: S,
  condition: (accumulated: R, index: number) => boolean,
): OperatorFunction<T, R> {
  return (source$: Observable<T>) =>
    source$.pipe(
      scan(reducer, initial),
      find(condition)
    )
}

The reducer and the condition now have an index parameter.

The scan operator gave me inspiration to separate the type of the seed and the accumulated.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.