Add receiver timeout to StreamProvider

README.md

@@ -22,7 +22,7 @@ This app allows remotely viewing source-specific multicast RTP streams (e.g. Deu
 
 Or with Docker (Linux only):
 1. [Install](https://docs.docker.com/v17.12/install/) Docker
-2. run `docker run -d --network host --restart always --name IPTV-ReStream [-e HOST="127.0.0.1" | -e PORT=3000 | -e MCAST_IF="0.0.0.0" | -e XSPF_PROTOCOL="https" | -e XSPF_HOST="my.server.com:8080" | -e XSPF_PATH_PREFIX="/iptv" | -e ALLOW_UNKNOWN="false" | -e DEBUG="iptv-restream:*"] nthumann/iptv-restream:latest` (parameters in brackets are optional)
+2. run `docker run -d --network host --restart always --name IPTV-ReStream [-e HOST="127.0.0.1" | -e PORT=3000 | -e MCAST_IF="0.0.0.0" | -e XSPF_PROTOCOL="https" | -e XSPF_HOST="my.server.com:8080" | -e XSPF_PATH_PREFIX="/iptv" | -e ALLOW_UNKNOWN="false" | -e RECEIVER_TIMEOUT=30 | -e DEBUG="iptv-restream:*"] nthumann/iptv-restream:latest` (parameters in brackets are optional)
 
 The image is also available on GitHub Container Registry (use `ghcr.io/n-thumann/iptv-restream:latest`).
 ### Configuration ###
@@ -38,6 +38,7 @@ It's highly recommended running this application behind a Reverse Proxy ([nginx]
 | XSPF_HOST | Host and Port for generating XSPF |
 | XSPF_PATH_PREFIX | Path Prefix for generating XSPF |
 | ALLOW_UNKNOWN | Allow forwarding streams that are not in the station list (default `0`/`false`) |
+| RECEIVER_TIMEOUT | Seconds with not data received after which the connection is closed (default `30`) |
 | DEBUG | Enable Debug Logging (`iptv-restream:*`) |
 
 ### Usage

src/providers/config.ts

@@ -7,6 +7,7 @@ class ConfigProvider {
     readonly xspf_host: string = '';
     readonly xspf_pathPrefix: string =  '';
     readonly allow_unknown: boolean;
+    readonly receiver_timeout: number;
 
     constructor() {
         this.host = process.env.HOST || '127.0.0.1';
@@ -16,8 +17,9 @@ class ConfigProvider {
         this.xspf_host = process.env.XSPF_HOST || '';
         this.xspf_pathPrefix = process.env.XSPF_PATH_PREFIX || '';
         this.allow_unknown = (process.env.ALLOW_UNKNOWN === 'true' || process.env.ALLOW_UNKNOWN === '1') ? true : false;
+        this.receiver_timeout = parseInt(process.env.RECEIVER_TIMEOUT || '30');
         const logger = debug('iptv-restream:config');
-        logger(`Config loaded: HOST=${this.host}, PORT=${this.port}, MCAST_IF=${this.mcast_if}, XSPF_PROTOCOL=${this.xspf_protocol}, XSPF_HOST=${this.xspf_host}, XSPF_PATH_PREFIX=${this.xspf_pathPrefix}, ALLOW_UNKNOWN=${this.allow_unknown}`);
+        logger(`Config loaded: HOST=${this.host}, PORT=${this.port}, MCAST_IF=${this.mcast_if}, XSPF_PROTOCOL=${this.xspf_protocol}, XSPF_HOST=${this.xspf_host}, XSPF_PATH_PREFIX=${this.xspf_pathPrefix}, ALLOW_UNKNOWN=${this.allow_unknown}, RECEIVER_TIMEOUT=${this.receiver_timeout}`);
     }
 }
 export default new ConfigProvider();

src/providers/stream.ts

@@ -1,6 +1,7 @@
 import { Socket } from 'net';
 import { Response } from 'express';
 import connectionProvider from './connection';
+import configProvider from '../providers/config';
 import dgram from 'dgram';
 import debug from 'debug';
 
@@ -12,6 +13,11 @@ class StreamProvider {
             this.logger(`Client requested ${mcast_source}@${mcast_group}:${mcast_port}.`);
             const receiver = dgram.createSocket({ type: 'udp4', reuseAddr: true });
 
+            const timeout = setTimeout(() => {
+                this.logger('data reception timed out')
+                reject('data reception timed out')
+            }, configProvider.receiver_timeout * 1_000);
+
             receiver.on('error', (err: Error) => {
                 this.logger(`Error (receiver): ${err}`);
                 reject(`Error (receiver): ${err}`);
@@ -34,6 +40,7 @@ class StreamProvider {
 
                 const mpegtsPacket = Buffer.from(message).slice(12);
                 res.write(mpegtsPacket);
+                timeout.refresh();
 
                 const packetCount = mpegtsPacket.length / 188;
                 for (let packetNumber = 0; packetNumber < packetCount; packetNumber++) {
@@ -72,6 +79,7 @@ class StreamProvider {
 
             socket.on('close', () => {
                 receiver.close();
+                clearTimeout(timeout);
                 this.logger(`Client ${socket.remoteAddress}:${socket.remotePort} disconnected. Closing receiver.`);
                 resolve();
             });