Update broadcast interface

11 files changed, 178 insertions, 47 deletions

diff --git a/src/main/java/org/traccar/api/BaseObjectResource.java b/src/main/java/org/traccar/api/BaseObjectResource.java
index 2aaed2bb5..02e1c2cbe 100644
--- a/src/main/java/org/traccar/api/BaseObjectResource.java
+++ b/src/main/java/org/traccar/api/BaseObjectResource.java
@@ -17,6 +17,7 @@
 package org.traccar.api;
 
 import org.traccar.api.security.ServiceAccountUser;
+import org.traccar.broadcast.ObjectOperation;
 import org.traccar.helper.LogAction;
 import org.traccar.model.BaseModel;
 import org.traccar.model.Group;
@@ -74,8 +75,8 @@ public abstract class BaseObjectResource<T extends BaseModel> extends BaseResour
 
         if (getUserId() != ServiceAccountUser.ID) {
             storage.addPermission(new Permission(User.class, getUserId(), baseClass, entity.getId()));
-            cacheManager.invalidatePermission(true, User.class, getUserId(), baseClass, entity.getId());
-            connectionManager.invalidatePermission(true, User.class, getUserId(), baseClass, entity.getId());
+            cacheManager.invalidatePermission(true, User.class, getUserId(), baseClass, entity.getId(), true);
+            connectionManager.invalidatePermission(true, User.class, getUserId(), baseClass, entity.getId(), true);
             LogAction.link(getUserId(), User.class, getUserId(), baseClass, entity.getId());
         }
 
@@ -110,7 +111,7 @@ public abstract class BaseObjectResource<T extends BaseModel> extends BaseResour
                         new Condition.Equals("id", entity.getId())));
             }
         }
-        cacheManager.updateOrInvalidate(true, entity);
+        cacheManager.updateOrInvalidate(true, entity, ObjectOperation.UPDATE);
         LogAction.edit(getUserId(), entity);
 
         return Response.ok(entity).build();
diff --git a/src/main/java/org/traccar/api/resource/PermissionsResource.java b/src/main/java/org/traccar/api/resource/PermissionsResource.java
index e8e4e96eb..2a8ac62f7 100644
--- a/src/main/java/org/traccar/api/resource/PermissionsResource.java
+++ b/src/main/java/org/traccar/api/resource/PermissionsResource.java
@@ -74,7 +74,8 @@ public class PermissionsResource  extends BaseResource {
             cacheManager.invalidatePermission(
                     true,
                     permission.getOwnerClass(), permission.getOwnerId(),
-                    permission.getPropertyClass(), permission.getPropertyId());
+                    permission.getPropertyClass(), permission.getPropertyId(),
+                    true);
             LogAction.link(getUserId(),
                     permission.getOwnerClass(), permission.getOwnerId(),
                     permission.getPropertyClass(), permission.getPropertyId());
@@ -99,7 +100,8 @@ public class PermissionsResource  extends BaseResource {
             cacheManager.invalidatePermission(
                     true,
                     permission.getOwnerClass(), permission.getOwnerId(),
-                    permission.getPropertyClass(), permission.getPropertyId());
+                    permission.getPropertyClass(), permission.getPropertyId(),
+                    false);
             LogAction.unlink(getUserId(),
                     permission.getOwnerClass(), permission.getOwnerId(),
                     permission.getPropertyClass(), permission.getPropertyId());
diff --git a/src/main/java/org/traccar/api/resource/ServerResource.java b/src/main/java/org/traccar/api/resource/ServerResource.java
index 8149ec3b8..bcd36a32e 100644
--- a/src/main/java/org/traccar/api/resource/ServerResource.java
+++ b/src/main/java/org/traccar/api/resource/ServerResource.java
@@ -16,6 +16,7 @@
 package org.traccar.api.resource;
 
 import org.traccar.api.BaseResource;
+import org.traccar.broadcast.ObjectOperation;
 import org.traccar.config.Config;
 import org.traccar.config.Keys;
 import org.traccar.database.OpenIdProvider;
@@ -111,7 +112,7 @@ public class ServerResource extends BaseResource {
         storage.updateObject(entity, new Request(
                 new Columns.Exclude("id"),
                 new Condition.Equals("id", entity.getId())));
-        cacheManager.updateOrInvalidate(true, entity);
+        cacheManager.updateOrInvalidate(true, entity, ObjectOperation.UPDATE);
         LogAction.edit(getUserId(), entity);
         return Response.ok(entity).build();
     }
diff --git a/src/main/java/org/traccar/broadcast/BaseBroadcastService.java b/src/main/java/org/traccar/broadcast/BaseBroadcastService.java
index a95d333f2..bb220d2bb 100644
--- a/src/main/java/org/traccar/broadcast/BaseBroadcastService.java
+++ b/src/main/java/org/traccar/broadcast/BaseBroadcastService.java
@@ -16,7 +16,6 @@
 package org.traccar.broadcast;
 
 import java.util.HashSet;
