Apply rate limits:

* send only most recent data if it is coming in faster than it can be uploaded/timeout (due to API outage etc) so avoid backlog
* drop really stale data (age > 1 hour)
* drop really fast data (interval < 1 minute)

Author: bertrik

sensor-data-bridge/src/main/java/nl/bertriksikken/pm/SensorData.java

@@ -1,5 +1,6 @@
 package nl.bertriksikken.pm;
 
+import java.time.Instant;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -11,6 +12,15 @@ public final class SensorData {
 
     // start with a simple map containing one Number value per item
     private final Map<ESensorItem, Number> items = new LinkedHashMap<>();
+    private final Instant timestamp;
+
+    public SensorData() {
+        this.timestamp = Instant.now();
+    }
+
+    public Instant getCreationTime() {
+        return timestamp;
+    }
 
     public void putValue(ESensorItem item, Number value) {
         if (value == null) {

sensor-data-bridge/src/main/java/nl/bertriksikken/senscom/SensComWorker.java

@@ -22,6 +22,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -41,6 +42,10 @@ final class SensComWorker {
 
     // map from device id to sensor.community id
     private final Map<AppDeviceId, String> sensComIds = new HashMap<>();
+    // map with only *latest* measurement per AppDeviceId
+    private final Map<AppDeviceId, SensorData> measurements = new ConcurrentHashMap<>();
+    // map from senscom id -> last upload time
+    private final Map<String, Instant> lastUploadTimes = new HashMap<>();
 
     SensComWorker(ObjectMapper mapper, ISensComApi restClient, String softwareVersion, String appId) {
         this.mapper = Objects.requireNonNull(mapper);
@@ -76,18 +81,49 @@ void stop() {
 
     // schedules an upload to all pins
     void scheduleUpload(AppDeviceId appDeviceId, SensorData data) {
-        executor.execute(new CatchingRunnable(LOG, () -> performUpload(appDeviceId, data)));
+        // store measurement in cache of most-recent data and schedule upload
+        SensorData oldData = measurements.put(appDeviceId, data);
+        if (oldData != null) {
+            LOG.warn("Replaced old sensor data for '{}'", appDeviceId);
+        }
+        executor.execute(new CatchingRunnable(LOG, () -> performUpload(appDeviceId)));
     }
 
     // uploads to all pins, runs on our executor
-    private void performUpload(AppDeviceId appDeviceId, SensorData data) {
+    private void performUpload(AppDeviceId appDeviceId) {
+        // check if we still have data to upload, or it was already sent
+        SensorData data = measurements.remove(appDeviceId);
+        if (data == null) {
+            return;
+        }
+
         // look up custom sensor.community id
         String sensorId = sensComIds.getOrDefault(appDeviceId, "");
         if (sensorId.isEmpty()) {
-            // no sensor.community id found, so no upload
             return;
         }
 
+        // drop stale data
+        Duration age = Duration.between(data.getCreationTime(), Instant.now());
+        LOG.info("Sensor data age {} before upload for {}", age, appDeviceId);
+        if (age.toSeconds() > 3600) {
+            LOG.warn("Dropped stale data for {}: {}", appDeviceId, data);
+            return;
+        }
+
+        // rate limit on sensor.community id
+        Instant now = Instant.now();
+        Instant lastUpload = lastUploadTimes.get(sensorId);
+        if (lastUpload != null) {
+            Duration interval = Duration.between(lastUpload, now);
+            LOG.info("Upload interval for {}: {}", appDeviceId, interval);
+            if (interval.toSeconds() < 60) {
+                LOG.warn("Dropped by rate limit: {}", appDeviceId);
+                return;
+            }
+        }
+        lastUploadTimes.put(sensorId, now);
+
         // pin 1 (dust sensors)
         if (data.hasValue(ESensorItem.PM10) || data.hasValue(ESensorItem.PM2_5) || data.hasValue(ESensorItem.PM1_0)
                 || data.hasValue(ESensorItem.PM4_0)) {