feat: schedule scan based on queue events

lib/scan.js

@@ -1,6 +1,7 @@
 import { logger } from './logger.js'
 import { options } from './args.js'
 import { queue as parseQueue} from './parse.js'
+import { cargoQueue } from './cargo.js'
 import { serializeError } from 'serialize-error'
 import fg from 'fast-glob'
 import lineByLine from 'n-readlines'
@@ -11,11 +12,37 @@ const component = 'scan'
 const historySet = new Set() // in memory history set
 let isWriteScheduled = false // flag to indicate if there is pending files to write to the history file
 let timeoutId // id of the active setTimeout
+let isParseQueueActive = false
+let isCargoQueueActive = false
+
+function testScheduleNextScan() {
+  if (!isParseQueueActive && !isCargoQueueActive && !options.oneShot) {
+    scheduleNextScan()
+  }
+}
 
 /**
  * Utility function that calls initHistory() and startScanner()
  */
 function initScanner() {
+  parseQueue.on('task_queued', () => {
+    logger.info({ component, message: `parse_queue_task_queued` })
+    isParseQueueActive = true
+  })
+  parseQueue.on('drain', () => {
+    logger.info({ component, message: `parse_queue_drained` })
+    isParseQueueActive = false
+    testScheduleNextScan()
+  })
+  cargoQueue.on('task_queued', () => {
+    logger.info({ component, message: `cargo_queue_task_queued` })
+    isCargoQueueActive = true
+  })
+  cargoQueue.on('drain', () => {
+    logger.info({ component, message: `cargo_queue_drained` })
+    isCargoQueueActive = false
+    testScheduleNextScan()
+  })
   initHistory()
   startScanner()
   Alarm.on('alarmRaised', onAlarmRaised)
@@ -53,18 +80,12 @@ async function startScanner() {
       //Remove stale files: those in historySet but not found in the current scan
       removeStaleFiles(discoveredFiles)
       logger.info({ component, message: `scan ended`, path: options.path })
+      // allow queue event handlers to fire before calling testScheduleNextScan()
+      setTimeout(testScheduleNextScan, 0)
   } 
   catch (e) {
       logger.error({ component, error: serializeError(e) })
   } 
-  finally {
-    if (!options.oneShot) {
-        scheduleNextScan()
-    } 
-    else {
-        logger.info({ component, message: `one-shot scan completed`, path: options.path })
-    }
-  }
 }
 
 /**
@@ -97,19 +118,6 @@ function scheduleNextScan() {
   })
 }
 
-/**
- * Cancels the next scan and logs.
- * References options properties {path}.
- */
-function cancelNextScan() {
-  clearTimeout(timeoutId)
-  logger.info({ 
-    component, 
-    message: `scan cancelled`, 
-    path: options.path
-  })
-}
-
 /**
  * Returns immediately if options.historyFile is falsy.
  * Initializes the history Set by reading it from a file and adding each line to the history set.
@@ -330,7 +338,8 @@ function onAlarmRaised(alarmType) {
     message: `handling raised alarm`,
     alarmType
   })
-  cancelNextScan()
+  parseQueue.pause()
+  cargoQueue.pause()
 }
 
 /**
@@ -344,7 +353,8 @@ function onAlarmLowered(alarmType) {
     message: `handling lowered alarm`,
     alarmType
   })
-  startScanner()
+  parseQueue.resume()
+  cargoQueue.resume()
 }