Add EventSource (#51)

Signed-off-by: Kevin Poirot <kevin@vazee.fr>

Author: VazeeKevin

README.md

@@ -6,7 +6,7 @@
 
 # React LogViewer
 
-React component that loads and views remote text in the browser lazily and efficiently. Logs can be loaded from static text, a URL, or a WebSocket and including ANSI highlighting.
+React component that loads and views remote text in the browser lazily and efficiently. Logs can be loaded from static text, a URL, a WebSocket, or an EventSource and including ANSI highlighting.
 Forked from [mozilla-frontend-infra/react-lazylog](https://github.com/mozilla-frontend-infra/react-lazylog).
 
 **If you like this project, please consider supporting me ❤️**

src/components/LazyLog/README.md

@@ -63,3 +63,22 @@ let socket = null;
   </div>
 </div>
 ```
+
+Log viewing using an eventsource
+
+```jsx harmony
+const url = 'https://my.eventsource.tld';
+
+<div>
+  <div style={{ height: 200, width: 902 }}>
+  <LazyLog
+    enableSearch
+    url={url}
+    eventsource
+    eventsourceOptions={{
+      formatMessage: e => JSON.parse(e).message,
+    }}
+  />
+  </div>
+</div>
+```

src/components/LazyLog/index.tsx

@@ -20,6 +20,7 @@ import {
     searchFormatPart,
 } from "../Utils/utils";
 import websocket from "../Utils/websocket";
+import eventsource from "../Utils/eventsource";
 import styles from "./index.module.css";
 
 export interface WebsocketOptions {
@@ -36,7 +37,7 @@ export interface WebsocketOptions {
      */
     onError?: ((e: Event) => void) | undefined;
     /**
-     * Callback allback which formats the websocket data stream.
+     * Callback which formats the websocket data stream.
      */
     formatMessage?: ((message: any) => string) | undefined;
     /**
@@ -50,6 +51,38 @@ export interface WebsocketOptions {
     reconnectWait?: number;
 }
 
+export interface EventSourceOptions {
+    /**
+     * Boolean indicating if CORS should be set to include credentials
+     */
+    withCredentials?: boolean;
+    /**
+     * Callback when the eventsource is opened
+     */
+    onOpen?: ((e: Event, eventSource: EventSource) => void) | undefined;
+    /**
+     * Callback when the eventsource is closed
+     */
+    onClose?: ((e: Event) => void) | undefined;
+    /**
+     * Callback when the eventsource has an error
+     */
+    onError?: ((e: Event) => void) | undefined;
+    /**
+     * Callback which formats the eventsource data stream.
+     */
+    formatMessage?: ((message: any) => string) | undefined;
+    /**
+     * Set to true, to reconnect the EventSource automatically.
+     */
+    reconnect?: boolean;
+    /**
+     * Set the time to wait between reconnects in seconds.
+     * Default is 1s
+     */
+    reconnectWait?: number;
+}
+
 export interface ErrorStatus extends Error {
     /**
      * Status code
@@ -262,6 +295,15 @@ export interface LazyLogProps {
      * Options object which will be passed through to websocket.
      */
     websocketOptions?: WebsocketOptions;
+    /**
+     * Set to `true` to specify that url is an eventsource URL.
+     * Defaults to `false` to download data until completion.
+     */
+    eventsource?: boolean;
+    /**
+     * Options object which will be passed through to evensource.
+     */
+    eventsourceOptions?: EventSourceOptions;
     /**
      * Set the width in pixels for the component.
      * Defaults to `'auto'` if unspecified.
@@ -331,6 +373,8 @@ export default class LazyLog extends Component<LazyLogProps, LazyLogState> {
         style: {},
         websocket: false,
         websocketOptions: {},
+        eventsource: false,
+        eventsourceOptions: {},
         width: "auto",
     };
 
@@ -477,15 +521,21 @@ export default class LazyLog extends Component<LazyLogProps, LazyLogState> {
         const {
             stream: isStream,
             websocket: isWebsocket,
+            eventsource: isEventsource,
             url,
             fetchOptions,
             websocketOptions,
+            eventsourceOptions,
         } = this.props;
 
         if (isWebsocket) {
             return websocket(url!, websocketOptions!);
         }
 
+        if (isEventsource) {
+            return eventsource(url!, eventsourceOptions!);
+        }
+
         if (isStream) {
             return stream(url!, fetchOptions);
         }
@@ -529,10 +579,10 @@ export default class LazyLog extends Component<LazyLogProps, LazyLogState> {
 
     handleUpdate = ({ lines: moreLines, encodedLog }: any) => {
         this.encodedLog = encodedLog;
-        const { scrollToLine, follow, stream, websocket } = this.props;
+        const { scrollToLine, follow, stream, websocket, eventsource } = this.props;
 
-        // handle stream and socket updates batched update mode
-        if (stream || websocket) {
+        // handle stream, socket and eventsource updates batched update mode
+        if (stream || websocket || eventsource) {
             this.setState((state, props) => {
                 const { scrollToLine, follow } = props;
                 const { count: previousCount } = state;
@@ -736,9 +786,9 @@ export default class LazyLog extends Component<LazyLogProps, LazyLogState> {
 
     handleSearch = (keywords: string | undefined) => {
         const { resultLines, searchKeywords } = this.state;
-        const { caseInsensitive, stream, websocket } = this.props;
+        const { caseInsensitive, stream, websocket, eventsource } = this.props;
         const currentResultLines =
-            !stream && !websocket && keywords === searchKeywords
+            !stream && !websocket && !eventsource && keywords === searchKeywords
                 ? resultLines
                 : searchLines(keywords, this.encodedLog!, caseInsensitive!);
 

src/components/Utils/eventsource.ts

@@ -0,0 +1,79 @@
+import { List } from "immutable";
+import mitt from "mitt";
+
+import { EventSourceOptions } from "../LazyLog";
+import { encode } from "./encoding";
+import { bufferConcat, convertBufferToLines } from "./utils";
+
+export default (url: string | URL, options: EventSourceOptions) => {
+    const { withCredentials, onOpen, onClose, onError, formatMessage } = options;
+    const emitter = mitt();
+    let encodedLog = new Uint8Array();
+    let overage: any = null;
+    let aborted: boolean = false;
+
+    emitter.on("data", (data) => {
+        encodedLog = bufferConcat(
+            encodedLog,
+            encode(data as unknown as string)
+        );
+
+        const { lines, remaining } = convertBufferToLines(
+            encode(data as unknown as string),
+            overage
+        );
+
+        overage = remaining;
+
+        emitter.emit("update", { lines, encodedLog });
+    });
+
+    emitter.on("done", () => {
+        if (overage) {
+            emitter.emit("update", { lines: List.of(overage), encodedLog });
+        }
+
+        emitter.emit("end", encodedLog);
+    });
+
+    emitter.on("start", () => {
+        try {
+            // try to connect to eventSource
+            const eventSource = new EventSource(url, { withCredentials });
+
+            eventSource.addEventListener("open", (e) => {
+                // relay on open events if a handler is registered
+                onOpen && onOpen(e, eventSource);
+            });
+
+            eventSource.addEventListener("close", (e) => {
+                onClose && onClose(e);
+                if(!aborted && options.reconnect) {
+                    const timeout = options.reconnectWait ?? 1;
+                    setTimeout(() => emitter.emit("start"), timeout*1000);
+                }
+            });
+
+            eventSource.addEventListener("error", (err) => {
+                onError && onError(err);
+            });
+
+            eventSource.addEventListener("message", (e) => {
+                let msg = formatMessage ? formatMessage(e.data) : e.data;
+
+                if (typeof msg !== "string") {
+                    return;
+                }
+                // add a new line character between each message if one doesn't exist.
+                // this allows our search index to properly distinguish new lines.
+                msg = msg.endsWith("\n") ? msg : `${msg}\n`;
+
+                emitter.emit("data", msg);
+            });
+        } catch (err) {
+            emitter.emit("error", err);
+        }
+    });
+
+    return emitter;
+};