From 24b5be474d9b7c16220712d6a1d4280245858e35 Mon Sep 17 00:00:00 2001
From: Teppo Kurki <teppo.kurki@iki.fi>
Date: Tue, 29 Aug 2023 21:44:19 +0300
Subject: [PATCH] feature: add write statistics to dashboard

Add new icon to indicate outbound traffic on Connections.

Change the previous feed icon to login to indicate incoming
traffic so that we can use the 'logout' icon for outbound
traffic.
---
 .../src/views/Dashboard/Dashboard.js          | 25 ++++++-
 packages/streams/serialport.js                |  5 ++
 packages/streams/simple.js                    | 10 +++
 packages/streams/tcp.js                       |  5 ++
 src/app.ts                                    |  6 +-
 src/deltastats.ts                             | 74 ++++++++++++++++---
 src/index.ts                                  |  5 +-
 7 files changed, 111 insertions(+), 19 deletions(-)

diff --git a/packages/server-admin-ui/src/views/Dashboard/Dashboard.js b/packages/server-admin-ui/src/views/Dashboard/Dashboard.js
index 50c674be7..c75a4eb39 100644
--- a/packages/server-admin-ui/src/views/Dashboard/Dashboard.js
+++ b/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>
diff --git a/packages/streams/serialport.js b/packages/streams/serialport.js
index deeb47268..c23ff75cd 100644
--- a/packages/streams/serialport.js
+++ b/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)
       })
diff --git a/packages/streams/simple.js b/packages/streams/simple.js
index 54f290ac5..f59ab16e5 100644
--- a/packages/streams/simple.js
+++ b/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(
diff --git a/packages/streams/tcp.js b/packages/streams/tcp.js
index ef447e9ad..92ecfba65 100644
--- a/packages/streams/tcp.js
+++ b/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,
+          })
+        })
       }
     })
   }
diff --git a/src/app.ts b/src/app.ts
index 59450e390..beed4a732 100644
--- a/src/app.ts
+++ b/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
 }
diff --git a/src/deltastats.ts b/src/deltastats.ts
index ada0afb2c..1933380f4 100644
--- a/src/deltastats.ts
+++ b/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
   })
 }
diff --git a/src/index.ts b/src/index.ts
index 186037cbe..b7f3f575e 100644
--- a/src/index.ts
+++ b/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 = []