Namespace reduce

Example

import { reduce } from '@numaproj/numaflow-js';

const server = new reduce.AsyncServer(async (keys, datums, metadata) => {
  let sum = 0;
  let count = 0;

  for await (const datum of datums) {
    const value = parseInt(datum.value.toString());
    sum += value;
    count++;
  }

  return [new reduce.Message(Buffer.from(JSON.stringify({
    keys,
    sum,
    count,
    window: metadata.intervalWindow
  })))];
});

server.start();
Copy