feature: add write statistics to dashboard

packages/server-admin-ui/src/views/Dashboard/Dashboard.js

@@ -74,18 +74,26 @@ const Dashboard = (props) => {
                 </Col>
                 <Col xs="12" md="6">
                   <div className="text-muted">
-                    Connection activity (deltas/second)
+                    Connection activity (input: deltas/second - output:
+                    messages/second)
                   </div>
                   <ul className="horizontal-bars type-2">
                     {Object.keys(providerStatistics || {}).map((providerId) => {
                       const providerStats = providerStatistics[providerId]
-                      const iconClass =
-                        'icon-feed text-primary' +
+                      const inputPulseIconClass =
+                        'icon-login text-primary' +
                         (providerStats.deltaRate > 50
                           ? ' fa-pulse-fast'
                           : providerStats.deltaRate > 0
                           ? ' fa-pulse'
                           : '')
+                      const outputPulseIconClass =
+                        'icon-logout text-primary' +
+                        (providerStats.writeRate > 50
+                          ? ' fa-pulse-fast'
+                          : providerStats.writeRate > 0
+                          ? ' fa-pulse'
+                          : '')
                       return (
                         <li
                           key={providerId}
@@ -95,7 +103,16 @@ const Dashboard = (props) => {
                             )
                           }
                         >
-                          <i className={iconClass} />
+                          <i className={inputPulseIconClass} />
+                          <i
+                            className={outputPulseIconClass}
+                            style={{
+                              transform: 'scaleX(-1)',
+                              visibility: providerStats.lastIntervalWriteCount
+                                ? ''
+                                : 'hidden',
+                            }}
+                          />
                           <span className="title">
                             {providerIdLink(providerId)}
                           </span>

packages/streams/serialport.js

@@ -166,6 +166,11 @@ SerialStream.prototype.start = function () {
         } else {
           that.serial.write(d + '\r\n')
         }
+        setImmediate(() => {
+          that.options.app.emit('connectionwrite', {
+            providerId: that.options.providerId,
+          })
+        })
         pendingWrites++
         that.serial.drain(onDrain)
       })

packages/streams/simple.js

@@ -76,6 +76,16 @@ function Simple(options) {
     ) {
       mappingType = 'NMEA2000YD'
     }
+    options.app.on('nmea2000out', () => {
+      setImmediate(() =>
+        options.app.emit('connectionwrite', { providerId: options.providerId })
+      )
+    })
+    options.app.on('nmea2000JsonOut', () => {
+      setImmediate(() =>
+        options.app.emit('connectionwrite', { providerId: options.providerId })
+      )
+    })
   }
 
   const pipeline = [].concat(

packages/streams/tcp.js

@@ -53,6 +53,11 @@ TcpStream.prototype.pipe = function (pipeTo) {
       if (that.tcpStream) {
         that.debug('sending %s', d)
         that.tcpStream.write(d)
+        setImmediate(() => {
+          that.options.app.emit('connectionwrite', {
+            providerId: that.options.providerId,
+          })
+        })
       }
     })
   }

src/app.ts

@@ -1,6 +1,8 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import { SKVersion } from '@signalk/server-api'
 import { FullSignalK } from '@signalk/signalk-schema'
+import { EventEmitter } from 'node:events'
+
 import { Config } from './config/config'
 import DeltaCache from './deltacache'
 
@@ -17,9 +19,7 @@ export interface ServerApp {
   clients: number
 }
 
