1 files changed, 57 insertions, 24 deletions
diff --git a/src/main/java/org/traccar/MainEventHandler.java b/src/main/java/org/traccar/MainEventHandler.java
index a3f6f4105..fb0171d63 100644
--- a/src/main/java/org/traccar/MainEventHandler.java
+++ b/src/main/java/org/traccar/MainEventHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2012 - 2020 Anton Tananaev (anton@traccar.org)
+ * Copyright 2012 - 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,6 +16,7 @@
 package org.traccar;
 
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.socket.DatagramChannel;
@@ -23,17 +24,32 @@ import io.netty.handler.codec.http.HttpRequestDecoder;
 import io.netty.handler.timeout.IdleStateEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.traccar.config.Config;
 import org.traccar.config.Keys;
 import org.traccar.database.StatisticsManager;
+import org.traccar.handler.AcknowledgementHandler;
 import org.traccar.helper.DateUtil;
+import org.traccar.helper.NetworkUtil;
+import org.traccar.helper.model.PositionUtil;
+import org.traccar.model.Device;
 import org.traccar.model.Position;
+import org.traccar.session.ConnectionManager;
+import org.traccar.session.cache.CacheManager;
+import org.traccar.storage.Storage;
 import org.traccar.storage.StorageException;
+import org.traccar.storage.query.Columns;
+import org.traccar.storage.query.Condition;
+import org.traccar.storage.query.Request;
 
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.Set;
 
+@Singleton
+@ChannelHandler.Sharable
 public class MainEventHandler extends ChannelInboundHandlerAdapter {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(MainEventHandler.class);
@@ -41,13 +57,24 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter {
     private final Set<String> connectionlessProtocols = new HashSet<>();
     private final Set<String> logAttributes = new LinkedHashSet<>();
 
-    public MainEventHandler() {
-        String connectionlessProtocolList = Context.getConfig().getString(Keys.STATUS_IGNORE_OFFLINE);
+    private final CacheManager cacheManager;
+    private final Storage storage;
+    private final ConnectionManager connectionManager;
+    private final StatisticsManager statisticsManager;
+
+    @Inject
+    public MainEventHandler(
+            Config config, CacheManager cacheManager, Storage storage, ConnectionManager connectionManager,
+            StatisticsManager statisticsManager) {
+        this.cacheManager = cacheManager;
+        this.storage = storage;
+        this.connectionManager = connectionManager;
+        this.statisticsManager = statisticsManager;
+        String connectionlessProtocolList = config.getString(Keys.STATUS_IGNORE_OFFLINE);
         if (connectionlessProtocolList != null) {
             connectionlessProtocols.addAll(Arrays.asList(connectionlessProtocolList.split("[, ]")));
         }
-        logAttributes.addAll(Arrays.asList(
-                Context.getConfig().getString(Keys.LOGGER_ATTRIBUTES).split("[, ]")));
+        logAttributes.addAll(Arrays.asList(config.getString(Keys.LOGGER_ATTRIBUTES).split("[, ]")));
     }
 
     @Override
@@ -55,17 +82,27 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter {
         if (msg instanceof Position) {
 
             Position position = (Position) msg;
+            Device device = cacheManager.getObject(Device.class, position.getDeviceId());
+
             try {
-                Context.getDeviceManager().updateLatestPosition(position);
+                if (PositionUtil.isLatest(cacheManager, position)) {
+                    Device updatedDevice = new Device();
+                    updatedDevice.setId(position.getDeviceId());
+                    updatedDevice.setPositionId(position.getId());
+                    storage.updateObject(updatedDevice, new Request(
+                            new Columns.Include("positionId"),
+                            new Condition.Equals("id", updatedDevice.getId())));
+
+                    cacheManager.updatePosition(position);
+                    connectionManager.updatePosition(true, position);
+                }
             } catch (StorageException error) {
                 LOGGER.warn("Failed to update device", error);
             }
 
-            String uniqueId = Context.getIdentityManager().getById(position.getDeviceId()).getUniqueId();
-
             StringBuilder builder = new StringBuilder();
-            builder.append(formatChannel(ctx.channel())).append(" ");
-            builder.append("id: ").append(uniqueId);
+            builder.append("[").append(NetworkUtil.session(ctx.channel())).append("] ");
+            builder.append("id: ").append(device.getUniqueId());
             for (String attribute : logAttributes) {
                 switch (attribute) {
                     case "time":
@@ -108,31 +145,27 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter {
             }
             LOGGER.info(builder.toString());
 
-            Main.getInjector().getInstance(StatisticsManager.class)
-                    .registerMessageStored(position.getDeviceId(), position.getProtocol());
-        }
-    }
+            statisticsManager.registerMessageStored(position.getDeviceId(), position.getProtocol());
 
-    private static String formatChannel(Channel channel) {
-        return String.format("[%s]", channel.id().asShortText());
+            ctx.writeAndFlush(new AcknowledgementHandler.EventHandled(position));
+        }
     }
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) {
         if (!(ctx.channel() instanceof DatagramChannel)) {
-            LOGGER.info(formatChannel(ctx.channel()) + " connected");
+            LOGGER.info("[{}] connected", NetworkUtil.session(ctx.channel()));
         }
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
-        LOGGER.info(formatChannel(ctx.channel()) + " disconnected");
+        LOGGER.info("[{}] disconnected", NetworkUtil.session(ctx.channel()));
         closeChannel(ctx.channel());
 
-        if (BasePipelineFactory.getHandler(ctx.pipeline(), HttpRequestDecoder.class) == null
-                && !connectionlessProtocols.contains(ctx.pipeline().get(BaseProtocolDecoder.class).getProtocolName())) {
-            Context.getConnectionManager().removeActiveDevice(ctx.channel());
-        }
+        boolean supportsOffline = BasePipelineFactory.getHandler(ctx.pipeline(), HttpRequestDecoder.class) == null
+                && !connectionlessProtocols.contains(ctx.pipeline().get(BaseProtocolDecoder.class).getProtocolName());
+        connectionManager.deviceDisconnected(ctx.channel(), supportsOffline);
     }
 
     @Override
@@ -140,14 +173,14 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter {
         while (cause.getCause() != null && cause.getCause() != cause) {
             cause = cause.getCause();
         }
-        LOGGER.warn(formatChannel(ctx.channel()) + " error", cause);
+        LOGGER.info("[{}] error", NetworkUtil.session(ctx.channel()), cause);
         closeChannel(ctx.channel());
     }
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
         if (evt instanceof IdleStateEvent) {
-            LOGGER.info(formatChannel(ctx.channel()) + " timed out");
+            LOGGER.info("[{}] timed out", NetworkUtil.session(ctx.channel()));
             closeChannel(ctx.channel());
         }
     }