cameras: fix fetch timeout bugs

plugins/amcrest/package.json

@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/amcrest",
-   "version": "0.0.142",
+   "version": "0.0.143",
    "description": "Amcrest Plugin for Scrypted",
    "author": "Scrypted",
    "license": "Apache",

plugins/hikvision/package.json

@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/hikvision",
-   "version": "0.0.144",
+   "version": "0.0.145",
    "description": "Hikvision Plugin for Scrypted",
    "author": "Scrypted",
    "license": "Apache",

plugins/onvif/package.json

@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/onvif",
-   "version": "0.1.11",
+   "version": "0.1.12",
    "description": "ONVIF Camera Plugin for Scrypted",
    "author": "Scrypted",
    "license": "Apache",

plugins/reolink/package.json

@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/reolink",
-   "version": "0.0.63",
+   "version": "0.0.64",
    "description": "Reolink Plugin for Scrypted",
    "author": "Scrypted",
    "license": "Apache",

plugins/snapshot/package.json

@@ -1,6 +1,6 @@
 {
    "name": "@scrypted/snapshot",
-   "version": "0.2.40",
+   "version": "0.2.41",
    "description": "Snapshot Plugin for Scrypted",
    "scripts": {
       "scrypted-setup-project": "scrypted-setup-project",

server/src/fetch/http-fetch.ts

@@ -93,44 +93,61 @@ export async function httpFetch<T extends HttpFetchOptions<Readable>>(options: T
         }
     }
 
+    let controller: AbortController;
+    let timeout: NodeJS.Timeout;
+    if (options.timeout) {
+        controller = new AbortController();
+        timeout = setTimeout(() => controller.abort(), options.timeout);
+    }
+
     const request = proto.request(url, {
         method: getFetchMethod(options),
         rejectUnauthorized: options.rejectUnauthorized,
         family: options.family,
         headers: nodeHeaders,
+        signal: controller?.signal || options.signal,
         timeout: options.timeout,
     });
 
-    options.signal?.addEventListener('abort', () => request.destroy(new Error('abort')));
+    if (controller)
+        options.signal?.addEventListener('abort', () => controller.abort('abort'));
+    else
+        options.signal?.addEventListener('abort', () => request.destroy(new Error('abort')));
 
     if (body)
         body.pipe(request);
     else
         request.end();
-    const [response] = await once(request, 'response') as [IncomingMessage];
 
-    if (!options?.ignoreStatusCode) {
-        try {
-            checkStatus(response.statusCode);
-        }
-        catch (e) {
-            readMessageBuffer(response).catch(() => { });
-            throw e;
+    try {
+        const [response] = await once(request, 'response') as [IncomingMessage];
+
+        if (!options?.ignoreStatusCode) {
+            try {
+                checkStatus(response.statusCode);
+            }
+            catch (e) {
+                readMessageBuffer(response).catch(() => { });
+                throw e;
+            }
         }
-    }
 
-    const incomingHeaders = new Headers();
-    for (const [k, v] of Object.entries(response.headers)) {
-        for (const vv of (typeof v === 'string' ? [v] : v)) {
-            incomingHeaders.append(k, vv)
+        const incomingHeaders = new Headers();
+        for (const [k, v] of Object.entries(response.headers)) {
+            for (const vv of (typeof v === 'string' ? [v] : v)) {
+                incomingHeaders.append(k, vv)
+            }
         }
-    }
 
-    return {
-        statusCode: response.statusCode,
-        headers: incomingHeaders,
-        body: await httpFetchParseIncomingMessage(response, options.responseType),
-    };
+        return {
+            statusCode: response.statusCode,
+            headers: incomingHeaders,
+            body: await httpFetchParseIncomingMessage(response, options.responseType),
+        };
+    }
+    finally {
+        clearTimeout(timeout);
+    }
 }
 
 function ensureType<T>(v: T) {

server/src/fetch/index.ts

@@ -124,30 +124,42 @@ export async function domFetch<T extends HttpFetchOptions<BodyInit>>(options: T)
         body = createStringOrBufferBody(headers, body);
     }
 
-    const { url } = options;
-    const response = await fetch(url, {
-        method: getFetchMethod(options),
-        credentials: options.withCredentials ? 'include' : undefined,
-        headers,
-        signal: options.signal || options.timeout ? AbortSignal.timeout(options.timeout) : undefined,
-        body,
-    });
+    let controller: AbortController;
+    let timeout: NodeJS.Timeout;
+    if (options.timeout) {
+        controller = new AbortController();
+        timeout = setTimeout(() => controller.abort(), options.timeout);
+    }
 
-    if (!options?.ignoreStatusCode) {
-        try {
-            checkStatus(response.status);
-        }
-        catch (e) {
-            response.arrayBuffer().catch(() => { });
-            throw e;
+    try {
+        const { url } = options;
+        const response = await fetch(url, {
+            method: getFetchMethod(options),
+            credentials: options.withCredentials ? 'include' : undefined,
+            headers,
+            signal: controller?.signal || options.signal,
+            body,
+        });
+
+        if (!options?.ignoreStatusCode) {
+            try {
+                checkStatus(response.status);
+            }
+            catch (e) {
+                response.arrayBuffer().catch(() => { });
+                throw e;
+            }
         }
-    }
 
-    return {
-        statusCode: response.status,
-        headers: response.headers,
-        body: await domFetchParseIncomingMessage(response, options.responseType),
-    };
+        return {
+            statusCode: response.status,
+            headers: response.headers,
+            body: await domFetchParseIncomingMessage(response, options.responseType),
+        };
+    }
+    finally {
+        clearTimeout(timeout);
+    }
 }
 
 function ensureType<T>(v: T) {