Combine active device and session

12 files changed, 143 insertions, 158 deletions

diff --git a/src/main/java/org/traccar/BaseProtocolDecoder.java b/src/main/java/org/traccar/BaseProtocolDecoder.java
index 3fc6e7697..d6c571b79 100644
--- a/src/main/java/org/traccar/BaseProtocolDecoder.java
+++ b/src/main/java/org/traccar/BaseProtocolDecoder.java
@@ -18,14 +18,9 @@ package org.traccar;
 import com.google.inject.Inject;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
-import io.netty.channel.socket.DatagramChannel;
-import io.netty.handler.codec.http.HttpRequestDecoder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.traccar.config.Config;
 import org.traccar.config.Keys;
 import org.traccar.database.CommandsManager;
-import org.traccar.session.ConnectionManager;
 import org.traccar.database.IdentityManager;
 import org.traccar.database.MediaManager;
 import org.traccar.database.StatisticsManager;
@@ -33,22 +28,19 @@ import org.traccar.helper.UnitsConverter;
 import org.traccar.model.Command;
 import org.traccar.model.Device;
 import org.traccar.model.Position;
+import org.traccar.session.ConnectionManager;
 import org.traccar.session.DeviceSession;
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Collection;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.HashSet;
-import java.util.Map;
 import java.util.Set;
 import java.util.TimeZone;
 
 public abstract class BaseProtocolDecoder extends ExtendedObjectDecoder {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(BaseProtocolDecoder.class);
-
     private static final String PROTOCOL_UNKNOWN = "unknown";
 
     private final Protocol protocol;
@@ -147,95 +139,8 @@ public abstract class BaseProtocolDecoder extends ExtendedObjectDecoder {
         return result;
     }
 
-    private DeviceSession channelDeviceSession; // connection-based protocols
-    private final Map<SocketAddress, DeviceSession> addressDeviceSessions = new HashMap<>(); // connectionless protocols
-
-    private long findDeviceId(SocketAddress remoteAddress, String... uniqueIds) {
-        if (uniqueIds.length > 0) {
-            long deviceId = 0;
-            Device device = null;
-            try {
-                for (String uniqueId : uniqueIds) {
-                    if (uniqueId != null) {
-                        device = identityManager.getByUniqueId(uniqueId);
-                        if (device != null) {
-                            deviceId = device.getId();
-                            break;
-                        }
-                    }
-                }
-            } catch (Exception e) {
-                LOGGER.warn("Find device error", e);
-            }
-            if (deviceId == 0 && config.getBoolean(Keys.DATABASE_REGISTER_UNKNOWN)) {
-                return identityManager.addUnknownDevice(uniqueIds[0]);
-            }
-            if (device != null && !device.getDisabled()) {
-                return deviceId;
-            }
-            StringBuilder message = new StringBuilder();
-            if (deviceId == 0) {
-                message.append("Unknown device -");
-            } else {
-                message.append("Disabled device -");
-            }
-            for (String uniqueId : uniqueIds) {
-                message.append(" ").append(uniqueId);
-            }
-            if (remoteAddress != null) {
-                message.append(" (").append(((InetSocketAddress) remoteAddress).getHostString()).append(")");
-            }
-            LOGGER.warn(message.toString());
-        }
-        return 0;
-    }
-
     public DeviceSession getDeviceSession(Channel channel, SocketAddress remoteAddress, String... uniqueIds) {
-        return getDeviceSession(channel, remoteAddress, false, uniqueIds);
-    }
-
-    public DeviceSession getDeviceSession(
-            Channel channel, SocketAddress remoteAddress, boolean ignoreCache, String... uniqueIds) {
-        if (channel != null && BasePipelineFactory.getHandler(channel.pipeline(), HttpRequestDecoder.class) != null
-                || ignoreCache || config.getBoolean(Keys.PROTOCOL_IGNORE_SESSIONS_CACHE.withPrefix(getProtocolName()))
-                || config.getBoolean(Keys.DECODER_IGNORE_SESSIONS_CACHE)) {
-            long deviceId = findDeviceId(remoteAddress, uniqueIds);
-            if (deviceId != 0) {
-                if (connectionManager != null) {
-                    connectionManager.addActiveDevice(deviceId, protocol, channel, remoteAddress);
-                }
-                return new DeviceSession(deviceId);
-            } else {
-                return null;
-            }
-        }
-        if (channel instanceof DatagramChannel) {
-            long deviceId = findDeviceId(remoteAddress, uniqueIds);
-            DeviceSession deviceSession = addressDeviceSessions.get(remoteAddress);
-            if (deviceSession != null && (deviceSession.getDeviceId() == deviceId || uniqueIds.length == 0)) {
-                return deviceSession;
-            } else if (deviceId != 0) {
-                deviceSession = new DeviceSession(deviceId);
-                addressDeviceSessions.put(remoteAddress, deviceSession);
-                if (connectionManager != null) {
-                    connectionManager.addActiveDevice(deviceId, protocol, channel, remoteAddress);
-                }
-                return deviceSession;
-            } else {
-                return null;
-            }
-        } else {
-            if (channelDeviceSession == null) {
-                long deviceId = findDeviceId(remoteAddress, uniqueIds);
-                if (deviceId != 0) {
-                    channelDeviceSession = new DeviceSession(deviceId);
-                    if (connectionManager != null) {
-                        connectionManager.addActiveDevice(deviceId, protocol, channel, remoteAddress);
-                    }
-                }
-            }
-            return channelDeviceSession;
-        }
+        return connectionManager.getDeviceSession(protocol, channel, remoteAddress, uniqueIds);
     }
 
     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 91706222a..0a25b7547 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().removeActiveDevice(ctx.channel());
+            Context.getConnectionManager().removeDeviceSessions(ctx.channel());
         }
     }
 