-import java.util.Map;
 import java.util.Set;
 
 import org.traccar.model.BaseModel;
@@ -69,9 +68,16 @@ public abstract class BaseBroadcastService implements BroadcastService {
     }
 
     @Override
-    public void invalidateObject(boolean local, Class<? extends BaseModel> clazz, long id) {
+    public void invalidateObject(
+            boolean local,
+            Class<? extends BaseModel> clazz, long id,
+            ObjectOperation operation) {
         BroadcastMessage message = new BroadcastMessage();
-        message.setChanges(Map.of(Permission.getKey(clazz), id));
+        var invalidateObject = new BroadcastMessage.InvalidateObject();
+        invalidateObject.setClazz(Permission.getKey(clazz));
+        invalidateObject.setId(id);
+        invalidateObject.setOperation(operation);
+        message.setInvalidateObject(invalidateObject);
         sendMessage(message);
     }
 
@@ -79,9 +85,16 @@ public abstract class BaseBroadcastService implements BroadcastService {
     public void invalidatePermission(
             boolean local,
             Class<? extends BaseModel> clazz1, long id1,
-            Class<? extends BaseModel> clazz2, long id2) {
+            Class<? extends BaseModel> clazz2, long id2,
+            boolean link) {
         BroadcastMessage message = new BroadcastMessage();
-        message.setChanges(Map.of(Permission.getKey(clazz1), id1, Permission.getKey(clazz2), id2));
+        var invalidatePermission = new BroadcastMessage.InvalidatePermission();
+        invalidatePermission.setClazz1(Permission.getKey(clazz1));
+        invalidatePermission.setId1(id1);
+        invalidatePermission.setClazz2(Permission.getKey(clazz2));
+        invalidatePermission.setId2(id2);
+        invalidatePermission.setLink(link);
+        message.setInvalidatePermission(invalidatePermission);
         sendMessage(message);
     }
 
@@ -96,22 +109,19 @@ public abstract class BaseBroadcastService implements BroadcastService {
             listeners.forEach(listener -> listener.updateEvent(false, message.getUserId(), message.getEvent()));
         } else if (message.getCommandDeviceId() != null) {
             listeners.forEach(listener -> listener.updateCommand(false, message.getCommandDeviceId()));
-        } else if (message.getChanges() != null) {
-            var iterator = message.getChanges().entrySet().iterator();
-            if (iterator.hasNext()) {
-                var first = iterator.next();
-                if (iterator.hasNext()) {
-                    var second = iterator.next();
-                    listeners.forEach(listener -> listener.invalidatePermission(
-                            false,
-                            Permission.getKeyClass(first.getKey()), first.getValue(),
-                            Permission.getKeyClass(second.getKey()), second.getValue()));
-                } else {
-                    listeners.forEach(listener -> listener.invalidateObject(
-                            false,
-                            Permission.getKeyClass(first.getKey()), first.getValue()));
-                }
-            }
+        } else if (message.getInvalidateObject() != null) {
+            var invalidateObject = message.getInvalidateObject();
+            listeners.forEach(listeners -> listeners.invalidateObject(
+                    false,
+                    Permission.getKeyClass(invalidateObject.getClazz()), invalidateObject.getId(),
+                    invalidateObject.getOperation()));
+        } else if (message.getInvalidatePermission() != null) {
+            var invalidatePermission = message.getInvalidatePermission();
+            listeners.forEach(listener -> listener.invalidatePermission(
+                    false,
+                    Permission.getKeyClass(invalidatePermission.getClazz1()), invalidatePermission.getId1(),
+                    Permission.getKeyClass(invalidatePermission.getClazz2()), invalidatePermission.getId2(),
+                    invalidatePermission.getLink()));
         }
     }
 
diff --git a/src/main/java/org/traccar/broadcast/BroadcastInterface.java b/src/main/java/org/traccar/broadcast/BroadcastInterface.java
index 673ebd8b8..ededbaa1a 100644
--- a/src/main/java/org/traccar/broadcast/BroadcastInterface.java
+++ b/src/main/java/org/traccar/broadcast/BroadcastInterface.java
@@ -34,12 +34,16 @@ public interface BroadcastInterface {
     default void updateCommand(boolean local, long deviceId) {
     }
 
-    default void invalidateObject(boolean local, Class<? extends BaseModel> clazz, long id) {
+    default void invalidateObject(
+            boolean local,
+            Class<? extends BaseModel> clazz, long id,
+            ObjectOperation operation) {
     }
 
     default void invalidatePermission(
             boolean local,
             Class<? extends BaseModel> clazz1, long id1,
-            Class<? extends BaseModel> clazz2, long id2) {
+            Class<? extends BaseModel> clazz2, long id2,
+            boolean link) {
     }
 }
diff --git a/src/main/java/org/traccar/broadcast/BroadcastMessage.java b/src/main/java/org/traccar/broadcast/BroadcastMessage.java
index 985848d04..6fe2d8b35 100644
--- a/src/main/java/org/traccar/broadcast/BroadcastMessage.java
+++ b/src/main/java/org/traccar/broadcast/BroadcastMessage.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 Anton Tananaev (anton@traccar.org)
+ * Copyright 2022 - 2023 Anton Tananaev (anton@traccar.org)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,8 +19,6 @@ import org.traccar.model.Device;
 import org.traccar.model.Event;
 import org.traccar.model.Position;
 
-import java.util.Map;
-
 public class BroadcastMessage {
 
     private Device device;
@@ -73,13 +71,112 @@ public class BroadcastMessage {
         this.commandDeviceId = commandDeviceId;
     }
 
-    private Map<String, Long> changes;
+    public static class InvalidateObject {
+
+        private String clazz;
+
+        public String getClazz() {
+            return clazz;
+        }
+
+        public void setClazz(String clazz) {
+            this.clazz = clazz;
+        }
+
+        private long id;
+
+        public long getId() {
+            return id;
+        }
+
+        public void setId(long id) {
+            this.id = id;
+        }
+
+        private ObjectOperation operation;
+
+        public ObjectOperation getOperation() {
+            return operation;
+        }
+
+        public void setOperation(ObjectOperation operation) {
+            this.operation = operation;
+        }
 
-    public Map<String, Long> getChanges() {
-        return changes;
     }
 
-    public void setChanges(Map<String, Long> changes) {
-        this.changes = changes;
+    private InvalidateObject invalidateObject;
+
+    public InvalidateObject getInvalidateObject() {
+        return invalidateObject;
+    }
+
+    public void setInvalidateObject(InvalidateObject invalidateObject) {
+        this.invalidateObject = invalidateObject;
     }
+
+    public static class InvalidatePermission {
+
+        private String clazz1;
+
+        public String getClazz1() {
+            return clazz1;
+        }
+
+        public void setClazz1(String clazz1) {
+            this.clazz1 = clazz1;
+        }
+
+        private long id1;
+
+        public long getId1() {
+            return id1;
+        }
+
+        public void setId1(long id1) {
+            this.id1 = id1;
+        }
+
+        private String clazz2;
+
+        public String getClazz2() {
+            return clazz2;
+        }
+
+        public void setClazz2(String clazz2) {
+            this.clazz2 = clazz2;
+        }
+
+        private long id2;
+
+        public long getId2() {
+            return id2;
+        }
+
+        public void setId2(long id2) {
+            this.id2 = id2;
+        }
+
+        private boolean link;
+
+        public boolean getLink() {
+            return link;
+        }
+
+        public void setLink(boolean link) {
+            this.link = link;
+        }
+
+    }
+
+    private InvalidatePermission invalidatePermission;
+
+    public InvalidatePermission getInvalidatePermission() {
+        return invalidatePermission;
+    }
+
+    public void setInvalidatePermission(InvalidatePermission invalidatePermission) {
+        this.invalidatePermission = invalidatePermission;
+    }
+
 }
diff --git a/src/main/java/org/traccar/broadcast/ObjectOperation.java b/src/main/java/org/traccar/broadcast/ObjectOperation.java
new file mode 100644
index 000000000..27e5fb253
--- /dev/null
+++ b/src/main/java/org/traccar/broadcast/ObjectOperation.java
@@ -0,0 +1,7 @@
+package org.traccar.broadcast;
+
+public enum ObjectOperation {
+    ADD,
+    UPDATE,
+    DELETE,
+}
diff --git a/src/main/java/org/traccar/notificators/NotificatorFirebase.java b/src/main/java/org/traccar/notificators/NotificatorFirebase.java
index be95fb28e..0402db49d 100644
--- a/src/main/java/org/traccar/notificators/NotificatorFirebase.java
+++ b/src/main/java/org/traccar/notificators/NotificatorFirebase.java
@@ -29,6 +29,7 @@ import com.google.firebase.messaging.MessagingErrorCode;
 import com.google.firebase.messaging.MulticastMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.traccar.broadcast.ObjectOperation;
 import org.traccar.config.Config;
 import org.traccar.config.Keys;
 import org.traccar.model.Event;
@@ -135,7 +136,7 @@ public class NotificatorFirebase implements Notificator {
                     storage.updateObject(user, new Request(
                             new Columns.Include("attributes"),
                             new Condition.Equals("id", user.getId())));
-                    cacheManager.updateOrInvalidate(true, user);
+                    cacheManager.updateOrInvalidate(true, user, ObjectOperation.UPDATE);
                 }
             } catch (FirebaseMessagingException | StorageException e) {
                 LOGGER.warn("Firebase error", e);
diff --git a/src/main/java/org/traccar/notificators/NotificatorTraccar.java b/src/main/java/org/traccar/notificators/NotificatorTraccar.java
index e354adccb..f693e9f30 100644
--- a/src/main/java/org/traccar/notificators/NotificatorTraccar.java
+++ b/src/main/java/org/traccar/notificators/NotificatorTraccar.java
@@ -18,6 +18,7 @@ package org.traccar.notificators;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.traccar.broadcast.ObjectOperation;
 import org.traccar.config.Config;
 import org.traccar.config.Keys;
 import org.traccar.model.Event;
@@ -128,7 +129,7 @@ public class NotificatorTraccar implements Notificator {
                     storage.updateObject(user, new Request(
                             new Columns.Include("attributes"),
                             new Condition.Equals("id", user.getId())));
-                    cacheManager.updateOrInvalidate(true, user);
+                    cacheManager.updateOrInvalidate(true, user, ObjectOperation.UPDATE);
                 }
             } catch (StorageException e) {
                 LOGGER.warn("Push error", e);
diff --git a/src/main/java/org/traccar/session/ConnectionManager.java b/src/main/java/org/traccar/session/ConnectionManager.java
index 28214840d..0b13a5a72 100644
--- a/src/main/java/org/traccar/session/ConnectionManager.java
+++ b/src/main/java/org/traccar/session/ConnectionManager.java
@@ -330,8 +330,9 @@ public class ConnectionManager implements BroadcastInterface {
     public synchronized void invalidatePermission(
             boolean local,
             Class<? extends BaseModel> clazz1, long id1,
-            Class<? extends BaseModel> clazz2, long id2) {
-        if (clazz1.equals(User.class) && clazz2.equals(Device.class)) {
+            Class<? extends BaseModel> clazz2, long id2,
+            boolean link) {
+        if (link && clazz1.equals(User.class) && clazz2.equals(Device.class)) {
             if (listeners.containsKey(id1)) {
                 userDevices.get(id1).add(id2);
                 deviceUsers.put(id2, new HashSet<>(List.of(id1)));
diff --git a/src/main/java/org/traccar/session/cache/CacheManager.java b/src/main/java/org/traccar/session/cache/CacheManager.java
index 4b6422de2..c40916218 100644
--- a/src/main/java/org/traccar/session/cache/CacheManager.java
+++ b/src/main/java/org/traccar/session/cache/CacheManager.java
@@ -19,6 +19,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.traccar.broadcast.BroadcastInterface;
 import org.traccar.broadcast.BroadcastService;
+import org.traccar.broadcast.ObjectOperation;
 import org.traccar.config.Config;
 import org.traccar.model.Attribute;
 import org.traccar.model.BaseModel;
@@ -201,12 +202,15 @@ public class CacheManager implements BroadcastInterface {
     }
 
     @Override
-    public void invalidateObject(boolean local, Class<? extends BaseModel> clazz, long id) {
+    public void invalidateObject(
+            boolean local,
+            Class<? extends BaseModel> clazz, long id,
+            ObjectOperation operation) {
         try {
             var object = storage.getObject(clazz, new Request(
                     new Columns.All(), new Condition.Equals("id", id)));
             if (object != null) {
-                updateOrInvalidate(local, object);
+                updateOrInvalidate(local, object, operation);
             } else {
                 invalidate(clazz, id);
             }
@@ -215,9 +219,10 @@ public class CacheManager implements BroadcastInterface {
         }
     }
 
-    public <T extends BaseModel> void updateOrInvalidate(boolean local, T object) throws StorageException {
+    public <T extends BaseModel> void updateOrInvalidate(
+            boolean local, T object, ObjectOperation operation) throws StorageException {
         if (local) {
-            broadcastService.invalidateObject(true, object.getClass(), object.getId());
+            broadcastService.invalidateObject(true, object.getClass(), object.getId(), operation);
         }
 
         if (object instanceof Server) {
@@ -262,9 +267,10 @@ public class CacheManager implements BroadcastInterface {
     public void invalidatePermission(
             boolean local,
             Class<? extends BaseModel> clazz1, long id1,
-            Class<? extends BaseModel> clazz2, long id2) {
+            Class<? extends BaseModel> clazz2, long id2,
+            boolean link) {
         if (local) {
-            broadcastService.invalidatePermission(true, clazz1, id1, clazz2, id2);
+            broadcastService.invalidatePermission(true, clazz1, id1, clazz2, id2, link);
         }
 
         try {