RxJS Operator Simplification / Creation 6.x

operators.js

export const basic = (subscriber) => (origSource) =>
  origSource.lift({
    call(dest, source) {
      return source.subscribe(subscriber(dest));
    }
  });

export const through = (cb) =>
  basic((dest) => {
    return {
      next(msg) {
        cb(msg, dest);
      },
      error: dest.error,
      complete: dest.complete
    };
  });

export const bufferThrough = (cb) => {
  const buffer = [];
  const throughCb = cb(buffer);
  // allow closure for handling state w/ buffer
  /*
    bufferThrough((buffer) => {
      let counter = 0 // etc...
      return (msg, dest) => {

      }
    })
  */
  return through(throughCb);
};

Reduces something like this ugliness

  const bufferTimelineCounter = () => (origSource) =>
    origSource.lift({
      call(dest, source) {
        const buffer = [];
        return source.subscribe({
          next(msg) {
            buffer.push(msg);
            dest.next(buffer);

            timelineCount += msg.body.data.timeline.length;
            if (timelineCount >= MAX_TAGS_THRESHOLD) {
              dest.complete();
            }
          },
          error: dest.error,
          complete: dest.complete
        });
      }
    });

to

      bufferThrough((buffer) => {
        let timelineCount = 0;
        return (msg, dest) => {
          buffer.push(msg);
          dest.next(buffer);

          timelineCount += msg.body.data.timeline.length;
          if (timelineCount >= MAX_TAGS_THRESHOLD) {
            dest.complete();
          }
        };
      })
    )