Merge pull request #66 from ccutrer/aggregate_notifications

batch notifications when possible

Author: Tim Harper

src/main/java/com/beowulfe/hap/impl/connections/PendingNotification.java

@@ -0,0 +1,15 @@
+package com.beowulfe.hap.impl.connections;
+
+import com.beowulfe.hap.characteristics.EventableCharacteristic;
+
+public class PendingNotification {
+  public int aid;
+  public int iid;
+  public EventableCharacteristic characteristic;
+
+  public PendingNotification(int aid, int iid, EventableCharacteristic characteristic) {
+    this.aid = aid;
+    this.iid = iid;
+    this.characteristic = characteristic;
+  }
+}

src/main/java/com/beowulfe/hap/impl/connections/SubscriptionManager.java

@@ -4,6 +4,7 @@
 import com.beowulfe.hap.impl.http.HomekitClientConnection;
 import com.beowulfe.hap.impl.http.HttpResponse;
 import com.beowulfe.hap.impl.json.EventController;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.Set;
@@ -20,6 +21,9 @@ public class SubscriptionManager {
       new ConcurrentHashMap<>();
   private final ConcurrentMap<HomekitClientConnection, Set<EventableCharacteristic>> reverse =
       new ConcurrentHashMap<>();
+  private final ConcurrentMap<HomekitClientConnection, ArrayList<PendingNotification>>
+      pendingNotifications = new ConcurrentHashMap<>();
+  private int nestedBatches = 0;
 
   public synchronized void addSubscription(
       int aid,
@@ -72,6 +76,7 @@ public synchronized void removeSubscription(
 
   public synchronized void removeConnection(HomekitClientConnection connection) {
     Set<EventableCharacteristic> characteristics = reverse.remove(connection);
+    pendingNotifications.remove(connection);
     if (characteristics != null) {
       for (EventableCharacteristic characteristic : characteristics) {
         Set<HomekitClientConnection> characteristicSubscriptions =
@@ -91,10 +96,42 @@ private <T> Set<T> newSet() {
     return Collections.newSetFromMap(new ConcurrentHashMap<T, Boolean>());
   }
 
-  public void publish(int accessoryId, int iid, EventableCharacteristic changed) {
+  public synchronized void batchUpdate() {
+    ++this.nestedBatches;
+  }
+
+  public synchronized void completeUpdateBatch() {
+    if (--this.nestedBatches == 0 && !pendingNotifications.isEmpty()) {
+      LOGGER.info("Publishing batched changes");
+      for (ConcurrentMap.Entry<HomekitClientConnection, ArrayList<PendingNotification>> entry :
+          pendingNotifications.entrySet()) {
+        try {
+          HttpResponse message = new EventController().getMessage(entry.getValue());
+          entry.getKey().outOfBand(message);
+        } catch (Exception e) {
+          LOGGER.error("Faled to create new event message", e);
+        }
+      }
+      pendingNotifications.clear();
+    }
+  }
+
+  public synchronized void publish(int accessoryId, int iid, EventableCharacteristic changed) {
+    if (nestedBatches != 0) {
+      LOGGER.info("Batching change for " + accessoryId);
+      PendingNotification notification = new PendingNotification(accessoryId, iid, changed);
+      for (HomekitClientConnection connection : subscriptions.get(changed)) {
+        if (!pendingNotifications.containsKey(connection)) {
+          pendingNotifications.put(connection, new ArrayList<PendingNotification>());
+        }
+        pendingNotifications.get(connection).add(notification);
+      }
+      return;
+    }
+
     try {
       HttpResponse message = new EventController().getMessage(accessoryId, iid, changed);
-      LOGGER.info("Publishing changes for " + accessoryId);
+      LOGGER.info("Publishing change for " + accessoryId);
       for (HomekitClientConnection connection : subscriptions.get(changed)) {
         connection.outOfBand(message);
       }

src/main/java/com/beowulfe/hap/impl/json/CharacteristicsController.java

@@ -68,28 +68,34 @@ public HttpResponse get(HttpRequest request) throws Exception {
 
   public HttpResponse put(HttpRequest request, HomekitClientConnection connection)
       throws Exception {
-    try (ByteArrayInputStream bais = new ByteArrayInputStream(request.getBody())) {
-      JsonArray jsonCharacteristics =
-          Json.createReader(bais).readObject().getJsonArray("characteristics");
-      for (JsonValue value : jsonCharacteristics) {
-        JsonObject jsonCharacteristic = (JsonObject) value;
-        int aid = jsonCharacteristic.getInt("aid");
-        int iid = jsonCharacteristic.getInt("iid");
-        Characteristic characteristic = registry.getCharacteristics(aid).get(iid);
+    subscriptions.batchUpdate();
+    try {
+      try (ByteArrayInputStream bais = new ByteArrayInputStream(request.getBody())) {
+        JsonArray jsonCharacteristics =
+            Json.createReader(bais).readObject().getJsonArray("characteristics");
+        for (JsonValue value : jsonCharacteristics) {
+          JsonObject jsonCharacteristic = (JsonObject) value;
+          int aid = jsonCharacteristic.getInt("aid");
+          int iid = jsonCharacteristic.getInt("iid");
+          Characteristic characteristic = registry.getCharacteristics(aid).get(iid);
 
-        if (jsonCharacteristic.containsKey("value")) {
-          characteristic.setValue(jsonCharacteristic.get("value"));
-        }
-        if (jsonCharacteristic.containsKey("ev")
-            && characteristic instanceof EventableCharacteristic) {
-          if (jsonCharacteristic.getBoolean("ev")) {
-            subscriptions.addSubscription(
-                aid, iid, (EventableCharacteristic) characteristic, connection);
-          } else {
-            subscriptions.removeSubscription((EventableCharacteristic) characteristic, connection);
+          if (jsonCharacteristic.containsKey("value")) {
+            characteristic.setValue(jsonCharacteristic.get("value"));
+          }
+          if (jsonCharacteristic.containsKey("ev")
+              && characteristic instanceof EventableCharacteristic) {
+            if (jsonCharacteristic.getBoolean("ev")) {
+              subscriptions.addSubscription(
+                  aid, iid, (EventableCharacteristic) characteristic, connection);
+            } else {
+              subscriptions.removeSubscription(
+                  (EventableCharacteristic) characteristic, connection);
+            }
           }
         }
       }
+    } finally {
+      subscriptions.completeUpdateBatch();
     }
     return new HapJsonNoContentResponse();
   }

src/main/java/com/beowulfe/hap/impl/json/EventController.java

@@ -1,8 +1,10 @@
 package com.beowulfe.hap.impl.json;
 
 import com.beowulfe.hap.characteristics.EventableCharacteristic;
+import com.beowulfe.hap.impl.connections.PendingNotification;
 import com.beowulfe.hap.impl.http.HttpResponse;
 import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
 import javax.json.Json;
 import javax.json.JsonArrayBuilder;
 import javax.json.JsonObject;
@@ -29,4 +31,25 @@ public HttpResponse getMessage(int accessoryId, int iid, EventableCharacteristic
       return new EventResponse(dataBytes);
     }
   }
+
+  public HttpResponse getMessage(ArrayList<PendingNotification> notifications) throws Exception {
+    JsonArrayBuilder characteristics = Json.createArrayBuilder();
+
+    for (PendingNotification notification : notifications) {
+      JsonObjectBuilder characteristicBuilder = Json.createObjectBuilder();
+      characteristicBuilder.add("aid", notification.aid);
+      characteristicBuilder.add("iid", notification.iid);
+      notification.characteristic.supplyValue(characteristicBuilder);
+      characteristics.add(characteristicBuilder.build());
+    }
+
+    JsonObject data = Json.createObjectBuilder().add("characteristics", characteristics).build();
+
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+      Json.createWriter(baos).write(data);
+      byte[] dataBytes = baos.toByteArray();
+
+      return new EventResponse(dataBytes);
+    }
+  }
 }