[highland enrich transform] #highland #transform

enrichTransform.js

/**
 * Enrich source stream items with some data
 *
 * @param pick Function that transform source item to pass result to enrichment transform
 * @param enrichmentData Enrichment transformation, which returns the data stream with which the original stream will be enriched
 * @param compose Function that enrich source item with enrichment data
 * @type {EnrichTransform}
 */
export const enrichTransform = (pick, enrichmentData, compose) => (source$) => {
  return source$
    .map(pick)
    .through(enrichmentData)
    .zip(source$.observe())
    .map(([data, src]) => compose(src, data));
};

enrichTransform.types.d.ts

type EnrichTransform = <Src, Pick, Data, Res>(
  pick: (item: Src) => Pick,
  enrichmentData: (source$: Highland.Stream<Pick>) => Highland.Stream<Data>,
  compose: (item: Src, data: Data) => Res
) => (source$: Highland.Stream<Src>) => Highland.Stream<Res>