diff --git a/src/main/java/org/traccar/config/Keys.java b/src/main/java/org/traccar/config/Keys.java
index eebdf7172..f5370874d 100644
--- a/src/main/java/org/traccar/config/Keys.java
+++ b/src/main/java/org/traccar/config/Keys.java
@@ -191,13 +191,6 @@ public final class Keys {
             Collections.singletonList(KeyType.GLOBAL));
 
     /**
-     * Skip device connection session cache. Per protocol configuration.
-     */
-    public static final ConfigSuffix<Boolean> PROTOCOL_IGNORE_SESSIONS_CACHE = new ConfigSuffix<>(
-            ".ignoreSessionCache",
-            Collections.singletonList(KeyType.GLOBAL));
-
-    /**
      * ORBCOMM API access id.
      */
     public static final ConfigKey<String> ORBCOMM_ACCESS_ID = new ConfigKey<>(
@@ -212,13 +205,6 @@ public final class Keys {
             Collections.singletonList(KeyType.GLOBAL));
 
     /**
-     * Skip device connection session cache. Global configuration.
-     */
-    public static final ConfigKey<Boolean> DECODER_IGNORE_SESSIONS_CACHE = new ConfigKey<>(
-            "decoder.ignoreSessionCache",
-            Collections.singletonList(KeyType.GLOBAL));
-
-    /**
      * Server wide connection timeout value in seconds. See protocol timeout for more information.
      */
     public static final ConfigKey<Integer> SERVER_TIMEOUT = new ConfigKey<>(
diff --git a/src/main/java/org/traccar/database/CommandsManager.java b/src/main/java/org/traccar/database/CommandsManager.java
index 3adf5d2e9..57ce0f9a4 100644
--- a/src/main/java/org/traccar/database/CommandsManager.java
+++ b/src/main/java/org/traccar/database/CommandsManager.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 - 2020 Anton Tananaev (anton@traccar.org)
+ * Copyright 2017 - 2022 Anton Tananaev (anton@traccar.org)
  * Copyright 2017 Andrey Kunitsyn (andrey@traccar.org)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -34,7 +34,7 @@ import org.traccar.Context;
 import org.traccar.model.Command;
 import org.traccar.model.Typed;
 import org.traccar.model.Position;
-import org.traccar.session.ActiveDevice;
+import org.traccar.session.DeviceSession;
 
 public class CommandsManager  extends ExtendedObjectManager<Command> {
 
@@ -75,10 +75,10 @@ public class CommandsManager  extends ExtendedObjectManager<Command> {
                 throw new RuntimeException("Command " + command.getType() + " is not supported");
             }
         } else {
-            ActiveDevice activeDevice = Context.getConnectionManager().getActiveDevice(deviceId);
-            if (activeDevice != null) {
-                if (activeDevice.supportsLiveCommands()) {
-                    activeDevice.sendCommand(command);
+            DeviceSession deviceSession = Context.getConnectionManager().getDeviceSession(deviceId);
+            if (deviceSession != null) {
+                if (deviceSession.supportsLiveCommands()) {
+                    deviceSession.sendCommand(command);
                 } else {
                     getDeviceQueue(deviceId).add(command);
                     return false;
diff --git a/src/main/java/org/traccar/database/DeviceManager.java b/src/main/java/org/traccar/database/DeviceManager.java
index 0e5056e57..b1ea0b8b7 100644
--- a/src/main/java/org/traccar/database/DeviceManager.java
+++ b/src/main/java/org/traccar/database/DeviceManager.java
@@ -68,7 +68,7 @@ public class DeviceManager extends BaseObjectManager<Device> implements Identity
     }
 
     @Override
-    public long addUnknownDevice(String uniqueId) {
+    public Device addUnknownDevice(String uniqueId) {
         Device device = new Device();
         device.setName(uniqueId);
         device.setUniqueId(uniqueId);
@@ -89,10 +89,10 @@ public class DeviceManager extends BaseObjectManager<Device> implements Identity
                 Context.getPermissionsManager().refreshAllExtendedPermissions();
             }
 
-            return device.getId();
+            return device;
         } catch (StorageException e) {
             LOGGER.warn("Automatic device registration error", e);
-            return 0;
+            return null;
         }
     }
 
diff --git a/src/main/java/org/traccar/database/IdentityManager.java b/src/main/java/org/traccar/database/IdentityManager.java
index af6a6ce71..ee386fdfd 100644
--- a/src/main/java/org/traccar/database/IdentityManager.java
+++ b/src/main/java/org/traccar/database/IdentityManager.java
@@ -20,7 +20,7 @@ import org.traccar.model.Position;
 
 public interface IdentityManager {
 
-    long addUnknownDevice(String uniqueId);
+    Device addUnknownDevice(String uniqueId);
 
     Device getById(long id);
 
diff --git a/src/main/java/org/traccar/protocol/EgtsProtocolDecoder.java b/src/main/java/org/traccar/protocol/EgtsProtocolDecoder.java
index 3a6af60a1..01d329580 100644
--- a/src/main/java/org/traccar/protocol/EgtsProtocolDecoder.java
+++ b/src/main/java/org/traccar/protocol/EgtsProtocolDecoder.java
@@ -291,7 +291,7 @@ public class EgtsProtocolDecoder extends BaseProtocolDecoder {
 
             if (serviceType == SERVICE_TELEDATA && position.getValid()) {
                 if (useObjectIdAsDeviceId && objectId != 0L) {
-                    deviceSession = getDeviceSession(channel, remoteAddress, true, String.valueOf(objectId));
+                    deviceSession = getDeviceSession(channel, remoteAddress, String.valueOf(objectId));
                     if (deviceSession != null) {
                         position.setDeviceId(deviceSession.getDeviceId());
                     }
diff --git a/src/main/java/org/traccar/protocol/Gt06ProtocolDecoder.java b/src/main/java/org/traccar/protocol/Gt06ProtocolDecoder.java
index 0f89597ce..4b9757874 100644
--- a/src/main/java/org/traccar/protocol/Gt06ProtocolDecoder.java
+++ b/src/main/java/org/traccar/protocol/Gt06ProtocolDecoder.java
@@ -471,7 +471,7 @@ public class Gt06ProtocolDecoder extends BaseProtocolDecoder {
                     TimeZone timeZone = deviceSession.get(DeviceSession.KEY_TIMEZONE);
                     if (timeZone.getRawOffset() == 0) {
                         timeZone.setRawOffset(offset * 1000);
-                        deviceSession.setTimeZone(timeZone);
+                        deviceSession.set(DeviceSession.KEY_TIMEZONE, timeZone);
                     }
                 }
             }
diff --git a/src/main/java/org/traccar/protocol/OrbcommProtocolDecoder.java b/src/main/java/org/traccar/protocol/OrbcommProtocolDecoder.java
index 8ec47908f..1164d72a1 100644
--- a/src/main/java/org/traccar/protocol/OrbcommProtocolDecoder.java
+++ b/src/main/java/org/traccar/protocol/OrbcommProtocolDecoder.java
@@ -70,8 +70,7 @@ public class OrbcommProtocolDecoder extends BaseProtocolDecoder {
         JsonArray messages = json.getJsonArray("Messages");
         for (int i = 0; i < messages.size(); i++) {
             JsonObject message = messages.getJsonObject(i);
-            DeviceSession deviceSession = getDeviceSession(
-                    channel, remoteAddress, true, message.getString("MobileID"));
+            DeviceSession deviceSession = getDeviceSession(channel, remoteAddress, message.getString("MobileID"));
             if (deviceSession != null) {
 
                 Position position = new Position(getProtocolName());
diff --git a/src/main/java/org/traccar/session/ConnectionManager.java b/src/main/java/org/traccar/session/ConnectionManager.java
index fbc15b00d..5d8a8c606 100644
--- a/src/main/java/org/traccar/session/ConnectionManager.java
+++ b/src/main/java/org/traccar/session/ConnectionManager.java
@@ -31,6 +31,7 @@ import org.traccar.model.Event;
 import org.traccar.model.Position;
 import org.traccar.storage.StorageException;
 
+import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Date;
 import java.util.HashMap;
@@ -47,7 +48,9 @@ public class ConnectionManager {
     private final long deviceTimeout;
     private final boolean updateDeviceState;
 
-    private final Map<Long, ActiveDevice> activeDevices = new ConcurrentHashMap<>();
+    private final Map<Long, DeviceSession> sessionsByDeviceId = new ConcurrentHashMap<>();
+    private final Map<Endpoint, Map<String, DeviceSession>> sessionsByEndpoint = new ConcurrentHashMap<>();
+
     private final Map<Long, Set<UpdateListener>> listeners = new ConcurrentHashMap<>();
     private final Map<Long, Timeout> timeouts = new ConcurrentHashMap<>();
 
@@ -59,22 +62,78 @@ public class ConnectionManager {
         timer = Main.getInjector().getInstance(Timer.class);
     }
 
-    public void addActiveDevice(long deviceId, Protocol protocol, Channel channel, SocketAddress remoteAddress) {
-        activeDevices.put(deviceId, new ActiveDevice(deviceId, protocol, channel, remoteAddress));
+    public DeviceSession getDeviceSession(long deviceId) {
+        return sessionsByDeviceId.get(deviceId);
     }
 
-    public void removeActiveDevice(Channel channel) {
-        for (ActiveDevice activeDevice : activeDevices.values()) {
-            if (activeDevice.getChannel() == channel) {
-                updateDevice(activeDevice.getDeviceId(), Device.STATUS_OFFLINE, null);
-                activeDevices.remove(activeDevice.getDeviceId());
-                break;
+    public DeviceSession getDeviceSession(
+            Protocol protocol, Channel channel, SocketAddress remoteAddress, String... uniqueIds) {
+
+        Endpoint endpoint = new Endpoint(channel, remoteAddress);
+        Map<String, DeviceSession> endpointSessions = sessionsByEndpoint.getOrDefault(
+                endpoint, new ConcurrentHashMap<>());
+        if (uniqueIds.length > 0) {
+            for (String uniqueId : uniqueIds) {
+                DeviceSession deviceSession = endpointSessions.get(uniqueId);
+                if (deviceSession != null) {
+                    return deviceSession;
+                }
             }
+        } else {
+            return endpointSessions.values().stream().findAny().orElse(null);
+        }
+
+        Device device = null;
+        try {
+            for (String uniqueId : uniqueIds) {
+                device = Context.getIdentityManager().getByUniqueId(uniqueId);
+                if (device != null) {
+                    break;
+                }
+            }
+        } catch (Exception e) {
+            LOGGER.warn("Find device error", e);
+        }
+
+        if (device == null && Context.getConfig().getBoolean(Keys.DATABASE_REGISTER_UNKNOWN)) {
+            device = Context.getIdentityManager().addUnknownDevice(uniqueIds[0]);
+        }
+
+        if (device != null && !device.getDisabled()) {
+            DeviceSession oldSession = sessionsByDeviceId.remove(device.getId());
+            if (oldSession != null) {
+                Endpoint oldEndpoint = new Endpoint(oldSession.getChannel(), oldSession.getRemoteAddress());
+                Map<String, DeviceSession> oldEndpointSessions = sessionsByEndpoint.get(oldEndpoint);
+                if (oldEndpointSessions.size() > 1) {
+                    oldEndpointSessions.remove(device.getUniqueId());
+                } else {
+                    sessionsByEndpoint.remove(oldEndpoint);
+                }
+            }
+
+            DeviceSession deviceSession = new DeviceSession(
+                    device.getId(), device.getUniqueId(), protocol, channel, remoteAddress);
+            endpointSessions.put(device.getUniqueId(), deviceSession);
+            sessionsByEndpoint.put(endpoint, endpointSessions);
+            sessionsByDeviceId.put(device.getId(), deviceSession);
+
+            return deviceSession;
+        } else {
+            LOGGER.warn((device == null ? "Unknown" : "Disabled") + " device - " + String.join(" ", uniqueIds)
+                    + " (" + ((InetSocketAddress) remoteAddress).getHostString() + ")");
+            return null;
         }
     }
 
-    public ActiveDevice getActiveDevice(long deviceId) {
-        return activeDevices.get(deviceId);
+    public void removeDeviceSessions(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());
+            }
+        }
     }
 
     public void updateDevice(final long deviceId, String status, Date time) {
diff --git a/src/main/java/org/traccar/session/DeviceSession.java b/src/main/java/org/traccar/session/DeviceSession.java
index 0d5b283fe..009f90f5a 100644
--- a/src/main/java/org/traccar/session/DeviceSession.java
+++ b/src/main/java/org/traccar/session/DeviceSession.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 - 2018 Anton Tananaev (anton@traccar.org)
+ * Copyright 2016 - 2022 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.
@@ -15,21 +15,57 @@
  */
 package org.traccar.session;
 
+import io.netty.channel.Channel;
+import io.netty.handler.codec.http.HttpRequestDecoder;
+import org.traccar.BasePipelineFactory;
+import org.traccar.Protocol;
+import org.traccar.model.Command;
+
+import java.net.SocketAddress;
 import java.util.HashMap;
 import java.util.Map;
 
 public class DeviceSession {
 
     private final long deviceId;
+    private final String uniqueId;
+    private final Protocol protocol;
+    private final Channel channel;
+    private final SocketAddress remoteAddress;
 
-    public DeviceSession(long deviceId) {
+    public DeviceSession(
+            long deviceId, String uniqueId, Protocol protocol, Channel channel, SocketAddress remoteAddress) {
         this.deviceId = deviceId;
+        this.uniqueId = uniqueId;
+        this.protocol = protocol;
+        this.channel = channel;
+        this.remoteAddress = remoteAddress;
     }
 
     public long getDeviceId() {
         return deviceId;
     }
 
+    public String getUniqueId() {
+        return uniqueId;
+    }
+
+    public Channel getChannel() {
+        return channel;
+    }
+
+    public SocketAddress getRemoteAddress() {
+        return remoteAddress;
+    }
+
+    public boolean supportsLiveCommands() {
+        return BasePipelineFactory.getHandler(channel.pipeline(), HttpRequestDecoder.class) == null;
+    }
+
+    public void sendCommand(Command command) {
+        protocol.sendDataCommand(channel, remoteAddress, command);
+    }
+
     public static final String KEY_TIMEZONE = "timezone";
 
     private final Map<String, Object> locals = new HashMap<>();
diff --git a/src/main/java/org/traccar/session/ActiveDevice.java b/src/main/java/org/traccar/session/Endpoint.java
index af19ba55b..76aac3444 100644
--- a/src/main/java/org/traccar/session/ActiveDevice.java
+++ b/src/main/java/org/traccar/session/Endpoint.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 - 2020 Anton Tananaev (anton@traccar.org)
+ * Copyright 2022 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.
@@ -16,43 +16,43 @@
 package org.traccar.session;
 
 import io.netty.channel.Channel;
-import io.netty.handler.codec.http.HttpRequestDecoder;
-import org.traccar.BasePipelineFactory;
-import org.traccar.Protocol;
-import org.traccar.model.Command;
 
 import java.net.SocketAddress;
+import java.util.Objects;
 
-public class ActiveDevice {
+public class Endpoint {
 
-    private final long deviceId;
-    private final Protocol protocol;
     private final Channel channel;
     private final SocketAddress remoteAddress;
-    private final boolean supportsLiveCommands;
 
-    public ActiveDevice(long deviceId, Protocol protocol, Channel channel, SocketAddress remoteAddress) {
-        this.deviceId = deviceId;
-        this.protocol = protocol;
+    public Endpoint(Channel channel, SocketAddress remoteAddress) {
         this.channel = channel;
         this.remoteAddress = remoteAddress;
-        supportsLiveCommands = BasePipelineFactory.getHandler(channel.pipeline(), HttpRequestDecoder.class) == null;
     }
 
     public Channel getChannel() {
         return channel;
     }
 
-    public long getDeviceId() {
-        return deviceId;
+    public SocketAddress getRemoteAddress() {
+        return remoteAddress;
     }
 
-    public boolean supportsLiveCommands() {
-        return supportsLiveCommands;
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        Endpoint endpoint = (Endpoint) o;
+        return channel.equals(endpoint.channel) && remoteAddress.equals(endpoint.remoteAddress);
     }
 
-    public void sendCommand(Command command) {
-        protocol.sendDataCommand(channel, remoteAddress, command);
+    @Override
+    public int hashCode() {
+        return Objects.hash(channel, remoteAddress);
     }
 
 }