doc: Add a watcher example to run with sideinput

binding.d.ts

@@ -335,6 +335,7 @@ export declare namespace sideInput {
         start(sockFile?: string | undefined | null, infoFile?: string | undefined | null): Promise<void>
         stop(): void
     }
+    export const DIR_PATH: string
 }
 
 export declare namespace sink {

examples/sideinput/sideinput-mapper/Dockerfile

@@ -0,0 +1,44 @@
+FROM rust:1.91-trixie as builder
+
+# Download and install nvm:
+RUN curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.40.3/install.sh | bash
+# in lieu of restarting the shell
+RUN \. "$HOME/.nvm/nvm.sh" && nvm install 24.11.1
+# Download and install Node.js:
+
+ENV PNPM_HOME=/opt/pnpm
+ENV PATH=/bin/versions/node/v24.11.1/bin:$PNPM_HOME:$PATH
+ENV SHELL=/bin/bash
+
+RUN corepack enable pnpm
+
+RUN pnpm setup && pnpm add -g @napi-rs/cli
+
+# Set working directory
+WORKDIR /app
+
+COPY . .
+
+RUN pnpm run build
+RUN cd examples/sideinput/sideinput-mapper/ && pnpm tsc
+
+FROM node:24-trixie-slim
+
+WORKDIR /app
+
+RUN npm install async-mutex
+
+COPY --from=builder /app/binding.js /app/binding.js
+COPY --from=builder /app/binding.d.ts /app/binding.d.ts
+COPY --from=builder /app/index.js /app/index.js
+COPY --from=builder /app/index.d.ts /app/index.d.ts
+COPY --from=builder /app/*.node /app/
+COPY --from=builder /app/examples /app/examples
+
+RUN addgroup --system appuser && adduser --system --ingroup appuser appuser
+RUN chown -R appuser:appuser /app
+USER appuser
+
+# Set default command (can be changed as needed)
+CMD ["node", "examples/sideinput/sideinput-mapper/sideinput-mapper.js"]
+

examples/sideinput/sideinput-mapper/Makefile

@@ -0,0 +1,16 @@
+TAG ?= stable
+PUSH ?= false
+IMAGE_REGISTRY = quay.io/numaio/numaflow-js/sideinput-mapper:${TAG}
+DOCKER_FILE_PATH = examples/sideinput/sideinput-mapper/Dockerfile
+
+.PHONY: update
+update:
+	cargo check
+	cargo update
+
+.PHONY: image
+image:
+	cd ../../../ && docker build \
+	-f ${DOCKER_FILE_PATH}  \
+	-t ${IMAGE_REGISTRY} . --load
+	@if [ "$(PUSH)" = "true" ]; then docker push ${IMAGE_REGISTRY}; fi

examples/sideinput/sideinput-mapper/README.md

@@ -0,0 +1,21 @@
+An example mapper that reads data from the side-input and forwards it to the sink as a value.
+
+## HOWTO build Image
+
+Use the Makefile:
+
+```bash
+make image
+```
+
+Load it to `kind` cluster
+
+```bash
+kind image import quay.io/numaio/numaflow-js/sideinput-mapper:stable --name <kind-cluster-name>
+```
+
+or `k3d`
+
+```bash
+k3d image import quay.io/numaio/numaflow-js/sideinput-mapper:stable --cluster <k3d-cluster-name>
+```

examples/sideinput/sideinput-mapper/sideinput-mapper.ts

@@ -0,0 +1,76 @@
+import { map, sideInput } from '../../../index'
+import * as fs from 'node:fs'
+import { Mutex } from 'async-mutex'
+
+class SideInputHandler {
+    mutex: Mutex
+    abortController: AbortController
+    dataValue = 'blank'
+    watchedFile = 'my-ticker'
+
+    constructor(mutex: Mutex, abortController: AbortController) {
+        this.mutex = mutex
+        this.abortController = abortController
+    }
+
+    handle = async (datum: map.Datum): Promise<map.Message[]> => {
+        const release = await this.mutex.acquire()
+        try {
+            return [{ keys: datum.keys, value: Buffer.from(this.dataValue.toString()) }]
+        } finally {
+            release()
+        }
+    }
+
+    fileWatcher = async () => {
+        const path = sideInput.DIR_PATH + '/' + this.watchedFile
+        fs.watch(path, { signal: this.abortController.signal }, (eventType, filename) => {
+            if (eventType != 'change') return
+            if (filename) {
+                this.updateDataFromFile(path)
+            }
+        })
+    }
+
+    initDataValue = async () => {
+        const path = sideInput.DIR_PATH + '/' + this.watchedFile
+        await this.updateDataFromFile(path)
+    }
+
+    updateDataFromFile = async (path: string) => {
+        const release = await this.mutex.acquire()
+        try {
+            this.dataValue = fs.readFileSync(path).toString()
+        } finally {
+            release()
+        }
+    }
+}
+
+async function main() {
+    const mutex = new Mutex()
+    const abortController = new AbortController()
+    const sideInputHandler = new SideInputHandler(mutex, abortController)
+    await sideInputHandler.initDataValue()
+    await sideInputHandler.fileWatcher()
+
+    const server = new map.AsyncServer(sideInputHandler.handle)
+
+    const shutdown = () => {
+        server.stop()
+    }
+
+    process.on('SIGTERM', shutdown)
+    process.on('SIGINT', shutdown)
+    process.on('SIGINT', () => {
+        abortController.abort()
+    })
+    process.on('SIGTERM', () => {
+        abortController.abort()
+    })
+
+    console.log('Starting session reduce async server')
+    await server.start()
+}
+
+main().catch(console.error)

examples/sideinput/simple-sideinput.yaml

@@ -4,7 +4,7 @@ metadata:
   name: simple-sideinput
 spec:
   sideInputs:
-    - name: myticker
+    - name: my-ticker
       container:
         image: quay.io/numaio/numaflow-js/simple-sideinput:stable
         imagePullPolicy: IfNotPresent
@@ -17,22 +17,22 @@ spec:
         generator:
           rpu: 30
           duration: 1s
-    - name: si-log
+    - name: si-mapper
       udf:
         container:
-          image: quay.io/numaio/numaflow-rs/udf-sideinput-example:stable
-          imagePullPolicy: Always
+          image: quay.io/numaio/numaflow-js/sideinput-mapper:stable
+          imagePullPolicy: IfNotPresent
       containerTemplate:
         env:
           - name: NUMAFLOW_DEBUG
             value: 'true'
       sideInputs:
-        - myticker
+        - my-ticker
     - name: out
       sink:
         log: {}
   edges:
     - from: in
-      to: si-log
-    - from: si-log
+      to: si-mapper
+    - from: si-mapper
       to: out

package.json

@@ -43,7 +43,7 @@
   },
   "scripts": {
     "artifacts": "napi artifacts",
-    "build:ts": "pnpm tsc -p tsconfig.index.json",
+    "build:ts": "pnpm install && pnpm tsc -p tsconfig.index.json",
     "build:rs": "napi build --platform --js binding.js --dts binding.d.ts",
     "build": "pnpm run build:rs --release && pnpm run build:ts && pnpm format",
     "build:debug": "pnpm run build:rs && pnpm run build:ts && pnpm format",
@@ -79,7 +79,8 @@
     "tsx": "^4.20.6",
     "typedoc": "0.28.15",
     "typescript": "^5.9.3",
-    "vitest": "^4.0.10"
+    "vitest": "^4.0.10",
+    "async-mutex": "^0.5.0"
   },
   "prettier": {
     "printWidth": 120,