Merge pull request #244 from sensslen/simon/use-setreaming-api

Use streaming api rather than polling

Author: sensslen

package-lock.json

@@ -12,12 +12,12 @@
     "preview": "vite preview"
   },
   "dependencies": {
-    "axios": "^1.7.3",
     "i18next": "^25.0.0",
     "i18next-browser-languagedetector": "^8.0.0",
     "react": "^19",
     "react-dom": "^19",
-    "react-i18next": "^16.0.0"
+    "react-i18next": "^16.0.0",
+    "zod": "^4.3.6"
   },
   "devDependencies": {
     "@eslint/js": "^9",

package.json

@@ -125,7 +125,7 @@ const App: React.FC = () => {
   useEffect(() => {
     if (error) {
       const timer = setTimeout(() => {
-        setError('');
+        setError(null);
       }, 3000); // Clear the error message after 3 seconds
       return () => clearTimeout(timer);
     }

src/App.tsx

@@ -1,7 +1,49 @@
-import axios from 'axios';
+import { z } from 'zod';
 import { Message, TriggerPayload } from '../types/proPresenter';
 
-  const isValidUrl = (inputUrl: string): boolean => {
+// Custom error class for stream failures with typed status code
+export class StreamError extends Error {
+    readonly statusCode: number;
+    readonly name = 'StreamError';
+
+    constructor(message: string, statusCode: number) {
+        super(message);
+        this.statusCode = statusCode;
+        Object.setPrototypeOf(this, StreamError.prototype);
+    }
+}
+
+// Zod schema for Message validation
+const messageIdSchema = z.object({
+    uuid: z.string(),
+    index: z.number(),
+    name: z.string()
+});
+
+const messageObjectSchema = z.object({
+    id: messageIdSchema,
+    message: z.string(),
+    tokens: z.array(z.any()), // tokens have complex optional structure, validate as array of any
+    visible_on_network: z.boolean()
+});
+
+// Accept either a single message or an array of messages
+const messageSchema = z.union([
+    messageObjectSchema,
+    z.array(messageObjectSchema)
+]);
+
+const validateMessage = (data: unknown) => {
+    const result = messageSchema.safeParse(data);
+    // Cast to proper Message type since we've validated the structure
+    return {
+        success: result.success,
+        data: result.success ? (result.data as Message | Message[]) : undefined,
+        error: result.error
+    };
+};
+
+const isValidUrl = (inputUrl: string): boolean => {
     try {
       // Use URL constructor for basic format validation
       new URL(inputUrl);
@@ -11,50 +53,216 @@ import { Message, TriggerPayload } from '../types/proPresenter';
     }
   };
 
-export const getMessages = async (url: string): Promise<Message[]> => {
-    if (!isValidUrl(url)) {
-        throw new Error('Invalid URL format');
+// Detect if an error is an AbortError
+const isAbortError = (err: unknown): boolean => {
+    if (err instanceof DOMException && err.name === 'AbortError') {
+        return true;
     }
-
-    const resource = `${url}/v1/messages`;
-    try {
-        const response = await axios.get<Message[]>(resource);
-        return response.data;
-    } catch (error) {
-        handleApiError(error);
-        throw new Error(`Failed to fetch messages from ${resource}`);
+    if ((err as { name?: string })?.name === 'AbortError') {
+        return true;
     }
+    return false;
 };
 
+// Stream messages using the chunked endpoint. Returns an AbortController to stop the stream.
+// The function returns the controller synchronously and runs the async connection logic in the background.
+export const streamMessages = (
+    url: string,
+    onChunk: (data: Message[] | Message) => void,
+    onOpen?: () => void,
+    onClose?: () => void,
+    onError?: (err: unknown) => void,
+): AbortController => {
+    const controller = new AbortController();
+
+    // Run connection logic asynchronously in the background without awaiting
+    (async () => {
+        // Validate URL synchronously before starting async work
+        if (!isValidUrl(url)) {
+            // Defer error/close callbacks to after controller is returned to caller
+            queueMicrotask(() => {
+                onError && onError(new Error('Invalid URL format'));
+                onClose && onClose();
+            });
+            return;
+        }
+
+        const resource = `${url}/v1/messages?chunked=true`;
+
+        try {
+            const response = await fetch(resource, { signal: controller.signal });
+            if (!response.ok) {
+                throw new StreamError(`Failed to stream messages: ${response.status}`, response.status);
+            }
+
+            onOpen && onOpen();
+
+            const reader = response.body?.getReader();
+            if (!reader) {
+                throw new Error('Stream reader not available');
+            }
+
+            const decoder = new TextDecoder();
+            let buffer = '';
+
+            const extractNextJson = (buf: string): { json: string; rest: string } | null => {
+                // Find first non-whitespace char
+                const startIdx = buf.search(/\S/);
+                if (startIdx === -1) return null;
+                const startChar = buf[startIdx];
+                if (startChar !== '{' && startChar !== '[') return null;
+
+                let depth = 0;
+                let inString = false;
+                let escape = false;
+                for (let i = startIdx; i < buf.length; i++) {
+                    const ch = buf[i];
+                    if (inString) {
+                        if (escape) {
+                            escape = false;
+                        } else if (ch === '\\') {
+                            escape = true;
+                        } else if (ch === '"') {
+                            inString = false;
+                        }
+                        continue;
+                    }
+
+                    if (ch === '"') {
+                        inString = true;
+                        continue;
+                    }
+
+                    if (ch === '{' || ch === '[') {
+                        depth++;
+                    } else if (ch === '}' || ch === ']') {
+                        depth--;
+                        if (depth === 0) {
+                            const json = buf.slice(startIdx, i + 1);
+                            const rest = buf.slice(i + 1);
+                            return { json, rest };
+                        }
+                    }
+                }
+
+                return null;
+            };
+
+            // Helper to process an extracted JSON chunk
+            const processChunkBuffer = (
+                extracted: { json: string; rest: string } | null,
+                errorContext: string
+            ): { parsedData: Message | Message[] | undefined; rest: string } => {
+                let parsedData: Message | Message[] | undefined;
+                
+                if (!extracted) {
+                    return { parsedData: undefined, rest: '' };
+                }
+
+                try {
+                    const parsed = JSON.parse(extracted.json);
+                    const validation = validateMessage(parsed);
+                    if (validation.success && validation.data) {
+                        parsedData = validation.data;
+                    } else {
+                        console.error(`Invalid message format in ${errorContext}`, validation.error);
+                    }
+                } catch (err) {
+                    console.error(`Failed to parse ${errorContext}`, err);
+                }
+
+                // Call onChunk outside try/catch to handle callback errors separately
+                if (parsedData !== undefined) {
+                    try {
+                        onChunk(parsedData);
+                    } catch (err) {
+                        onError && onError(err);
+                    }
+                }
+
+                return { parsedData, rest: extracted.rest };
+            };
+
+            const pump = async () => {
+                try {
+                    while (true) {
+                        const { value, done } = await reader.read();
+                        if (done) {
+                            // Flush the TextDecoder to get any remaining bytes
+                            buffer += decoder.decode();
+                            
+                            // Drain the entire buffer using the same while-loop as streaming
+                            let extracted = extractNextJson(buffer);
+                            while (extracted) {
+                                const { rest } = processChunkBuffer(extracted, 'chunk');
+                                buffer = rest;
+                                extracted = extractNextJson(buffer);
+                            }
+                            onClose && onClose();
+                            break;
+                        }
+
+                        buffer += decoder.decode(value, { stream: true });
+
+                        let extracted = extractNextJson(buffer);
+                        while (extracted) {
+                            const { rest } = processChunkBuffer(extracted, 'stream');
+                            buffer = rest;
+                            extracted = extractNextJson(buffer);
+                        }
+                    }
+                } catch (err) {
+                    // Only call onError if this is not an abort error (user/code explicitly cancelled)
+                    if (!isAbortError(err)) {
+                        onError && onError(err);
+                    }
+                }
+            };
+
+            // Start background pump (don't await)
+            void pump();
+        } catch (error) {
+            // Only call onError if this is not an abort error (user/code explicitly cancelled)
+            if (!isAbortError(error)) {
+                onError && onError(error);
+            }
+        }
+    })();
+
+    return controller;
+};
 export const triggerMessage = async (url: string, id: string, payload: TriggerPayload): Promise<void> => {
     const resource = `${url}/v1/message/${id}/trigger`;
     try {
-        await axios.post(resource, payload);
+        const response = await fetch(resource, {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify(payload)
+        });
+        if (!response.ok) {
+            throw new StreamError(`Failed to trigger message: ${response.status}`, response.status);
+        }
     } catch (error) {
-        handleApiError(error);
+        if (error instanceof StreamError) {
+            throw error;
+        }
+        console.error('Error triggering message:', error);
         throw new Error(`Failed to trigger message at ${resource} with payload: ${JSON.stringify(payload)}`);
     }
 };
 
 export const clearMessage = async (url: string, id: string): Promise<void> => {
     const resource = `${url}/v1/message/${id}/clear`;
     try {
-        await axios.get(resource);
+        const response = await fetch(resource, { method: 'GET' });
+        if (!response.ok) {
+            throw new StreamError(`Failed to clear message: ${response.status}`, response.status);
+        }
     } catch (error) {
-        handleApiError(error);
-        throw new Error(`Failed to clear message at ${resource}`);
-    }
-};
-
-const handleApiError = (error: unknown) => {
-    if (axios.isAxiosError(error)) {
-        // Handle Axios-specific errors
-        console.error('API Error:', error.message);
-        if (error.response) {
-            console.error('Response Data:', error.response.data);
+        if (error instanceof StreamError) {
+            throw error;
         }
-    } else {
-        // Handle other types of errors
-        console.error('Unexpected Error:', error);
+        console.error('Error clearing message:', error);
+        throw new Error(`Failed to clear message at ${resource}`);
     }
 };

src/api/proPresenter.ts

@@ -1,7 +1,8 @@
-import React, { useState } from 'react';
+import React, { useState, useCallback } from 'react';
 import { Message } from '../types/proPresenter';
 import TokenInput from './TokenInput';
 import { Trans, useTranslation } from 'react-i18next';
+import { replaceAllTokens } from '../utils/tokenReplacement';
 
 interface MessageItemProps {
     message: Message;
@@ -22,12 +23,9 @@ const MessageItem: React.FC<MessageItemProps> = ({ message, onShowMessage, onHid
         return initialTokenValues;
     });
 
-    const renderMessageWithTokens = (message: string): string => {
-        Object.entries(tokenValues).forEach(([name, value]) => {
-            message = message.replace(`{${name}}`, value);
-        });
-        return message;
-    };
+    const renderMessageWithTokens = useCallback((messageStr: string): string => {
+        return replaceAllTokens(messageStr, tokenValues);
+    }, [tokenValues]);
 
     return (
         <li className="mb-4 p-2 border">