Namespace sink

Example

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

const server = new sink.AsyncServer(async (datums) => {
  const responses: sink.Response[] = [];

  for await (const datum of datums) {
    try {
      // Write to your external system
      await writeToDatabase(datum.getValue());
      responses.push(sink.Response.ok(datum.id));
    } catch (error) {
      responses.push(sink.Response.failure(datum.id, error.message));
    }
  }

  return responses;
});

server.start();
Copy