code refactoring, switch to rxjs websocket wrapper

.github/workflows/docker-publish.yml

@@ -14,7 +14,7 @@ on:
   pull_request:
     branches: [ master ]
   schedule:
-    - cron : '0 1 * * *'
+    - cron : '0 1 * * 0'
 
 env:
   # Use docker.io for Docker Hub if empty

frontend/src/App.tsx

@@ -32,7 +32,7 @@ import { AppBar } from './components/AppBar'
 import { Drawer } from './components/Drawer'
 import { toggleListView } from './features/settings/settingsSlice'
 import { RootState, store } from './stores/store'
-import { formatGiB, getWebSocketEndpoint } from './utils'
+import { formatGiB } from './utils'
 
 function AppContent() {
   const [open, setOpen] = useState(false)
@@ -41,8 +41,6 @@ function AppContent() {
   const status = useSelector((state: RootState) => state.status)
   const dispatch = useDispatch()
 
-  const socket = useMemo(() => new WebSocket(getWebSocketEndpoint()), [])
-
   const mode = settings.theme
   const theme = useMemo(() =>
     createTheme({
@@ -170,10 +168,10 @@ function AppContent() {
           >
             <Toolbar />
             <Routes>
-              <Route path="/" element={<Home socket={socket} />} />
+              <Route path="/" element={<Home />} />
               <Route path="/settings" element={
                 <Suspense fallback={<CircularProgress />}>
-                  <Settings socket={socket} />
+                  <Settings />
                 </Suspense>
               } />
             </Routes>

frontend/src/Home.tsx

@@ -25,17 +25,13 @@ import { DownloadsListView } from './components/DownloadsListView'
 import FormatsGrid from './components/FormatsGrid'
 import { CliArguments } from './features/core/argsParser'
 import I18nBuilder from './features/core/intl'
-import { RPCClient } from './features/core/rpcClient'
+import { RPCClient, socket$ } from './features/core/rpcClient'
 import { connected, setFreeSpace } from './features/status/statusSlice'
 import { RootState } from './stores/store'
-import type { DLMetadata, RPCResult } from './types'
+import type { DLMetadata, RPCResponse, RPCResult } from './types'
 import { isValidURL, toFormatArgs } from './utils'
 
-type Props = {
-  socket: WebSocket
-}
-
-export default function Home({ socket }: Props) {
+export default function Home() {
   // redux state
   const settings = useSelector((state: RootState) => state.settings)
   const status = useSelector((state: RootState) => state.status)
@@ -60,9 +56,11 @@ export default function Home({ socket }: Props) {
   const [showBackdrop, setShowBackdrop] = useState(true)
   const [showToast, setShowToast] = useState(true)
 
+  const [socketHasError, setSocketHasError] = useState(false)
+
   // memos
   const i18n = useMemo(() => new I18nBuilder(settings.language), [settings.language])
-  const client = useMemo(() => new RPCClient(socket), [settings.serverAddr, settings.serverPort])
+  const client = useMemo(() => new RPCClient(), [settings.serverAddr, settings.serverPort])
   const cliArgs = useMemo(() => new CliArguments().fromString(settings.cliArgs), [settings.cliArgs])
 
   // refs
@@ -73,12 +71,19 @@ export default function Home({ socket }: Props) {
 
   /* WebSocket connect event handler*/
   useEffect(() => {
-    socket.onopen = () => {
-      dispatch(connected())
-      setCustomArgs(localStorage.getItem('last-input-args') ?? '')
-      setFilenameOverride(localStorage.getItem('last-filename-override') ?? '')
-    }
-  }, [])
+    const sub = socket$.subscribe({
+      next: () => {
+        dispatch(connected())
+        setCustomArgs(localStorage.getItem('last-input-args') ?? '')
+        setFilenameOverride(localStorage.getItem('last-filename-override') ?? '')
+      },
+      complete: () => {
+        setSocketHasError(true)
+        setShowBackdrop(false)
+      },
+    })
+    return () => sub.unsubscribe()
+  }, [socket$])
 
   useEffect(() => {
     if (status.connected) {
@@ -93,21 +98,23 @@ export default function Home({ socket }: Props) {
   }, [])
 
   useEffect(() => {
-    socket.onmessage = (event) => {
-      const res = client.decode(event.data)
-      switch (typeof res.result) {
-        case 'object':
-          setActiveDownloads(
-            (res.result ?? [])
-              .filter((r: RPCResult) => !!r.info.url)
-              .sort((a: RPCResult, b: RPCResult) => a.info.title.localeCompare(b.info.title))
-          )
-          break
-        default:
-          break
-      }
+    if (status.connected) {
+      const sub = socket$.subscribe((event: RPCResponse<RPCResult[]>) => {
+        switch (typeof event.result) {
+          case 'object':
+            setActiveDownloads(
+              (event.result ?? [])
+                .filter((r) => !!r.info.url)
+                .sort((a, b) => a.info.title.localeCompare(b.info.title))
+            )
+            break
+          default:
+            break
+        }
+      })
+      return () => sub.unsubscribe()
     }
-  }, [])
+  }, [socket$, status.connected])
 
   useEffect(() => {
     if (activeDownloads && activeDownloads.length >= 0) {
@@ -122,18 +129,6 @@ export default function Home({ socket }: Props) {
       })
   }, [])
 
-  const [socketHasError, setSocketHasError] = useState(false)
-
-  useEffect(() => {
-    socket.onerror = () => {
-      setSocketHasError(true)
-      setShowBackdrop(false)
-    }
-    return () => {
-      socket.onerror = null
-    }
-  }, [socket])
-
   /* -------------------- callbacks-------------------- */
 
   /**

frontend/src/Settings.tsx

@@ -40,7 +40,7 @@ import { updated } from './features/status/statusSlice'
 import { RootState } from './stores/store'
 import { validateDomain, validateIP } from './utils'
 
-export default function Settings({ socket }: { socket: WebSocket }) {
+export default function Settings() {
   const dispatch = useDispatch()
 
   const status = useSelector((state: RootState) => state.status)
@@ -50,7 +50,7 @@ export default function Settings({ socket }: { socket: WebSocket }) {
 
   const i18n = useMemo(() => new I18nBuilder(settings.language), [settings.language])
 
-  const client = useMemo(() => new RPCClient(socket), [])
+  const client = useMemo(() => new RPCClient(), [])
   const cliArgs = useMemo(() => new CliArguments().fromString(settings.cliArgs), [])
 
   const serverAddr$ = useMemo(() => new Subject<string>(), [])
@@ -79,6 +79,7 @@ export default function Settings({ socket }: { socket: WebSocket }) {
   useEffect(() => {
     const sub = serverPort$
       .pipe(
+        debounceTime(500),
         map(val => Number(val)),
         takeWhile(val => isFinite(val) && val <= 65535),
       )

frontend/src/components/ArchiveResult.tsx

@@ -1,5 +1,12 @@
-import { Card, CardActionArea, CardContent, CardMedia, Skeleton, Typography } from "@mui/material";
-import { ellipsis } from "../utils";
+import {
+    Card,
+    CardActionArea,
+    CardContent,
+    CardMedia,
+    Skeleton,
+    Typography
+} from '@mui/material'
+import { ellipsis } from '../utils'
 
 type Props = {
     title: string,

frontend/src/components/DownloadsCardView.tsx

@@ -1,5 +1,6 @@
 import { Grid } from "@mui/material"
 import { Fragment } from "react"
+
 import type { RPCResult } from "../types"
 import { StackableResult } from "./StackableResult"
 

frontend/src/components/DownloadsListView.tsx

@@ -1,6 +1,18 @@
-import { Button, CircularProgress, Grid, LinearProgress, Paper, Table, TableBody, TableCell, TableContainer, TableHead, TableRow, Typography } from "@mui/material"
-import { RPCResult } from "../types"
+import {
+  Button,
+  Grid,
+  LinearProgress,
+  Paper,
+  Table,
+  TableBody,
+  TableCell,
+  TableContainer,
+  TableHead,
+  TableRow,
+  Typography
+} from "@mui/material"
 import { ellipsis, formatSpeedMiB, roundMiB } from "../utils"
+import type { RPCResult } from "../types"
 
 type Props = {
   downloads: RPCResult[]

frontend/src/components/FormatsGrid.tsx

@@ -1,5 +1,5 @@
 import { Button, ButtonGroup, Grid, Paper, Typography } from "@mui/material"
-import type { DLMetadata } from "../types"
+import type { DLMetadata } from '../types'
 
 type Props = {
   downloadFormats: DLMetadata

frontend/src/features/core/rpcClient.ts

@@ -1,13 +1,14 @@
-import type { RPCRequest, RPCResponse, DLMetadata } from "../../types"
+import type { DLMetadata, RPCRequest, RPCResponse } from '../../types'
 
-import { getHttpRPCEndpoint } from '../../utils'
+import { webSocket } from 'rxjs/webSocket'
+import { getHttpRPCEndpoint, getWebSocketEndpoint } from '../../utils'
+
+export const socket$ = webSocket<any>(getWebSocketEndpoint())
 
 export class RPCClient {
-  private socket: WebSocket
   private seq: number
 
-  constructor(socket: WebSocket) {
-    this.socket = socket
+  constructor() {
     this.seq = 0
   }
 
@@ -16,7 +17,10 @@ export class RPCClient {
   }
 
   private send(req: RPCRequest) {
-    this.socket.send(JSON.stringify(req))
+    socket$.next({
+      ...req,
+      id: this.incrementSeq(),
+    })
   }
 
   private async sendHTTP<T>(req: RPCRequest) {
@@ -35,7 +39,6 @@ export class RPCClient {
   public download(url: string, args: string, pathOverride = '', renameTo = '') {
     if (url) {
       this.send({
-        id: this.incrementSeq(),
         method: 'Service.Exec',
         params: [{
           URL: url.split("?list").at(0)!,
@@ -50,7 +53,6 @@ export class RPCClient {
   public formats(url: string) {
     if (url) {
       return this.sendHTTP<DLMetadata>({
-        id: this.incrementSeq(),
         method: 'Service.Formats',
         params: [{
           URL: url.split("?list").at(0)!,
@@ -61,7 +63,6 @@ export class RPCClient {
 
   public running() {
     this.send({
-      id: this.incrementSeq(),
       method: 'Service.Running',
       params: [],
     })
@@ -101,8 +102,4 @@ export class RPCClient {
       params: []
     })
   }
-
-  public decode(data: any): RPCResponse<any> {
-    return JSON.parse(data)
-  }
 }

server/rx/extensions.go

@@ -11,6 +11,7 @@ import "time"
 //
 // Debounce emits a string from the source channel only after a particular
 // time span determined a Go Interval
+//
 // --A--B--CD--EFG-------|>
 //
 //	-t->                 |>

server/server.go

@@ -34,6 +34,10 @@ func RunBlocking(port int, frontend fs.FS) {
 	// RPC handlers
 	// websocket
 	app.Get("/ws-rpc", websocket.New(func(c *websocket.Conn) {
+		c.WriteMessage(websocket.TextMessage, []byte(`{
+			"status": "connected"
+		}`))
+
 		for {
 			mtype, reader, err := c.NextReader()
 			if err != nil {