Integrate cache manager

4 files changed, 36 insertions, 7 deletions

diff --git a/src/main/java/org/traccar/BaseProtocolDecoder.java b/src/main/java/org/traccar/BaseProtocolDecoder.java
index d6c571b79..71ef686fa 100644
--- a/src/main/java/org/traccar/BaseProtocolDecoder.java
+++ b/src/main/java/org/traccar/BaseProtocolDecoder.java
@@ -30,6 +30,7 @@ import org.traccar.model.Device;
 import org.traccar.model.Position;
 import org.traccar.session.ConnectionManager;
 import org.traccar.session.DeviceSession;
+import org.traccar.storage.StorageException;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -140,7 +141,11 @@ public abstract class BaseProtocolDecoder extends ExtendedObjectDecoder {
     }
 
     public DeviceSession getDeviceSession(Channel channel, SocketAddress remoteAddress, String... uniqueIds) {
-        return connectionManager.getDeviceSession(protocol, channel, remoteAddress, uniqueIds);
+        try {
+            return connectionManager.getDeviceSession(protocol, channel, remoteAddress, uniqueIds);
+        } catch (StorageException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     public void getLastLocation(Position position, Date deviceTime) {
diff --git a/src/main/java/org/traccar/MainEventHandler.java b/src/main/java/org/traccar/MainEventHandler.java
index 0a25b7547..f04307fdb 100644
--- a/src/main/java/org/traccar/MainEventHandler.java
+++ b/src/main/java/org/traccar/MainEventHandler.java
@@ -130,7 +130,7 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter {
 
         if (BasePipelineFactory.getHandler(ctx.pipeline(), HttpRequestDecoder.class) == null
                 && !connectionlessProtocols.contains(ctx.pipeline().get(BaseProtocolDecoder.class).getProtocolName())) {
-            Context.getConnectionManager().removeDeviceSessions(ctx.channel());
+            Context.getConnectionManager().deviceDisconnected(ctx.channel());
         }
     }
 
diff --git a/src/main/java/org/traccar/session/ConnectionManager.java b/src/main/java/org/traccar/session/ConnectionManager.java
index 5d8a8c606..e01a568aa 100644
--- a/src/main/java/org/traccar/session/ConnectionManager.java
+++ b/src/main/java/org/traccar/session/ConnectionManager.java
@@ -29,6 +29,7 @@ import org.traccar.handler.events.OverspeedEventHandler;
 import org.traccar.model.Device;
 import org.traccar.model.Event;
 import org.traccar.model.Position;
+import org.traccar.session.cache.CacheManager;
 import org.traccar.storage.StorageException;
 
 import java.net.InetSocketAddress;
@@ -51,6 +52,8 @@ public class ConnectionManager {
     private final Map<Long, DeviceSession> sessionsByDeviceId = new ConcurrentHashMap<>();
     private final Map<Endpoint, Map<String, DeviceSession>> sessionsByEndpoint = new ConcurrentHashMap<>();
 
+    private final CacheManager cacheManager;
+
     private final Map<Long, Set<UpdateListener>> listeners = new ConcurrentHashMap<>();
     private final Map<Long, Timeout> timeouts = new ConcurrentHashMap<>();
 
@@ -60,6 +63,7 @@ public class ConnectionManager {
         deviceTimeout = Context.getConfig().getLong(Keys.STATUS_TIMEOUT) * 1000;
         updateDeviceState = Context.getConfig().getBoolean(Keys.STATUS_UPDATE_DEVICE_STATE);
         timer = Main.getInjector().getInstance(Timer.class);
+        cacheManager = Main.getInjector().getInstance(CacheManager.class);
     }
 
     public DeviceSession getDeviceSession(long deviceId) {
@@ -67,7 +71,8 @@ public class ConnectionManager {
     }
 
     public DeviceSession getDeviceSession(
-            Protocol protocol, Channel channel, SocketAddress remoteAddress, String... uniqueIds) {
+            Protocol protocol, Channel channel, SocketAddress remoteAddress,
+            String... uniqueIds) throws StorageException {
 
         Endpoint endpoint = new Endpoint(channel, remoteAddress);
         Map<String, DeviceSession> endpointSessions = sessionsByEndpoint.getOrDefault(
@@ -116,6 +121,7 @@ public class ConnectionManager {
             endpointSessions.put(device.getUniqueId(), deviceSession);
             sessionsByEndpoint.put(endpoint, endpointSessions);
             sessionsByDeviceId.put(device.getId(), deviceSession);
+            cacheManager.addDevice(device.getId());
 
             return deviceSession;
         } else {
@@ -125,17 +131,31 @@ public class ConnectionManager {
         }
     }
 
-    public void removeDeviceSessions(Channel channel) {
+    public void deviceDisconnected(Channel channel) {
         Endpoint endpoint = new Endpoint(channel, channel.remoteAddress());
         Map<String, DeviceSession> endpointSessions = sessionsByEndpoint.remove(endpoint);
         if (endpointSessions != null) {
             for (DeviceSession deviceSession : endpointSessions.values()) {
                 updateDevice(deviceSession.getDeviceId(), Device.STATUS_OFFLINE, null);
                 sessionsByDeviceId.remove(deviceSession.getDeviceId());
+                cacheManager.removeDevice(deviceSession.getDeviceId());
             }
         }
     }
 
+    public void deviceUnknown(long deviceId) {
+        updateDevice(deviceId, Device.STATUS_UNKNOWN, null);
+        DeviceSession deviceSession = sessionsByDeviceId.remove(deviceId);
+        cacheManager.removeDevice(deviceId);
+        if (deviceSession != null) {
+            Endpoint endpoint = new Endpoint(deviceSession.getChannel(), deviceSession.getRemoteAddress());
+            sessionsByEndpoint.computeIfPresent(endpoint, (e, sessions) -> {
+                sessions.remove(deviceSession.getUniqueId());
+                return sessions.isEmpty() ? null : sessions;
+            });
+        }
+    }
+
     public void updateDevice(final long deviceId, String status, Date time) {
         Device device = Context.getIdentityManager().getById(deviceId);
         if (device == null) {
@@ -181,7 +201,7 @@ public class ConnectionManager {
         if (status.equals(Device.STATUS_ONLINE)) {
             timeouts.put(deviceId, timer.newTimeout(timeout1 -> {
                 if (!timeout1.isCancelled()) {
-                    updateDevice(deviceId, Device.STATUS_UNKNOWN, null);
+                    deviceUnknown(deviceId);
                 }
             }, deviceTimeout, TimeUnit.MILLISECONDS));
         }
diff --git a/src/main/java/org/traccar/session/cache/CacheManager.java b/src/main/java/org/traccar/session/cache/CacheManager.java
index ae514ce8b..d019b072b 100644
--- a/src/main/java/org/traccar/session/cache/CacheManager.java
+++ b/src/main/java/org/traccar/session/cache/CacheManager.java
@@ -99,7 +99,9 @@ public class CacheManager {
     public void addDevice(long deviceId) throws StorageException {
         try {
             lock.writeLock().lock();
-            unsafeAddDevice(deviceId);
+            if (!deviceLinks.containsKey(deviceId)) {
+                unsafeAddDevice(deviceId);
+            }
         } finally {
             lock.writeLock().unlock();
         }
@@ -108,7 +110,9 @@ public class CacheManager {
     public void removeDevice(long deviceId) {
         try {
             lock.writeLock().lock();
-            unsafeRemoveDevice(deviceId);
+            if (deviceLinks.containsKey(deviceId)) {
+                unsafeRemoveDevice(deviceId);
+            }
         } finally {
             lock.writeLock().unlock();
         }