Implemented leaky bucket algorithm

Author: sivar2311

src/LeakyBucket.h

@@ -0,0 +1,61 @@
+/*
+ *  Copyright (c) 2019 Sinric. All rights reserved.
+ *  Licensed under Creative Commons Attribution-Share Alike (CC BY-SA)
+ *
+ *  This file is part of the Sinric Pro (https://github.com/sinricpro/)
+ */
+
+
+#ifndef _LEAKY_BUCKET_H_
+#define _LEAKY_BUCKET_H_
+
+class LeakyBucket_t {
+  public:
+    LeakyBucket_t() : dropsInBucket(0), lastDrop(-DROP_IN_TIME), once(false) {}
+    bool addDrop();
+  private:
+    void leak();
+    int dropsInBucket;
+    unsigned long lastDrop;
+    bool once;
+    unsigned long lastWarning;
+};
+
+bool LeakyBucket_t::addDrop() {
+  leak();
+  unsigned long actualMillis = millis();
+
+  if (dropsInBucket < BUCKET_SIZE && actualMillis-lastDrop > dropsInBucket + DROP_IN_TIME) { // new drop can be placed into bucket?
+    dropsInBucket++;                                                                          // place drop in bucket
+    lastDrop = actualMillis;                                                                  // store last drop time
+    if (dropsInBucket == BUCKET_SIZE && once==false) {
+      Serial.printf("[SinricPro]: WARNING: YOU SENT TOO MUCH EVENTS IN A SHORT PERIOD OF TIME!\r\n  - PLEASE CHECK YOUR CODE AND SEND EVENTS ONLY IF DEVICE STATE HAS CHANGED!\r\n");      // Print a warning when bucket is full
+      once = true;
+    }
+    return true;
+  }
+
+  if (dropsInBucket >= BUCKET_SIZE) {
+    if (actualMillis-lastWarning > 1000) {
+      Serial.printf("[SinricPro]: EVENTS ARE BLOCKED FOR %lu SECONDS!\r\n",(DROP_OUT_TIME-(actualMillis-lastDrop))/1000);
+      lastWarning = actualMillis;
+    }
+  }
+  return false;
+}
+
+void LeakyBucket_t::leak() {
+// leack bucket...
+  unsigned long actualMillis = millis();
+  int drops_to_leak = (actualMillis - lastDrop) / DROP_OUT_TIME;
+  if (drops_to_leak > 0) {
+    if (dropsInBucket <= drops_to_leak) {
+      dropsInBucket = 0;
+    } else {
+      dropsInBucket -= drops_to_leak;
+    }
+  }
+}
+
+
+#endif

src/SinricProConfig.h

@@ -7,20 +7,37 @@
 
 #ifndef __SINRICPRO_CONFIG_H__
 #define __SINRICPRO_CONFIG_H__
+/* 
+ * !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+ * !!                                                 !!
+ * !!             WARNING: DON'T TOUCH !              !!
+ * !!             ======================              !!
+ * !! PLEASE DO NOT MODIFY ANY OF THESE SETTINGS HERE !!
+ * !!     THIS IS FOR INTERNAL CONFIGURATION ONLY     !!
+ * !!   SINRIC PRO MIGHT NOT WORK IF YOU MODIFY THIS  !!
+ * !!                                                 !!
+ * !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+ */
 
+// Version Configuration
 #define SDK_VERSION "2.2.3"
 
+// Server Configuration
 #define SINRICPRO_SERVER_URL "ws.sinric.pro"
 #define SINRICPRO_SERVER_PORT 80
 
+// UDP Configuration
 #define UDP_MULTICAST_IP IPAddress(224,9,9,9)
 #define UDP_MULTICAST_PORT 3333
 
-// websocket sends every WEBSOCKET_PING_INTERVAL milliseconds a ping to Server
-// if there is no pong received after WEBSOCKET_PING_TIMEOUT milliseconds, retry count is incremented by one
-// if retry count reaches WEBSOCKET_RETRY_COUNT websocket connection is closed and try to reconnect server
+// WebSocket Configuration
 #define WEBSOCKET_PING_INTERVAL 300000
 #define WEBSOCKET_PING_TIMEOUT 10000
 #define WEBSOCKET_RETRY_COUNT 2
 
+// LeakyBucket Configuration
+#define BUCKET_SIZE 10
+#define DROP_OUT_TIME 60000
+#define DROP_IN_TIME 1000u
+
 #endif

src/SinricProDevice.h

@@ -9,16 +9,9 @@
 #define _SINRICDEVICE_H_
 
 #include "SinricProDeviceInterface.h"
