fix(connection): Block device state updates to Alexa when too many incoming msgs are received

Author: csuermann

MqttClient.js

@@ -7,7 +7,7 @@ const globalLmiter = new Bottleneck({
   reservoirIncreaseAmount: 2,
   reservoirIncreaseInterval: 10000, // must be divisible by 250
   reservoirIncreaseMaximum: 40,
-  strategy: Bottleneck.strategy.LEAK,
+  strategy: Bottleneck.strategy.BLOCK,
 
   // also use maxConcurrent and/or minTime for safety
   maxConcurrent: 5,
@@ -32,7 +32,7 @@ function MqttClient (options, callbacksObj) {
     reservoirIncreaseAmount: 2,
     reservoirIncreaseInterval: 10000, // must be divisible by 250
     reservoirIncreaseMaximum: 40,
-    strategy: Bottleneck.strategy.LEAK,
+    strategy: Bottleneck.strategy.BLOCK,
 
     // also use maxConcurrent and/or minTime for safety
     maxConcurrent: 5,

RateLimiter.js

@@ -0,0 +1,35 @@
+'use strict'
+
+function RateLimiter ({
+  highWaterMark = 10,
+  intervalInSec = 5,
+  onExhaustionCb
+} = {}) {
+  this.highWaterMark = highWaterMark
+  this.intervalInSec = intervalInSec * 1000
+  this.exhaustedCallback = onExhaustionCb
+  this.counter = 0
+
+  this.resetCounter = function () {
+    this.counter = 0
+  }
+
+  this.intervalHandle = setInterval(
+    this.resetCounter.bind(this),
+    this.intervalInSec
+  )
+}
+
+RateLimiter.prototype.execute = function (callback) {
+  this.counter++
+
+  if (this.counter <= this.highWaterMark) {
+    callback()
+  }
+
+  if (this.counter == this.highWaterMark) {
+    this.exhaustedCallback()
+  }
+}
+
+module.exports = RateLimiter

virtual-device.js

@@ -1,5 +1,6 @@
 const merge = require('deepmerge')
 const deepEql = require('deep-eql')
+const RateLimiter = require('./RateLimiter')
 const {
   getValidators,
   getDecorator,
@@ -23,6 +24,19 @@ module.exports = function (RED) {
     const validators = getValidators(config.template)
     const decorator = getDecorator(config.template)
 
+    let isIncomingMsgProcessingAllowed = true
+
+    const rater = new RateLimiter({
+      highWaterMark: 15,
+      intervalInSec: 60,
+      onExhaustionCb: () => {
+        isIncomingMsgProcessingAllowed = false
+        console.log(
+          'Blocking device state sync to Alexa from now on! Quota exhausted!'
+        )
+      }
+    })
+
     const getLocalState = function () {
       return { ...localState }
     }
@@ -77,10 +91,15 @@ module.exports = function (RED) {
       const mergedState = merge(oldLocalState, approvedState)
       const newLocalState = { ...mergedState, source: 'device' }
 
-      if (!deepEql(oldLocalState, newLocalState)) {
+      if (
+        isIncomingMsgProcessingAllowed &&
+        !deepEql(oldLocalState, newLocalState)
+      ) {
         setLocalState(newLocalState)
 
-        connectionNode.updateShadow({ nodeId, type: 'desired' })
+        rater.execute(() =>
+          connectionNode.updateShadow({ nodeId, type: 'desired' })
+        )
       }
 
       if (config.passthrough && Object.keys(approvedState).length > 0) {