-export interface SignalKMessageHub {
-  emit: any
-  on: any
+export interface SignalKMessageHub extends EventEmitter {
   signalk: FullSignalK
   handleMessage: (id: string, Delta: any, skVersion?: SKVersion) => void
 }

src/deltastats.ts

@@ -16,37 +16,83 @@
 */
 
 import { isUndefined, values } from 'lodash'
+import { EventEmitter } from 'node:events'
 
-export function startDeltaStatistics(app: any) {
+const STATS_UPDATE_INTERVAL_SECONDS = 5
+
+interface ConnectionWriteEvent {
+  providerId: string
+}
+
+class ProviderStats {
+  writeRate: number
+  writeCount: number
+  lastIntervalWriteCount: number
+  deltaRate: number
+  deltaCount: number
+  lastIntervalDeltaCount: number
+  constructor() {
+    this.writeRate =
+      this.writeCount =
+      this.lastIntervalWriteCount =
+      this.deltaRate =
+      this.deltaCount =
+      this.lastIntervalDeltaCount =
+        0
+  }
+}
+
+export interface WithProviderStatistics {
+  deltaCount: number
+  lastIntervalDeltaCount: number
+  providerStatistics: {
+    [providerId: string]: ProviderStats
+  }
+}
+
+export function startDeltaStatistics(
+  app: EventEmitter & WithProviderStatistics
+) {
   app.deltaCount = 0
   app.lastIntervalDeltaCount = 0
   app.providerStatistics = {}
 
+  app.on('connectionwrite', (msg: ConnectionWriteEvent) => {
+    const stats =
+      app.providerStatistics[msg.providerId] ||
+      (app.providerStatistics[msg.providerId] = new ProviderStats())
+    stats.writeCount++
+  })
+
   return setInterval(() => {
     updateProviderPeriodStats(app)
+    const anyApp = app as any
     app.emit('serverevent', {
       type: 'SERVERSTATISTICS',
       from: 'signalk-server',
       data: {
-        deltaRate: (app.deltaCount - app.lastIntervalDeltaCount) / 5,
-        numberOfAvailablePaths: app.streambundle.getAvailablePaths().length,
-        wsClients: app.interfaces.ws ? app.interfaces.ws.numClients() : 0,
+        deltaRate:
+          (app.deltaCount - app.lastIntervalDeltaCount) /
+          STATS_UPDATE_INTERVAL_SECONDS,
+        numberOfAvailablePaths: anyApp.streambundle.getAvailablePaths().length,
+        wsClients: anyApp.interfaces.ws ? anyApp.interfaces.ws.numClients() : 0,
         providerStatistics: app.providerStatistics,
         uptime: process.uptime()
       }
     })
     app.lastIntervalDeltaCount = app.deltaCount
-  }, 5 * 1000)
+  }, STATS_UPDATE_INTERVAL_SECONDS * 1000)
 }
 
-export function incDeltaStatistics(app: any, providerId: any) {
+export function incDeltaStatistics(
+  app: WithProviderStatistics,
+  providerId: any
+) {
   app.deltaCount++
 
   const stats =
     app.providerStatistics[providerId] ||
-    (app.providerStatistics[providerId] = {
-      deltaCount: 0
-    })
+    (app.providerStatistics[providerId] = new ProviderStats())
   stats.deltaCount++
 }
 
@@ -60,8 +106,14 @@ function updateProviderPeriodStats(app: any) {
     }
   })
 
-  values(app.providerStatistics).forEach((stats: any) => {
-    stats.deltaRate = (stats.deltaCount - stats.lastIntervalDeltaCount) / 5
+  values(app.providerStatistics).forEach((stats: ProviderStats) => {
+    stats.deltaRate =
+      (stats.deltaCount - stats.lastIntervalDeltaCount) /
+      STATS_UPDATE_INTERVAL_SECONDS
     stats.lastIntervalDeltaCount = stats.deltaCount
+    stats.writeRate =
+      (stats.writeCount - stats.lastIntervalWriteCount) /
+      STATS_UPDATE_INTERVAL_SECONDS
+    stats.lastIntervalWriteCount = stats.writeCount
   })
 }

src/index.ts

@@ -59,6 +59,7 @@ import { setupCors } from './cors'
 import SubscriptionManager from './subscriptionmanager'
 import { PluginId, PluginManager } from './interfaces/plugins'
 import { OpenApiDescription, OpenApiRecord } from './api/swagger'
+import { WithProviderStatistics } from './deltastats'
 const debug = createDebug('signalk-server')
 
 const { StreamBundle } = require('./streambundle')
@@ -74,7 +75,8 @@ class Server {
     SignalKMessageHub &
     PluginManager &
     WithSecurityStrategy &
-    IRouter
+    IRouter &
+    WithProviderStatistics
   constructor(opts: ServerOptions) {
     const FILEUPLOADSIZELIMIT = process.env.FILEUPLOADSIZELIMIT || '10mb'
     const bodyParser = require('body-parser')
@@ -329,6 +331,7 @@ class Server {
         }
       }
       expressAppEmit(eventName, ...args)
+      return true
     }
 
     this.app.intervals = []