server: torrentService: stop polling when there is no listener

Bug: #45, Flood-UI/flood#405

server/middleware/clientActivityStream.ts

@@ -3,6 +3,7 @@ import type TypedEmitter from 'typed-emitter';
 
 import type {HistorySnapshot} from '@shared/constants/historySnapshotTypes';
 import type {TransferHistory, TransferSummaryDiff} from '@shared/types/TransferData';
+import type {TorrentListDiff} from '@shared/types/Torrent';
 
 import ServerEvent from '../models/ServerEvent';
 import services from '../services';
@@ -10,13 +11,10 @@ import DiskUsageService from '../services/diskUsageService';
 
 import type {DiskUsage} from '../services/diskUsageService';
 
-export default (req: Request<unknown, unknown, unknown, {historySnapshot: HistorySnapshot}>, res: Response) => {
+export default async (req: Request<unknown, unknown, unknown, {historySnapshot: HistorySnapshot}>, res: Response) => {
   const {
     query: {historySnapshot = 'FIVE_MINUTE'},
     user,
-  }: {
-    query: {historySnapshot: HistorySnapshot};
-    user?: Express.User;
   } = req;
 
   if (user == null) {
@@ -25,9 +23,7 @@ export default (req: Request<unknown, unknown, unknown, {historySnapshot: Histor
 
   const serviceInstances = services.getAllServices(user);
   const serverEvent = new ServerEvent(res);
-  const taxonomy = serviceInstances.taxonomyService.getTaxonomy();
-  const torrentList = serviceInstances.torrentService.getTorrentList();
-  const transferSummary = serviceInstances.historyService.getTransferSummary();
+  const fetchTorrentList = serviceInstances.torrentService.fetchTorrentList();
 
   // Hook into events and stop listening when connection is closed
   const handleEvents = <T extends TypedEmitter<Record<string, unknown>>>(
@@ -46,19 +42,17 @@ export default (req: Request<unknown, unknown, unknown, {historySnapshot: Histor
     isConnected: !serviceInstances.clientGatewayService.hasError,
   });
 
-  const handleDiskUsageChange = (diskUsageChange: DiskUsage) => {
+  // Disk usage change event
+  handleEvents(DiskUsageService, 'DISK_USAGE_CHANGE', (diskUsageChange: DiskUsage) => {
     serverEvent.emit(diskUsageChange.id, 'DISK_USAGE_CHANGE', diskUsageChange.disks);
-  };
+  });
 
-  DiskUsageService.updateDisks()
-    .then(() => {
-      const diskUsage = DiskUsageService.getDiskUsage();
-      serverEvent.emit(diskUsage.id, 'DISK_USAGE_CHANGE', diskUsage.disks);
-      handleEvents(DiskUsageService, 'DISK_USAGE_CHANGE', handleDiskUsageChange);
-    })
-    .catch(() => {
-      // do nothing.
-    });
+  // Trigger an immediate update
+  DiskUsageService.updateDisks().catch((e) => console.error(e));
+
+  const torrentList = (await fetchTorrentList) || serviceInstances.torrentService.getTorrentListSummary();
+  const taxonomy = serviceInstances.taxonomyService.getTaxonomy();
+  const transferSummary = serviceInstances.historyService.getTransferSummary();
 
   serverEvent.emit(torrentList.id, 'TORRENT_LIST_FULL_UPDATE', torrentList.torrents);
   serverEvent.emit(taxonomy.id, 'TAXONOMY_FULL_UPDATE', taxonomy.taxonomy);
@@ -107,11 +101,10 @@ export default (req: Request<unknown, unknown, unknown, {historySnapshot: Histor
 
   // Add diff event listeners.
   handleEvents(
-    serviceInstances.historyService,
-    'TRANSFER_SUMMARY_DIFF_CHANGE',
-    (payload: {id: number; diff: TransferSummaryDiff}) => {
-      const {diff, id} = payload;
-      serverEvent.emit(id, 'TRANSFER_SUMMARY_DIFF_CHANGE', diff);
+    serviceInstances.torrentService,
+    'TORRENT_LIST_DIFF_CHANGE',
+    ({id, diff}: {id: number; diff: TorrentListDiff}) => {
+      serverEvent.emit(id, 'TORRENT_LIST_DIFF_CHANGE', diff);
     },
   );
 
@@ -120,8 +113,11 @@ export default (req: Request<unknown, unknown, unknown, {historySnapshot: Histor
     serverEvent.emit(id, 'TAXONOMY_DIFF_CHANGE', diff);
   });
 
-  handleEvents(serviceInstances.torrentService, 'TORRENT_LIST_DIFF_CHANGE', (payload) => {
-    const {diff, id} = payload;
-    serverEvent.emit(id, 'TORRENT_LIST_DIFF_CHANGE', diff);
-  });
+  handleEvents(
+    serviceInstances.historyService,
+    'TRANSFER_SUMMARY_DIFF_CHANGE',
+    ({id, diff}: {id: number; diff: TransferSummaryDiff}) => {
+      serverEvent.emit(id, 'TRANSFER_SUMMARY_DIFF_CHANGE', diff);
+    },
+  );
 };

server/services/torrentService.ts

@@ -18,12 +18,15 @@ interface TorrentServiceEvents {
   FETCH_TORRENT_LIST_SUCCESS: () => void;
   FETCH_TORRENT_LIST_ERROR: () => void;
   TORRENT_LIST_DIFF_CHANGE: (payload: {id: number; diff: TorrentListDiff}) => void;
+  newListener: (event: keyof Omit<TorrentServiceEvents, 'newListener' | 'removeListener'>) => void;
+  removeListener: (event: keyof Omit<TorrentServiceEvents, 'newListener' | 'removeListener'>) => void;
 }
 
 const torrentListMethodCallConfig = methodCallUtil.getMethodCallConfigFromPropMap(torrentListPropMap);
 
 class TorrentService extends BaseService<TorrentServiceEvents> {
   errorCount = 0;
+  pollEnabled = false;
   pollTimeout: NodeJS.Timeout | null = null;
   torrentListSummary: {
     id: number;
@@ -63,6 +66,21 @@ class TorrentService extends BaseService<TorrentServiceEvents> {
       clientGatewayService.on('PROCESS_TORRENT', this.handleTorrentProcessed);
 
       this.fetchTorrentList();
+
+      // starts polling when the first streaming listener is added
+      this.on('newListener', (event) => {
+        if (!this.pollEnabled && event === 'TORRENT_LIST_DIFF_CHANGE') {
+          this.pollEnabled = true;
+          this.deferFetchTorrentList();
+        }
+      });
+
+      // stops polling when the last streaming listener is removed
+      this.on('removeListener', (event) => {
+        if (event === 'TORRENT_LIST_DIFF_CHANGE' && this.listenerCount('TORRENT_LIST_DIFF_CHANGE') === 0) {
+          this.pollEnabled = false;
+        }
+      });
     };
   }
 
@@ -112,7 +130,9 @@ class TorrentService extends BaseService<TorrentServiceEvents> {
   }
 
   deferFetchTorrentList(interval = config.torrentClientPollInterval || 2000) {
-    this.pollTimeout = setTimeout(this.fetchTorrentList, interval);
+    if (this.pollEnabled) {
+      this.pollTimeout = setTimeout(this.fetchTorrentList, interval);
+    }
   }
 
   destroy() {
@@ -126,7 +146,7 @@ class TorrentService extends BaseService<TorrentServiceEvents> {
       clearTimeout(this.pollTimeout);
     }
 
-    this.services?.clientGatewayService
+    return this.services?.clientGatewayService
       .fetchTorrentList(torrentListMethodCallConfig)
       .then(this.handleFetchTorrentListSuccess)
       .catch(this.handleFetchTorrentListError);
@@ -137,7 +157,7 @@ class TorrentService extends BaseService<TorrentServiceEvents> {
   }
 
   getTorrentList() {
-    return this.torrentListSummary;
+    return this.torrentListSummary.torrents;
   }
 
   getTorrentListDiff(nextTorrentListSummary: this['torrentListSummary']) {
@@ -206,6 +226,7 @@ class TorrentService extends BaseService<TorrentServiceEvents> {
     this.deferFetchTorrentList(nextInterval);
 
     this.emit('FETCH_TORRENT_LIST_ERROR');
+    return null;
   }
 
   handleFetchTorrentListSuccess(nextTorrentListSummary: this['torrentListSummary']) {
@@ -220,6 +241,7 @@ class TorrentService extends BaseService<TorrentServiceEvents> {
 
     this.errorCount = 0;
     this.emit('FETCH_TORRENT_LIST_SUCCESS');
+    return this.torrentListSummary;
   }
 
   handleTorrentProcessed(nextTorrentProperties: TorrentProperties) {