-#include <map>
-
-#define BUCKET_SIZE 20
-#define DROP_WAIT_TIME 30000
-#define DROP_ADD_FACTOR 100
+#include "LeakyBucket.h"
 
-struct leackyBucket_t {
-  int dropsInBucket=0;
-  unsigned long lastDrop;
-};
+#include <map>
 
 class SinricProDevice : public SinricProDeviceInterface {
   public:
@@ -46,7 +39,7 @@ class SinricProDevice : public SinricProDeviceInterface {
   private:
     SinricProInterface* eventSender;
     unsigned long eventWaitTime;
-    std::map<String, leackyBucket_t> eventFilter;
+    std::map<String, LeakyBucket_t> eventFilter;
 };
 
 SinricProDevice::SinricProDevice(const char* newDeviceId, unsigned long eventWaitTime) : 
@@ -90,52 +83,30 @@ DynamicJsonDocument SinricProDevice::prepareEvent(const char* deviceId, const ch
   return DynamicJsonDocument(1024);
 }
 
+
 bool SinricProDevice::sendEvent(JsonDocument& event) {
-  unsigned long actualMillis = millis();
   String eventName = event["payload"]["action"] | ""; // get event name
 
-  leackyBucket_t bucket;
+  LeakyBucket_t bucket; // leaky bucket algorithm is used to prevent flooding the server
 
-// get Bucket for eventName
-
-  if (eventFilter.find(eventName) == eventFilter.end()) {  // if eventFilter is not initialized
-//    eventFilter[eventName] = -eventWaitTime; // initialize eventFilter
-    bucket.dropsInBucket = 0;
-    bucket.lastDrop = -eventWaitTime;
-    eventFilter[eventName] = bucket;
-  } else {  // if eventFilter is initialized, get bucket
-    bucket = eventFilter[eventName]; 
-  }
-  DEBUG_SINRIC("Bucket dropsInBucket: %d\r\n", bucket.dropsInBucket);
-// leack bucket...
-  int drops_to_leak = (actualMillis - bucket.lastDrop) / DROP_WAIT_TIME;
-  DEBUG_SINRIC("Bucket leaking: %d\r\n", drops_to_leak);
-  if (drops_to_leak > 0) {
-    if (bucket.dropsInBucket <= drops_to_leak) {
-      bucket.dropsInBucket = 0;
-    } else {
-      bucket.dropsInBucket -= drops_to_leak;
-    }
+  // get leaky bucket for event from eventFilter
+  if (eventFilter.find(eventName) == eventFilter.end()) {  // if there is no bucket ...
+    eventFilter[eventName] = bucket;                       // ...add a new bucket
+  } else {  
+    bucket = eventFilter[eventName];                       // else get bucket
   }
 
-//  unsigned long lastEventMillis = eventFilter[eventName] | 0; // get the last timestamp for event
-//  if (actualMillis - lastEventMillis < eventWaitTime) return false; // if last event was before waitTime return...
-//  if (actualMillis - bucket.lastDrop < eventWaitTime) return false; 
-  if (bucket.dropsInBucket < BUCKET_SIZE && actualMillis-bucket.lastDrop > bucket.dropsInBucket * DROP_ADD_FACTOR) { // new drop can be placed into bucket?
-    Serial.printf("SinricProDevice::sendMessage(): %d event's left before limiting to 1 event per %d seconds. %lu ms until next event\r\n", BUCKET_SIZE-bucket.dropsInBucket-1, DROP_WAIT_TIME / 1000, ((bucket.dropsInBucket+1) * DROP_ADD_FACTOR));
-    bucket.dropsInBucket++; // place drop in bucket
-    bucket.lastDrop = actualMillis; // store last drop time
-    eventFilter[eventName] = bucket; // save bucket back to map
-    if (eventSender) eventSender->sendMessage(event); // send event
+  if (bucket.addDrop()) {                                  // if we can add a new drop
+    if (eventSender) eventSender->sendMessage(event);      // send event
+    eventFilter[eventName] = bucket;                       // update bucket on eventFilter
     return true;
   }
 
-  if (bucket.dropsInBucket >= BUCKET_SIZE) {
-    Serial.printf("- WARNING: EVENTS ARE BLOCKED FOR %lu SECONDS (%lu seconds left)\r", DROP_WAIT_TIME/1000, (DROP_WAIT_TIME-(actualMillis-bucket.lastDrop))/1000);
-  }
+  eventFilter[eventName] = bucket;                        // update bucket on eventFilter
   return false;
 }
 
+
 bool SinricProDevice::sendPowerStateEvent(bool state, String cause) {
   DynamicJsonDocument eventMessage = prepareEvent(deviceId, "setPowerState", cause.c_str());
   JsonObject event_value = eventMessage["payload"]["value"];