feat: binary messages through ws + camera demo (#34)

* feat: binary messages through ws + camera demo

* chore: prepare release

apps/demo-nextjs-app-router/app/camera-turbo/page.tsx

@@ -0,0 +1,205 @@
+/* eslint-disable @next/next/no-img-element */
+'use client';
+
+import * as fal from '@fal-ai/serverless-client';
+import { MutableRefObject, useEffect, useRef, useState } from 'react';
+
+fal.config({
+  proxyUrl: '/api/fal/proxy',
+});
+
+const EMPTY_IMG =
+  'data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAAXNSR0IArs4c6QAAAA1JREFUGFdjOHPmzH8ACDADZKt3GNsAAAAASUVORK5CYII=';
+
+type WebcamOptions = {
+  videoRef: MutableRefObject<HTMLVideoElement | null>;
+  previewRef: MutableRefObject<HTMLCanvasElement | null>;
+  onFrameUpdate?: (data: Uint8Array) => void;
+  width?: number;
+  height?: number;
+};
+const useWebcam = ({
+  videoRef,
+  previewRef,
+  onFrameUpdate,
+  width = 512,
+  height = 512,
+}: WebcamOptions) => {
+  useEffect(() => {
+    if (navigator.mediaDevices && navigator.mediaDevices.getUserMedia) {
+      navigator.mediaDevices.getUserMedia({ video: true }).then((stream) => {
+        if (videoRef.current !== null) {
+          videoRef.current.srcObject = stream;
+          videoRef.current.play();
+        }
+      });
+    }
+  }, [videoRef]);
+
+  const captureFrame = () => {
+    const canvas = previewRef.current;
+    const video = videoRef.current;
+    if (canvas === null || video === null) {
+      return;
+    }
+
+    // Calculate the aspect ratio and crop dimensions
+    const aspectRatio = video.videoWidth / video.videoHeight;
+    let sourceX, sourceY, sourceWidth, sourceHeight;
+
+    if (aspectRatio > 1) {
+      // If width is greater than height
+      sourceWidth = video.videoHeight;
+      sourceHeight = video.videoHeight;
+      sourceX = (video.videoWidth - video.videoHeight) / 2;
+      sourceY = 0;
+    } else {
+      // If height is greater than or equal to width
+      sourceWidth = video.videoWidth;
+      sourceHeight = video.videoWidth;
+      sourceX = 0;
+      sourceY = (video.videoHeight - video.videoWidth) / 2;
+    }
+
+    // Resize the canvas to the target dimensions
+    canvas.width = width;
+    canvas.height = height;
+
+    const context = canvas.getContext('2d');
+    if (context === null) {
+      return;
+    }
+
+    // Draw the image on the canvas (cropped and resized)
+    context.drawImage(
+      video,
+      sourceX,
+      sourceY,
+      sourceWidth,
+      sourceHeight,
+      0,
+      0,
+      width,
+      height
+    );
+
+    // Callback with frame data
+    if (onFrameUpdate) {
+      canvas.toBlob(
+        (blob) => {
+          blob?.arrayBuffer().then((buffer) => {
+            const frameData = new Uint8Array(buffer);
+            onFrameUpdate(frameData);
+          });
+        },
+        'image/jpeg',
+        0.7
+      );
+    }
+  };
+
+  useEffect(() => {
+    const interval = setInterval(() => {
+      captureFrame();
+    }, 16); // Adjust interval as needed
+
+    return () => clearInterval(interval);
+  });
+};
+
+type LCMInput = {
+  prompt: string;
+  image: Uint8Array;
+  strength?: number;
+  negative_prompt?: string;
+  seed?: number | null;
+  guidance_scale?: number;
+  num_inference_steps?: number;
+  enable_safety_checks?: boolean;
+  request_id?: string;
+  height?: number;
+  width?: number;
+};
+
+type LCMOutput = {
+  image: Uint8Array;
+  timings: Record<string, number>;
+  seed: number;
+  num_inference_steps: number;
+  request_id: string;
+  nsfw_content_detected: boolean[];
+};
+
+export default function WebcamPage() {
+  const [enabled, setEnabled] = useState(false);
+  const processedImageRef = useRef<HTMLImageElement | null>(null);
+  const videoRef = useRef<HTMLVideoElement | null>(null);
+  const previewRef = useRef<HTMLCanvasElement | null>(null);
+
+  const { send } = fal.realtime.connect<LCMInput, LCMOutput>(
+    '110602490-sd-turbo-real-time-high-fps-msgpack',
+    {
+      connectionKey: 'camera-turbo-demo',
+      // not throttling the client, handling throttling of the camera itself
+      // and letting all requests through in real-time
+      throttleInterval: 0,
+      onResult(result) {
+        if (processedImageRef.current && result.image) {
+          const blob = new Blob([result.image], { type: 'image/jpeg' });
+          const url = URL.createObjectURL(blob);
+          processedImageRef.current.src = url;
+        }
+      },
+    }
+  );
+
+  const onFrameUpdate = (data: Uint8Array) => {
+    if (!enabled) {
+      return;
+    }
+    send({
+      prompt: 'a picture of leonardo di caprio, elegant, in a suit, 8k, uhd',
+      image: data,
+      num_inference_steps: 3,
+      strength: 0.44,
+      guidance_scale: 1,
+      seed: 6252023,
+    });
+  };
+
+  useWebcam({
+    videoRef,
+    previewRef,
+    onFrameUpdate,
+  });
+
+  return (
+    <main className="flex-col px-32 mx-auto my-20">
+      <h1 className="text-4xl font-mono mb-8 text-current text-center">
+        fal<code className="font-light text-pink-600">camera</code>
+      </h1>
+      <video ref={videoRef} style={{ display: 'none' }}></video>
+      <div className="py-12 flex items-center justify-center">
+        <button
+          className="py-3 px-4 bg-indigo-700 text-white text-lg rounded"
+          onClick={() => {
+            setEnabled(!enabled);
+          }}
+        >
+          {enabled ? 'Stop' : 'Start'}
+        </button>
+      </div>
+      <div className="flex flex-col lg:flex-row space-y-4 lg:space-y-0 lg:space-x-4 justify-between">
+        <canvas ref={previewRef} width="512" height="512"></canvas>
+        <img
+          ref={processedImageRef}
+          src={EMPTY_IMG}
+          width={512}
+          height={512}
+          className="min-w-[512px] min-h-[512px]"
+          alt="generated"
+        />
+      </div>
+    </main>
+  );
+}

libs/client/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@fal-ai/serverless-client",
   "description": "The fal serverless JS/TS client",
-  "version": "0.7.0",
+  "version": "0.7.1",
   "license": "MIT",
   "repository": {
     "type": "git",
@@ -16,6 +16,7 @@
     "ml"
   ],
   "dependencies": {
+    "msgpackr": "^1.10.0",
     "robot3": "^0.4.1"
   },
   "engines": {

libs/client/src/realtime.ts

@@ -1,4 +1,5 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
+import { pack, unpack } from 'msgpackr';
 import {
   ContextFunction,
   createMachine,
@@ -75,7 +76,14 @@ function closeConnection(context: Context): Context {
 
 function sendMessage(context: Context, event: SendEvent): Context {
   if (context.websocket && context.websocket.readyState === WebSocket.OPEN) {
-    context.websocket.send(JSON.stringify(event.message));
+    if (event.message instanceof Uint8Array) {
+      context.websocket.send(event.message);
+    } else if (shouldSendBinary(event.message)) {
+      context.websocket.send(pack(event.message));
+    } else {
+      context.websocket.send(JSON.stringify(event.message));
+    }
+
     return {
       ...context,
       enqueuedMessage: undefined,
@@ -260,6 +268,16 @@ function buildRealtimeUrl(
 const TOKEN_EXPIRATION_SECONDS = 120;
 const DEFAULT_THROTTLE_INTERVAL = 128;
 
+function shouldSendBinary(message: any): boolean {
+  return Object.values(message).some(
+    (value) =>
+      value instanceof Buffer ||
+      value instanceof Blob ||
+      value instanceof ArrayBuffer ||
+      value instanceof Uint8Array
+  );
+}
+
 /**
  * Get a token to connect to the realtime endpoint.
  */
@@ -452,7 +470,33 @@ export const realtimeImpl: RealtimeClient = {
             onError(new ApiError({ message: 'Unknown error', status: 500 }));
           };
           ws.onmessage = (event) => {
+            const { onResult } = getCallbacks();
+
+            // Handle binary messages as msgpack messages
+            if (event.data instanceof ArrayBuffer) {
+              const result = unpack(new Uint8Array(event.data));
+              onResult(result);
+              return;
+            }
+            if (
+              event.data instanceof Buffer ||
+              event.data instanceof Uint8Array
+            ) {
+              const result = unpack(event.data);
+              onResult(result);
+              return;
+            }
+            if (event.data instanceof Blob) {
+              event.data.arrayBuffer().then((buffer) => {
+                const result = unpack(new Uint8Array(buffer));
+                onResult(result);
+              });
+              return;
+            }
+
+            // Otherwise handle strings as plain JSON messages
             const data = JSON.parse(event.data);
+
             // Drop messages that are not related to the actual result.
             // In the future, we might want to handle other types of messages.
             // TODO: specify the fal ws protocol format
@@ -461,7 +505,6 @@ export const realtimeImpl: RealtimeClient = {
               return;
             }
             if (isSuccessfulResult(data)) {
-              const { onResult } = getCallbacks();
               onResult(data);
               return;
             }
@@ -485,12 +528,18 @@ export const realtimeImpl: RealtimeClient = {
 
     const send = (input: Input & Partial<WithRequestId>) => {
       // Use throttled send to avoid sending too many messages
+
+      const message =
+        input instanceof Uint8Array
+          ? input
+          : {
+              ...input,
+              request_id: input['request_id'] ?? crypto.randomUUID(),
+            };
+
       stateMachine.throttledSend({
         type: 'send',
-        message: {
+        message,
-          ...input,
-          request_id: input['request_id'] ?? crypto.randomUUID(),
-        },
       });
     };
 

package-lock.json

@@ -24,6 +24,7 @@
         "http-proxy": "^1.18.1",
         "http-proxy-middleware": "^2.0.6",
         "js-base64": "^3.7.5",
+        "msgpackr": "^1.10.0",
         "next": "^14.0.3",
         "react": "^18.2.0",
         "react-dom": "^18.2.0",
@@ -3716,6 +3717,78 @@
       "integrity": "sha512-Hcv+nVC0kZnQ3tD9GVu5xSMR4VVYOteQIr/hwFPVEvPdlXqgGEuRjiheChHgdM+JyqdgNcmzZOX/tnl0JOiI7A==",
       "dev": true
     },
+    "node_modules/@msgpackr-extract/msgpackr-extract-darwin-arm64": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-arm64/-/msgpackr-extract-darwin-arm64-3.0.2.tgz",
+      "integrity": "sha512-9bfjwDxIDWmmOKusUcqdS4Rw+SETlp9Dy39Xui9BEGEk19dDwH0jhipwFzEff/pFg95NKymc6TOTbRKcWeRqyQ==",
+      "cpu": [
+        "arm64"
+      ],
+      "optional": true,
+      "os": [
+        "darwin"
+      ]
+    },
+    "node_modules/@msgpackr-extract/msgpackr-extract-darwin-x64": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-x64/-/msgpackr-extract-darwin-x64-3.0.2.tgz",
+      "integrity": "sha512-lwriRAHm1Yg4iDf23Oxm9n/t5Zpw1lVnxYU3HnJPTi2lJRkKTrps1KVgvL6m7WvmhYVt/FIsssWay+k45QHeuw==",
+      "cpu": [
+        "x64"
+      ],
+      "optional": true,
+      "os": [
+        "darwin"
+      ]
+    },
+    "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm/-/msgpackr-extract-linux-arm-3.0.2.tgz",
+      "integrity": "sha512-MOI9Dlfrpi2Cuc7i5dXdxPbFIgbDBGgKR5F2yWEa6FVEtSWncfVNKW5AKjImAQ6CZlBK9tympdsZJ2xThBiWWA==",
+      "cpu": [
+        "arm"
+      ],
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm64": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm64/-/msgpackr-extract-linux-arm64-3.0.2.tgz",
+      "integrity": "sha512-FU20Bo66/f7He9Fp9sP2zaJ1Q8L9uLPZQDub/WlUip78JlPeMbVL8546HbZfcW9LNciEXc8d+tThSJjSC+tmsg==",
+      "cpu": [
+        "arm64"
+      ],
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@msgpackr-extract/msgpackr-extract-linux-x64": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-x64/-/msgpackr-extract-linux-x64-3.0.2.tgz",
+      "integrity": "sha512-gsWNDCklNy7Ajk0vBBf9jEx04RUxuDQfBse918Ww+Qb9HCPoGzS+XJTLe96iN3BVK7grnLiYghP/M4L8VsaHeA==",
+      "cpu": [
+        "x64"
+      ],
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@msgpackr-extract/msgpackr-extract-win32-x64": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-win32-x64/-/msgpackr-extract-win32-x64-3.0.2.tgz",
+      "integrity": "sha512-O+6Gs8UeDbyFpbSh2CPEz/UOrrdWPTBYNblZK5CxxLisYt4kGX3Sc+czffFonyjiGSq3jWLwJS/CCJc7tBr4sQ==",
+      "cpu": [
+        "x64"
+      ],
+      "optional": true,
+      "os": [
+        "win32"
+      ]
+    },
     "node_modules/@next/env": {
       "version": "14.0.3",
       "resolved": "https://registry.npmjs.org/@next/env/-/env-14.0.3.tgz",
@@ -18867,6 +18940,35 @@
       "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.2.tgz",
       "integrity": "sha512-sGkPx+VjMtmA6MX27oA4FBFELFCZZ4S4XqeGOXCv68tT+jb3vk/RyaKWP0PTKyWtmLSM0b+adUTEvbs1PEaH2w=="
     },
+    "node_modules/msgpackr": {
+      "version": "1.10.0",
+      "resolved": "https://registry.npmjs.org/msgpackr/-/msgpackr-1.10.0.tgz",
+      "integrity": "sha512-rVQ5YAQDoZKZLX+h8tNq7FiHrPJoeGHViz3U4wIcykhAEpwF/nH2Vbk8dQxmpX5JavkI8C7pt4bnkJ02ZmRoUw==",
+      "optionalDependencies": {
+        "msgpackr-extract": "^3.0.2"
+      }
+    },
+    "node_modules/msgpackr-extract": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/msgpackr-extract/-/msgpackr-extract-3.0.2.tgz",
+      "integrity": "sha512-SdzXp4kD/Qf8agZ9+iTu6eql0m3kWm1A2y1hkpTeVNENutaB0BwHlSvAIaMxwntmRUAUjon2V4L8Z/njd0Ct8A==",
+      "hasInstallScript": true,
+      "optional": true,
+      "dependencies": {
+        "node-gyp-build-optional-packages": "5.0.7"
+      },
+      "bin": {
+        "download-msgpackr-prebuilds": "bin/download-prebuilds.js"
+      },
+      "optionalDependencies": {
+        "@msgpackr-extract/msgpackr-extract-darwin-arm64": "3.0.2",
+        "@msgpackr-extract/msgpackr-extract-darwin-x64": "3.0.2",
+        "@msgpackr-extract/msgpackr-extract-linux-arm": "3.0.2",
+        "@msgpackr-extract/msgpackr-extract-linux-arm64": "3.0.2",
+        "@msgpackr-extract/msgpackr-extract-linux-x64": "3.0.2",
+        "@msgpackr-extract/msgpackr-extract-win32-x64": "3.0.2"
+      }
+    },
     "node_modules/multicast-dns": {
       "version": "7.2.5",
       "resolved": "https://registry.npmjs.org/multicast-dns/-/multicast-dns-7.2.5.tgz",
@@ -19131,6 +19233,17 @@
         "node-gyp-build-test": "build-test.js"
       }
     },
+    "node_modules/node-gyp-build-optional-packages": {
+      "version": "5.0.7",
+      "resolved": "https://registry.npmjs.org/node-gyp-build-optional-packages/-/node-gyp-build-optional-packages-5.0.7.tgz",
+      "integrity": "sha512-YlCCc6Wffkx0kHkmam79GKvDQ6x+QZkMjFGrIMxgFNILFvGSbCp2fCBC55pGTT9gVaz8Na5CLmxt/urtzRv36w==",
+      "optional": true,
+      "bin": {
+        "node-gyp-build-optional-packages": "bin.js",
+        "node-gyp-build-optional-packages-optional": "optional.js",
+        "node-gyp-build-optional-packages-test": "build-test.js"
+      }
+    },
     "node_modules/node-int64": {
       "version": "0.4.0",
       "resolved": "https://registry.npmjs.org/node-int64/-/node-int64-0.4.0.tgz",

package.json

@@ -40,6 +40,7 @@
     "http-proxy": "^1.18.1",
     "http-proxy-middleware": "^2.0.6",
     "js-base64": "^3.7.5",
+    "msgpackr": "^1.10.0",
     "next": "^14.0.3",
     "react": "^18.2.0",
     "react-dom": "^18.2.0",