Skip to content

Commit 3d271dc

Browse files
yfreyfre
authoredNov 23, 2022
Merge pull request #172 from ccutrer/maintain-subscriptions
·
hap-2.0.7hap-2.0.4
2 parents de16264 + a781e85 commit 3d271dc

File tree

6 files changed

+216
-79
lines changed

6 files changed

+216
-79
lines changed
 

‎src/main/java/io/github/hapjava/characteristics/Characteristic.java

+2
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
* @author Andy Lintner
1919
*/
2020
public interface Characteristic {
21+
/** @return The UUID type for this characteristic. */
22+
String getType();
2123

2224
/**
2325
* Adds an attribute to the passed JsonObjectBuilder named "value" with the current value of the

‎src/main/java/io/github/hapjava/characteristics/impl/base/BaseCharacteristic.java

+6
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,12 @@ public BaseCharacteristic(
7070
this.unsubscriber = unsubscriber;
7171
}
7272

73+
@Override
74+
/** {@inheritDoc} */
75+
public String getType() {
76+
return type;
77+
}
78+
7379
@Override
7480
/** {@inheritDoc} */
7581
public final CompletableFuture<JsonObject> toJson(int iid) {

‎src/main/java/io/github/hapjava/server/impl/HomekitRegistry.java

+7-3
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import io.github.hapjava.accessories.HomekitAccessory;
44
import io.github.hapjava.characteristics.Characteristic;
5+
import io.github.hapjava.server.impl.connections.SubscriptionManager;
56
import io.github.hapjava.services.Service;
67
import io.github.hapjava.services.impl.AccessoryInformationService;
78
import java.util.ArrayList;
@@ -19,14 +20,16 @@ public class HomekitRegistry {
1920
private static final Logger logger = LoggerFactory.getLogger(HomekitRegistry.class);
2021

2122
private final String label;
23+
private final SubscriptionManager subscriptions;
2224
private final Map<Integer, HomekitAccessory> accessories;
2325
private final Map<HomekitAccessory, Map<Integer, Service>> services = new HashMap<>();
2426
private final Map<HomekitAccessory, Map<Integer, Characteristic>> characteristics =
2527
new HashMap<>();
2628
private boolean isAllowUnauthenticatedRequests = false;
2729

28-
public HomekitRegistry(String label) {
30+
public HomekitRegistry(String label, SubscriptionManager subscriptions) {
2931
this.label = label;
32+
this.subscriptions = subscriptions;
3033
this.accessories = new ConcurrentHashMap<>();
3134
reset();
3235
}
@@ -61,6 +64,7 @@ public synchronized void reset() {
6164
services.put(accessory, newServicesByInterfaceId);
6265
characteristics.put(accessory, newCharacteristicsByInterfaceId);
6366
}
67+
subscriptions.resync(this);
6468
}
6569

6670
public String getLabel() {
@@ -87,8 +91,8 @@ public void add(HomekitAccessory accessory) {
8791
accessories.put(accessory.getId(), accessory);
8892
}
8993

90-
public void remove(HomekitAccessory accessory) {
91-
accessories.remove(accessory.getId());
94+
public boolean remove(HomekitAccessory accessory) {
95+
return accessories.remove(accessory.getId()) != null;
9296
}
9397

9498
public boolean isAllowUnauthenticatedRequests() {

‎src/main/java/io/github/hapjava/server/impl/HomekitRoot.java

+37-15
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ public class HomekitRoot {
3636
private final SubscriptionManager subscriptions = new SubscriptionManager();
3737
private boolean started = false;
3838
private int configurationIndex = 1;
39+
private int nestedBatches = 0;
40+
private boolean madeChanges = false;
3941

4042
HomekitRoot(
4143
String label, HomekitWebHandler webHandler, InetAddress host, HomekitAuthInfo authInfo)
@@ -65,7 +67,7 @@ public class HomekitRoot {
6567
this.authInfo = authInfo;
6668
this.label = label;
6769
this.category = category;
68-
this.registry = new HomekitRegistry(label);
70+
this.registry = new HomekitRegistry(label, subscriptions);
6971
}
7072

7173
HomekitRoot(
@@ -83,11 +85,27 @@ public class HomekitRoot {
8385
this(
8486
label, DEFAULT_ACCESSORY_CATEGORY, webHandler, authInfo, new JmdnsHomekitAdvertiser(jmdns));
8587
}
88+
8689
/**
87-
* Add an accessory to be handled and advertised by this root. Any existing HomeKit connections
88-
* will be terminated to allow the clients to reconnect and see the updated accessory list. When
89-
* using this for a bridge, the ID of the accessory must be greater than 1, as that ID is reserved
90-
* for the Bridge itself.
90+
* Begin a batch update of accessories.
91+
*
92+
* <p>After calling this, you can call addAccessory() and removeAccessory() multiple times without
93+
* causing HAP-Java to re-publishing the metadata to HomeKit. You'll need to call
94+
* completeUpdateBatch in order to publish all accumulated changes.
95+
*/
96+
public synchronized void batchUpdate() {
97+
if (this.nestedBatches == 0) madeChanges = false;
98+
++this.nestedBatches;
99+
}
100+
101+
/** Publish accumulated accessory changes since batchUpdate() was called. */
102+
public synchronized void completeUpdateBatch() {
103+
if (--this.nestedBatches == 0 && madeChanges) registry.reset();
104+
}
105+
106+
/**
107+
* Add an accessory to be handled and advertised by this root. When using this for a bridge, the
108+
* ID of the accessory must be greater than 1, as that ID is reserved for the Bridge itself.
91109
*
92110
* @param accessory to advertise and handle.
93111
*/
@@ -110,25 +128,28 @@ void addAccessorySkipRangeCheck(HomekitAccessory accessory) {
110128
if (logger.isTraceEnabled()) {
111129
accessory.getName().thenAccept(name -> logger.trace("Added accessory {}", name));
112130
}
113-
if (started) {
131+
madeChanges = true;
132+
if (started && nestedBatches == 0) {
114133
registry.reset();
115134
}
116135
}
117136

118137
/**
119-
* Removes an accessory from being handled or advertised by this root. Any existing HomeKit
120-
* connections will be terminated to allow the clients to reconnect and see the updated accessory
121-
* list.
138+
* Removes an accessory from being handled or advertised by this root.
122139
*
123140
* @param accessory accessory to cease advertising and handling
124141
*/
125142
public void removeAccessory(HomekitAccessory accessory) {
126-
this.registry.remove(accessory);
127-
if (logger.isTraceEnabled()) {
128-
accessory.getName().thenAccept(name -> logger.trace("Removed accessory {}", name));
129-
}
130-
if (started) {
131-
registry.reset();
143+
if (this.registry.remove(accessory)) {
144+
if (logger.isTraceEnabled()) {
145+
accessory.getName().thenAccept(name -> logger.trace("Removed accessory {}", name));
146+
}
147+
madeChanges = true;
148+
if (started && nestedBatches == 0) {
149+
registry.reset();
150+
}
151+
} else {
152+
accessory.getName().thenAccept(name -> logger.warn("Could not remove accessory {}", name));
132153
}
133154
}
134155

@@ -140,6 +161,7 @@ public void removeAccessory(HomekitAccessory accessory) {
140161
*/
141162
public void start() {
142163
started = true;
164+
madeChanges = false;
143165
registry.reset();
144166
webHandler
145167
.start(

‎src/main/java/io/github/hapjava/server/impl/connections/SubscriptionManager.java

+162-59
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,41 @@
11
package io.github.hapjava.server.impl.connections;
22

3+
import io.github.hapjava.characteristics.Characteristic;
34
import io.github.hapjava.characteristics.EventableCharacteristic;
5+
import io.github.hapjava.server.impl.HomekitRegistry;
46
import io.github.hapjava.server.impl.http.HomekitClientConnection;
57
import io.github.hapjava.server.impl.http.HttpResponse;
68
import io.github.hapjava.server.impl.json.EventController;
79
import java.util.ArrayList;
8-
import java.util.Collections;
10+
import java.util.HashMap;
11+
import java.util.HashSet;
912
import java.util.Iterator;
13+
import java.util.List;
14+
import java.util.Map;
1015
import java.util.Set;
11-
import java.util.concurrent.ConcurrentHashMap;
12-
import java.util.concurrent.ConcurrentMap;
1316
import org.slf4j.Logger;
1417
import org.slf4j.LoggerFactory;
1518

1619
public class SubscriptionManager {
1720

1821
private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionManager.class);
1922

20-
private final ConcurrentMap<EventableCharacteristic, Set<HomekitClientConnection>> subscriptions =
21-
new ConcurrentHashMap<>();
22-
private final ConcurrentMap<HomekitClientConnection, Set<EventableCharacteristic>> reverse =
23-
new ConcurrentHashMap<>();
24-
private final ConcurrentMap<HomekitClientConnection, ArrayList<PendingNotification>>
25-
pendingNotifications = new ConcurrentHashMap<>();
23+
private static class ConnectionsWithIds {
24+
Set<HomekitClientConnection> connections;
25+
int aid, iid;
26+
27+
ConnectionsWithIds(int aid, int iid) {
28+
this.aid = aid;
29+
this.iid = iid;
30+
this.connections = new HashSet<>();
31+
}
32+
}
33+
34+
private final Map<EventableCharacteristic, ConnectionsWithIds> subscriptions = new HashMap<>();
35+
private final Map<HomekitClientConnection, Set<EventableCharacteristic>> reverse =
36+
new HashMap<>();
37+
private final Map<HomekitClientConnection, List<PendingNotification>> pendingNotifications =
38+
new HashMap<>();
2639
private int nestedBatches = 0;
2740

2841
public synchronized void addSubscription(
@@ -31,96 +44,111 @@ public synchronized void addSubscription(
3144
EventableCharacteristic characteristic,
3245
HomekitClientConnection connection) {
3346
synchronized (this) {
34-
if (!subscriptions.containsKey(characteristic)) {
35-
subscriptions.putIfAbsent(characteristic, newSet());
36-
}
37-
subscriptions.get(characteristic).add(connection);
38-
if (subscriptions.get(characteristic).size() == 1) {
39-
characteristic.subscribe(
40-
() -> {
41-
publish(aid, iid, characteristic);
42-
});
47+
ConnectionsWithIds subscribers;
48+
if (subscriptions.containsKey(characteristic)) {
49+
subscribers = subscriptions.get(characteristic);
50+
} else {
51+
subscribers = new ConnectionsWithIds(aid, iid);
52+
subscriptions.put(characteristic, subscribers);
53+
subscribe(aid, iid, characteristic);
4354
}
55+
subscribers.connections.add(connection);
4456

4557
if (!reverse.containsKey(connection)) {
46-
reverse.putIfAbsent(connection, newSet());
58+
reverse.put(connection, new HashSet<>());
4759
}
4860
reverse.get(connection).add(characteristic);
4961
LOGGER.trace(
50-
"Added subscription to " + characteristic.getClass() + " for " + connection.hashCode());
62+
"Added subscription to {}:{} ({}) for {}",
63+
aid,
64+
iid,
65+
characteristic.getClass().getSimpleName(),
66+
connection.hashCode());
5167
}
5268
}
5369

5470
public synchronized void removeSubscription(
5571
EventableCharacteristic characteristic, HomekitClientConnection connection) {
56-
Set<HomekitClientConnection> subscriptions = this.subscriptions.get(characteristic);
57-
if (subscriptions != null) {
58-
subscriptions.remove(connection);
59-
if (subscriptions.size() == 0) {
72+
ConnectionsWithIds subscribers = subscriptions.get(characteristic);
73+
if (subscribers != null) {
74+
subscribers.connections.remove(connection);
75+
if (subscribers.connections.isEmpty()) {
76+
LOGGER.trace("Unsubscribing from characteristic as all subscriptions are closed");
6077
characteristic.unsubscribe();
78+
subscriptions.remove(characteristic);
79+
}
80+
81+
// Remove pending notifications for this no-longer-subscribed characteristic
82+
List<PendingNotification> connectionNotifications = pendingNotifications.get(connection);
83+
if (connectionNotifications != null) {
84+
connectionNotifications.removeIf(n -> n.aid == subscribers.aid && n.iid == subscribers.iid);
85+
if (connectionNotifications.isEmpty()) pendingNotifications.remove(connection);
6186
}
87+
88+
LOGGER.trace(
89+
"Removed subscription from {}:{} ({}) for {}",
90+
subscribers.aid,
91+
subscribers.iid,
92+
characteristic.getClass().getSimpleName(),
93+
connection.hashCode());
6294
}
6395

6496
Set<EventableCharacteristic> reverse = this.reverse.get(connection);
6597
if (reverse != null) {
6698
reverse.remove(characteristic);
99+
if (reverse.isEmpty()) this.reverse.remove(connection);
67100
}
68-
LOGGER.trace(
69-
"Removed subscription to " + characteristic.getClass() + " for " + connection.hashCode());
70101
}
71102

72103
public synchronized void removeConnection(HomekitClientConnection connection) {
73-
Set<EventableCharacteristic> characteristics = reverse.remove(connection);
104+
removeConnection(connection, reverse.remove(connection));
105+
}
106+
107+
private void removeConnection(
108+
HomekitClientConnection connection, Set<EventableCharacteristic> characteristics) {
74109
pendingNotifications.remove(connection);
75110
if (characteristics != null) {
76111
for (EventableCharacteristic characteristic : characteristics) {
77-
Set<HomekitClientConnection> characteristicSubscriptions =
78-
subscriptions.get(characteristic);
79-
characteristicSubscriptions.remove(connection);
80-
if (characteristicSubscriptions.isEmpty()) {
81-
LOGGER.trace("Unsubscribing from characteristic as all subscriptions are closed");
82-
characteristic.unsubscribe();
83-
subscriptions.remove(characteristic);
84-
}
112+
removeSubscription(characteristic, connection);
85113
}
86114
}
87115
LOGGER.trace("Removed connection {}", connection.hashCode());
88116
}
89117

90-
private <T> Set<T> newSet() {
91-
return Collections.newSetFromMap(new ConcurrentHashMap<T, Boolean>());
92-
}
93-
94118
public synchronized void batchUpdate() {
95119
++this.nestedBatches;
96120
}
97121

98122
public synchronized void completeUpdateBatch() {
99-
if (--this.nestedBatches == 0 && !pendingNotifications.isEmpty()) {
100-
LOGGER.trace("Publishing batched changes");
101-
for (ConcurrentMap.Entry<HomekitClientConnection, ArrayList<PendingNotification>> entry :
102-
pendingNotifications.entrySet()) {
103-
try {
104-
HttpResponse message = new EventController().getMessage(entry.getValue());
105-
entry.getKey().outOfBand(message);
106-
} catch (Exception e) {
107-
LOGGER.warn("Failed to create new event message", e);
108-
}
123+
if (--this.nestedBatches == 0) flushUpdateBatch();
124+
}
125+
126+
private void flushUpdateBatch() {
127+
if (pendingNotifications.isEmpty()) return;
128+
129+
LOGGER.trace("Publishing batched changes");
130+
for (Map.Entry<HomekitClientConnection, List<PendingNotification>> entry :
131+
pendingNotifications.entrySet()) {
132+
try {
133+
HttpResponse message = new EventController().getMessage(entry.getValue());
134+
entry.getKey().outOfBand(message);
135+
} catch (Exception e) {
136+
LOGGER.warn("Failed to create new event message", e);
109137
}
110-
pendingNotifications.clear();
111138
}
139+
pendingNotifications.clear();
112140
}
113141

114142
public synchronized void publish(int accessoryId, int iid, EventableCharacteristic changed) {
115-
final Set<HomekitClientConnection> subscribers = subscriptions.get(changed);
116-
if ((subscribers == null) || (subscribers.isEmpty())) {
117-
LOGGER.debug("No subscribers to characteristic {} at accessory {} ", changed, accessoryId);
143+
final ConnectionsWithIds subscribers = subscriptions.get(changed);
144+
if (subscribers == null || subscribers.connections.isEmpty()) {
145+
LOGGER.trace("No subscribers to characteristic {} at accessory {} ", changed, accessoryId);
118146
return; // no subscribers
119147
}
120148
if (nestedBatches != 0) {
121149
LOGGER.trace("Batching change for accessory {} and characteristic {} " + accessoryId, iid);
122150
PendingNotification notification = new PendingNotification(accessoryId, iid, changed);
123-
for (HomekitClientConnection connection : subscribers) {
151+
for (HomekitClientConnection connection : subscribers.connections) {
124152
if (!pendingNotifications.containsKey(connection)) {
125153
pendingNotifications.put(connection, new ArrayList<PendingNotification>());
126154
}
@@ -132,22 +160,97 @@ public synchronized void publish(int accessoryId, int iid, EventableCharacterist
132160
try {
133161
HttpResponse message = new EventController().getMessage(accessoryId, iid, changed);
134162
LOGGER.trace("Publishing change for " + accessoryId);
135-
for (HomekitClientConnection connection : subscribers) {
163+
for (HomekitClientConnection connection : subscribers.connections) {
136164
connection.outOfBand(message);
137165
}
138166
} catch (Exception e) {
139167
LOGGER.warn("Failed to create new event message", e);
140168
}
141169
}
142170

171+
/**
172+
* The accessory registry has changed; go through all subscriptions and link to any new/changed
173+
* characteristics
174+
*/
175+
public synchronized void resync(HomekitRegistry registry) {
176+
LOGGER.trace("Resyncing subscriptions");
177+
flushUpdateBatch();
178+
179+
Map<EventableCharacteristic, ConnectionsWithIds> newSubscriptions = new HashMap<>();
180+
Iterator<Map.Entry<EventableCharacteristic, ConnectionsWithIds>> i =
181+
subscriptions.entrySet().iterator();
182+
while (i.hasNext()) {
183+
Map.Entry<EventableCharacteristic, ConnectionsWithIds> entry = i.next();
184+
EventableCharacteristic oldCharacteristic = entry.getKey();
185+
ConnectionsWithIds subscribers = entry.getValue();
186+
Characteristic newCharacteristic =
187+
registry.getCharacteristics(subscribers.aid).get(subscribers.iid);
188+
if (newCharacteristic == null || newCharacteristic.getType() != oldCharacteristic.getType()) {
189+
// characteristic is gone or has completely changed; drop all subscriptions for it
190+
LOGGER.trace(
191+
"{}:{} ({}) has gone missing; dropping subscriptions.",
192+
subscribers.aid,
193+
subscribers.iid,
194+
oldCharacteristic.getClass().getSimpleName());
195+
i.remove();
196+
for (HomekitClientConnection conn : subscribers.connections) {
197+
removeSubscription(oldCharacteristic, conn);
198+
}
199+
} else if (newCharacteristic != oldCharacteristic) {
200+
EventableCharacteristic newEventableCharacteristic =
201+
(EventableCharacteristic) newCharacteristic;
202+
LOGGER.trace(
203+
"{}:{} has been replaced by a compatible characteristic; re-connecting subscriptions",
204+
subscribers.aid,
205+
subscribers.iid);
206+
// characteristic has been replaced by another instance of the same thing;
207+
// re-connect subscriptions
208+
i.remove();
209+
oldCharacteristic.unsubscribe();
210+
subscribe(subscribers.aid, subscribers.iid, newEventableCharacteristic);
211+
// we can't replace the key, and we can't add to the map while we're iterating,
212+
// so save it off to a temporary map that we'll add them all at the end
213+
newSubscriptions.put(newEventableCharacteristic, subscribers);
214+
215+
for (HomekitClientConnection conn : subscribers.connections) {
216+
Set<EventableCharacteristic> subscribedCharacteristics = reverse.get(conn);
217+
subscribedCharacteristics.remove(oldCharacteristic);
218+
subscribedCharacteristics.add(newEventableCharacteristic);
219+
220+
// and also update references for any pending notifications, so they'll get the proper
221+
// value
222+
List<PendingNotification> connectionPendingNotifications = pendingNotifications.get(conn);
223+
if (connectionPendingNotifications != null) {
224+
for (PendingNotification notification : connectionPendingNotifications) {
225+
if (notification.characteristic == oldCharacteristic) {
226+
notification.characteristic = newEventableCharacteristic;
227+
}
228+
}
229+
}
230+
}
231+
}
232+
}
233+
subscriptions.putAll(newSubscriptions);
234+
}
235+
236+
private void subscribe(int aid, int iid, EventableCharacteristic characteristic) {
237+
characteristic.subscribe(
238+
() -> {
239+
publish(aid, iid, characteristic);
240+
});
241+
}
242+
143243
/** Remove all existing subscriptions */
144-
public void removeAll() {
244+
public synchronized void removeAll() {
145245
LOGGER.trace("Removing {} reverse connections from subscription manager", reverse.size());
146-
Iterator<HomekitClientConnection> i = reverse.keySet().iterator();
246+
Iterator<Map.Entry<HomekitClientConnection, Set<EventableCharacteristic>>> i =
247+
reverse.entrySet().iterator();
147248
while (i.hasNext()) {
148-
HomekitClientConnection connection = i.next();
249+
Map.Entry<HomekitClientConnection, Set<EventableCharacteristic>> entry = i.next();
250+
HomekitClientConnection connection = entry.getKey();
149251
LOGGER.trace("Removing connection {}", connection.hashCode());
150-
removeConnection(connection);
252+
i.remove();
253+
removeConnection(connection, entry.getValue());
151254
}
152255
LOGGER.trace("Subscription sizes are {} and {}", reverse.size(), subscriptions.size());
153256
}

‎src/main/java/io/github/hapjava/server/impl/json/EventController.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import io.github.hapjava.server.impl.connections.PendingNotification;
55
import io.github.hapjava.server.impl.http.HttpResponse;
66
import java.io.ByteArrayOutputStream;
7-
import java.util.ArrayList;
7+
import java.util.List;
88
import javax.json.Json;
99
import javax.json.JsonArrayBuilder;
1010
import javax.json.JsonObject;
@@ -34,7 +34,7 @@ public HttpResponse getMessage(int accessoryId, int iid, EventableCharacteristic
3434
}
3535
}
3636

37-
public HttpResponse getMessage(ArrayList<PendingNotification> notifications) throws Exception {
37+
public HttpResponse getMessage(List<PendingNotification> notifications) throws Exception {
3838
JsonArrayBuilder characteristics = Json.createArrayBuilder();
3939

4040
for (PendingNotification notification : notifications) {

0 commit comments

Comments
 (0)
Please sign in to comment.