From f84e2710e05660822633ec9e61cde44c03a42d7e Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sat, 30 Mar 2024 18:46:37 -0700 Subject: Refactor position and event handlers --- src/main/java/org/traccar/BaseDataHandler.java | 38 ---- src/main/java/org/traccar/BasePipelineFactory.java | 97 +++------- .../java/org/traccar/ExtendedObjectDecoder.java | 2 +- src/main/java/org/traccar/MainEventHandler.java | 194 -------------------- .../org/traccar/PositionForwardingHandler.java | 141 --------------- src/main/java/org/traccar/ProcessingHandler.java | 145 +++++++++++++++ .../traccar/handler/AcknowledgementHandler.java | 121 ------------- .../org/traccar/handler/BasePositionHandler.java | 27 +++ .../traccar/handler/ComputedAttributesHandler.java | 40 ++--- .../org/traccar/handler/CopyAttributesHandler.java | 16 +- .../java/org/traccar/handler/DatabaseHandler.java | 49 ++++++ .../org/traccar/handler/DefaultDataHandler.java | 55 ------ .../java/org/traccar/handler/DistanceHandler.java | 13 +- .../org/traccar/handler/EngineHoursHandler.java | 16 +- .../java/org/traccar/handler/FilterHandler.java | 25 +-- .../java/org/traccar/handler/GeocoderHandler.java | 65 ++++--- .../java/org/traccar/handler/GeofenceHandler.java | 15 +- .../org/traccar/handler/GeolocationHandler.java | 73 ++++---- .../org/traccar/handler/HemisphereHandler.java | 16 +- .../java/org/traccar/handler/MotionHandler.java | 16 +- .../traccar/handler/NetworkForwarderHandler.java | 72 -------- .../org/traccar/handler/NetworkMessageHandler.java | 57 ------ .../org/traccar/handler/OpenChannelHandler.java | 42 ----- .../traccar/handler/PositionForwardingHandler.java | 137 +++++++++++++++ .../org/traccar/handler/RemoteAddressHandler.java | 56 ------ .../org/traccar/handler/SpeedLimitHandler.java | 49 ++---- .../traccar/handler/StandardLoggingHandler.java | 89 ---------- src/main/java/org/traccar/handler/TimeHandler.java | 15 +- .../traccar/handler/events/AlertEventHandler.java | 17 +- .../traccar/handler/events/BaseEventHandler.java | 32 +--- .../handler/events/BehaviorEventHandler.java | 18 +- .../handler/events/CommandResultEventHandler.java | 17 +- .../traccar/handler/events/DriverEventHandler.java | 18 +- .../traccar/handler/events/FuelEventHandler.java | 22 +-- .../handler/events/GeofenceEventHandler.java | 21 +-- .../handler/events/IgnitionEventHandler.java | 25 +-- .../handler/events/MaintenanceEventHandler.java | 19 +- .../traccar/handler/events/MediaEventHandler.java | 16 +- .../traccar/handler/events/MotionEventHandler.java | 22 +-- .../handler/events/OverspeedEventHandler.java | 27 ++- .../handler/network/AcknowledgementHandler.java | 121 +++++++++++++ .../traccar/handler/network/MainEventHandler.java | 195 +++++++++++++++++++++ .../handler/network/NetworkForwarderHandler.java | 72 ++++++++ .../handler/network/NetworkMessageHandler.java | 57 ++++++ .../handler/network/OpenChannelHandler.java | 42 +++++ .../handler/network/RemoteAddressHandler.java | 54 ++++++ .../handler/network/StandardLoggingHandler.java | 89 ++++++++++ .../org/traccar/handler/DistanceHandlerTest.java | 5 +- .../org/traccar/handler/MotionHandlerTest.java | 3 +- .../handler/events/AlertEventHandlerTest.java | 12 +- .../events/CommandResultEventHandlerTest.java | 10 +- .../handler/events/IgnitionEventHandlerTest.java | 8 +- .../events/MaintenanceEventHandlerTest.java | 25 ++- 53 files changed, 1257 insertions(+), 1371 deletions(-) delete mode 100644 src/main/java/org/traccar/BaseDataHandler.java delete mode 100644 src/main/java/org/traccar/MainEventHandler.java delete mode 100644 src/main/java/org/traccar/PositionForwardingHandler.java create mode 100644 src/main/java/org/traccar/ProcessingHandler.java delete mode 100644 src/main/java/org/traccar/handler/AcknowledgementHandler.java create mode 100644 src/main/java/org/traccar/handler/BasePositionHandler.java create mode 100644 src/main/java/org/traccar/handler/DatabaseHandler.java delete mode 100644 src/main/java/org/traccar/handler/DefaultDataHandler.java delete mode 100644 src/main/java/org/traccar/handler/NetworkForwarderHandler.java delete mode 100644 src/main/java/org/traccar/handler/NetworkMessageHandler.java delete mode 100644 src/main/java/org/traccar/handler/OpenChannelHandler.java create mode 100644 src/main/java/org/traccar/handler/PositionForwardingHandler.java delete mode 100644 src/main/java/org/traccar/handler/RemoteAddressHandler.java delete mode 100644 src/main/java/org/traccar/handler/StandardLoggingHandler.java create mode 100644 src/main/java/org/traccar/handler/network/AcknowledgementHandler.java create mode 100644 src/main/java/org/traccar/handler/network/MainEventHandler.java create mode 100644 src/main/java/org/traccar/handler/network/NetworkForwarderHandler.java create mode 100644 src/main/java/org/traccar/handler/network/NetworkMessageHandler.java create mode 100644 src/main/java/org/traccar/handler/network/OpenChannelHandler.java create mode 100644 src/main/java/org/traccar/handler/network/RemoteAddressHandler.java create mode 100644 src/main/java/org/traccar/handler/network/StandardLoggingHandler.java (limited to 'src') diff --git a/src/main/java/org/traccar/BaseDataHandler.java b/src/main/java/org/traccar/BaseDataHandler.java deleted file mode 100644 index 48794b0d7..000000000 --- a/src/main/java/org/traccar/BaseDataHandler.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 2015 - 2018 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar; - -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import org.traccar.model.Position; - -public abstract class BaseDataHandler extends ChannelInboundHandlerAdapter { - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (msg instanceof Position) { - Position position = handlePosition((Position) msg); - if (position != null) { - ctx.fireChannelRead(position); - } - } else { - super.channelRead(ctx, msg); - } - } - - protected abstract Position handlePosition(Position position); - -} diff --git a/src/main/java/org/traccar/BasePipelineFactory.java b/src/main/java/org/traccar/BasePipelineFactory.java index ca4a4ae63..6d9431ad6 100644 --- a/src/main/java/org/traccar/BasePipelineFactory.java +++ b/src/main/java/org/traccar/BasePipelineFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2012 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2012 - 2024 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. @@ -25,36 +25,13 @@ import io.netty.channel.ChannelPipeline; import io.netty.handler.timeout.IdleStateHandler; import org.traccar.config.Config; import org.traccar.config.Keys; -import org.traccar.handler.AcknowledgementHandler; -import org.traccar.handler.ComputedAttributesHandler; -import org.traccar.handler.CopyAttributesHandler; -import org.traccar.handler.DefaultDataHandler; -import org.traccar.handler.DistanceHandler; -import org.traccar.handler.EngineHoursHandler; -import org.traccar.handler.FilterHandler; -import org.traccar.handler.GeocoderHandler; -import org.traccar.handler.GeofenceHandler; -import org.traccar.handler.GeolocationHandler; -import org.traccar.handler.HemisphereHandler; -import org.traccar.handler.MotionHandler; -import org.traccar.handler.NetworkForwarderHandler; -import org.traccar.handler.NetworkMessageHandler; -import org.traccar.handler.OpenChannelHandler; -import org.traccar.handler.RemoteAddressHandler; -import org.traccar.handler.SpeedLimitHandler; -import org.traccar.handler.StandardLoggingHandler; -import org.traccar.handler.TimeHandler; -import org.traccar.handler.events.AlertEventHandler; -import org.traccar.handler.events.BehaviorEventHandler; -import org.traccar.handler.events.CommandResultEventHandler; -import org.traccar.handler.events.DriverEventHandler; -import org.traccar.handler.events.FuelEventHandler; -import org.traccar.handler.events.GeofenceEventHandler; -import org.traccar.handler.events.IgnitionEventHandler; -import org.traccar.handler.events.MaintenanceEventHandler; -import org.traccar.handler.events.MediaEventHandler; -import org.traccar.handler.events.MotionEventHandler; -import org.traccar.handler.events.OverspeedEventHandler; +import org.traccar.handler.network.AcknowledgementHandler; +import org.traccar.handler.network.MainEventHandler; +import org.traccar.handler.network.NetworkForwarderHandler; +import org.traccar.handler.network.NetworkMessageHandler; +import org.traccar.handler.network.OpenChannelHandler; +import org.traccar.handler.network.RemoteAddressHandler; +import org.traccar.handler.network.StandardLoggingHandler; import java.util.Map; @@ -83,15 +60,7 @@ public abstract class BasePipelineFactory extends ChannelInitializer { protected abstract void addProtocolHandlers(PipelineBuilder pipeline); - @SafeVarargs - private void addHandlers(ChannelPipeline pipeline, Class... handlerClasses) { - for (Class handlerClass : handlerClasses) { - if (handlerClass != null) { - pipeline.addLast(injector.getInstance(handlerClass)); - } - } - } - + @SuppressWarnings("unchecked") public static T getHandler(ChannelPipeline pipeline, Class clazz) { for (Map.Entry handlerEntry : pipeline) { ChannelHandler handler = handlerEntry.getValue(); @@ -107,6 +76,11 @@ public abstract class BasePipelineFactory extends ChannelInitializer { return null; } + private T injectMembers(T object) { + injector.injectMembers(object); + return object; + } + @Override protected void initChannel(Channel channel) { final ChannelPipeline pipeline = channel.pipeline(); @@ -119,15 +93,11 @@ public abstract class BasePipelineFactory extends ChannelInitializer { pipeline.addLast(new OpenChannelHandler(connector)); if (config.hasKey(Keys.SERVER_FORWARD)) { int port = config.getInteger(Keys.PROTOCOL_PORT.withPrefix(protocol)); - var handler = new NetworkForwarderHandler(port); - injector.injectMembers(handler); - pipeline.addLast(handler); + pipeline.addLast(injectMembers(new NetworkForwarderHandler(port))); } pipeline.addLast(new NetworkMessageHandler()); - - var loggingHandler = new StandardLoggingHandler(protocol); - injector.injectMembers(loggingHandler); - pipeline.addLast(loggingHandler); + pipeline.addLast(new RemoteAddressHandler(config)); + pipeline.addLast(injectMembers(new StandardLoggingHandler(protocol))); if (!connector.isDatagram() && !config.getBoolean(Keys.SERVER_INSTANT_ACKNOWLEDGEMENT)) { pipeline.addLast(new AcknowledgementHandler()); @@ -135,7 +105,7 @@ public abstract class BasePipelineFactory extends ChannelInitializer { addProtocolHandlers(handler -> { if (handler instanceof BaseProtocolDecoder || handler instanceof BaseProtocolEncoder) { - injector.injectMembers(handler); + injectMembers(handler); } else { if (handler instanceof ChannelInboundHandler) { handler = new WrapperInboundHandler((ChannelInboundHandler) handler); @@ -146,35 +116,8 @@ public abstract class BasePipelineFactory extends ChannelInitializer { pipeline.addLast(handler); }); - addHandlers( - pipeline, - TimeHandler.class, - GeolocationHandler.class, - HemisphereHandler.class, - DistanceHandler.class, - RemoteAddressHandler.class, - FilterHandler.class, - GeofenceHandler.class, - GeocoderHandler.class, - SpeedLimitHandler.class, - MotionHandler.class, - CopyAttributesHandler.class, - EngineHoursHandler.class, - ComputedAttributesHandler.class, - PositionForwardingHandler.class, - DefaultDataHandler.class, - MediaEventHandler.class, - CommandResultEventHandler.class, - OverspeedEventHandler.class, - BehaviorEventHandler.class, - FuelEventHandler.class, - MotionEventHandler.class, - GeofenceEventHandler.class, - AlertEventHandler.class, - IgnitionEventHandler.class, - MaintenanceEventHandler.class, - DriverEventHandler.class, - MainEventHandler.class); + pipeline.addLast(injector.getInstance(ProcessingHandler.class)); + pipeline.addLast(injector.getInstance(MainEventHandler.class)); } } diff --git a/src/main/java/org/traccar/ExtendedObjectDecoder.java b/src/main/java/org/traccar/ExtendedObjectDecoder.java index cddddcd80..9468e2fff 100644 --- a/src/main/java/org/traccar/ExtendedObjectDecoder.java +++ b/src/main/java/org/traccar/ExtendedObjectDecoder.java @@ -23,7 +23,7 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.util.ReferenceCountUtil; import org.traccar.config.Config; import org.traccar.config.Keys; -import org.traccar.handler.AcknowledgementHandler; +import org.traccar.handler.network.AcknowledgementHandler; import org.traccar.helper.DataConverter; import org.traccar.model.Position; diff --git a/src/main/java/org/traccar/MainEventHandler.java b/src/main/java/org/traccar/MainEventHandler.java deleted file mode 100644 index fb0171d63..000000000 --- a/src/main/java/org/traccar/MainEventHandler.java +++ /dev/null @@ -1,194 +0,0 @@ -/* - * 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -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; -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); - - private final Set connectionlessProtocols = new HashSet<>(); - private final Set logAttributes = new LinkedHashSet<>(); - - 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(config.getString(Keys.LOGGER_ATTRIBUTES).split("[, ]"))); - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { - if (msg instanceof Position) { - - Position position = (Position) msg; - Device device = cacheManager.getObject(Device.class, position.getDeviceId()); - - try { - 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); - } - - StringBuilder builder = new StringBuilder(); - builder.append("[").append(NetworkUtil.session(ctx.channel())).append("] "); - builder.append("id: ").append(device.getUniqueId()); - for (String attribute : logAttributes) { - switch (attribute) { - case "time": - builder.append(", time: ").append(DateUtil.formatDate(position.getFixTime(), false)); - break; - case "position": - builder.append(", lat: ").append(String.format("%.5f", position.getLatitude())); - builder.append(", lon: ").append(String.format("%.5f", position.getLongitude())); - break; - case "speed": - if (position.getSpeed() > 0) { - builder.append(", speed: ").append(String.format("%.1f", position.getSpeed())); - } - break; - case "course": - builder.append(", course: ").append(String.format("%.1f", position.getCourse())); - break; - case "accuracy": - if (position.getAccuracy() > 0) { - builder.append(", accuracy: ").append(String.format("%.1f", position.getAccuracy())); - } - break; - case "outdated": - if (position.getOutdated()) { - builder.append(", outdated"); - } - break; - case "invalid": - if (!position.getValid()) { - builder.append(", invalid"); - } - break; - default: - Object value = position.getAttributes().get(attribute); - if (value != null) { - builder.append(", ").append(attribute).append(": ").append(value); - } - break; - } - } - LOGGER.info(builder.toString()); - - statisticsManager.registerMessageStored(position.getDeviceId(), position.getProtocol()); - - ctx.writeAndFlush(new AcknowledgementHandler.EventHandled(position)); - } - } - - @Override - public void channelActive(ChannelHandlerContext ctx) { - if (!(ctx.channel() instanceof DatagramChannel)) { - LOGGER.info("[{}] connected", NetworkUtil.session(ctx.channel())); - } - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) { - LOGGER.info("[{}] disconnected", NetworkUtil.session(ctx.channel())); - closeChannel(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 - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - while (cause.getCause() != null && cause.getCause() != cause) { - cause = cause.getCause(); - } - 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("[{}] timed out", NetworkUtil.session(ctx.channel())); - closeChannel(ctx.channel()); - } - } - - private void closeChannel(Channel channel) { - if (!(channel instanceof DatagramChannel)) { - channel.close(); - } - } - -} diff --git a/src/main/java/org/traccar/PositionForwardingHandler.java b/src/main/java/org/traccar/PositionForwardingHandler.java deleted file mode 100644 index a79b01367..000000000 --- a/src/main/java/org/traccar/PositionForwardingHandler.java +++ /dev/null @@ -1,141 +0,0 @@ -/* - * Copyright 2015 - 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar; - -import io.netty.channel.ChannelHandler; -import io.netty.util.Timeout; -import io.netty.util.Timer; -import io.netty.util.TimerTask; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.traccar.config.Config; -import org.traccar.config.Keys; -import org.traccar.forward.PositionData; -import org.traccar.forward.PositionForwarder; -import org.traccar.forward.ResultHandler; -import org.traccar.model.Device; -import org.traccar.model.Position; -import org.traccar.session.cache.CacheManager; - -import jakarta.annotation.Nullable; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -@Singleton -@ChannelHandler.Sharable -public class PositionForwardingHandler extends BaseDataHandler { - - private static final Logger LOGGER = LoggerFactory.getLogger(PositionForwardingHandler.class); - - private final CacheManager cacheManager; - private final Timer timer; - - private final PositionForwarder positionForwarder; - - private final boolean retryEnabled; - private final int retryDelay; - private final int retryCount; - private final int retryLimit; - - private final AtomicInteger deliveryPending; - - @Inject - public PositionForwardingHandler( - Config config, CacheManager cacheManager, Timer timer, @Nullable PositionForwarder positionForwarder) { - - this.cacheManager = cacheManager; - this.timer = timer; - this.positionForwarder = positionForwarder; - - this.retryEnabled = config.getBoolean(Keys.FORWARD_RETRY_ENABLE); - this.retryDelay = config.getInteger(Keys.FORWARD_RETRY_DELAY); - this.retryCount = config.getInteger(Keys.FORWARD_RETRY_COUNT); - this.retryLimit = config.getInteger(Keys.FORWARD_RETRY_LIMIT); - - this.deliveryPending = new AtomicInteger(); - } - - class AsyncRequestAndCallback implements ResultHandler, TimerTask { - - private final PositionData positionData; - - private int retries = 0; - - AsyncRequestAndCallback(PositionData positionData) { - this.positionData = positionData; - deliveryPending.incrementAndGet(); - } - - private void send() { - positionForwarder.forward(positionData, this); - } - - private void retry(Throwable throwable) { - boolean scheduled = false; - try { - if (retryEnabled && deliveryPending.get() <= retryLimit && retries < retryCount) { - schedule(); - scheduled = true; - } - } finally { - int pending = scheduled ? deliveryPending.get() : deliveryPending.decrementAndGet(); - LOGGER.warn("Position forwarding failed: " + pending + " pending", throwable); - } - } - - private void schedule() { - timer.newTimeout(this, retryDelay * (long) Math.pow(2, retries++), TimeUnit.MILLISECONDS); - } - - @Override - public void onResult(boolean success, Throwable throwable) { - if (success) { - deliveryPending.decrementAndGet(); - } else { - retry(throwable); - } - } - - @Override - public void run(Timeout timeout) { - boolean sent = false; - try { - if (!timeout.isCancelled()) { - send(); - sent = true; - } - } finally { - if (!sent) { - deliveryPending.decrementAndGet(); - } - } - } - } - - @Override - protected Position handlePosition(Position position) { - if (positionForwarder != null) { - PositionData positionData = new PositionData(); - positionData.setPosition(position); - positionData.setDevice(cacheManager.getObject(Device.class, position.getDeviceId())); - new AsyncRequestAndCallback(positionData).send(); - } - return position; - } - -} diff --git a/src/main/java/org/traccar/ProcessingHandler.java b/src/main/java/org/traccar/ProcessingHandler.java new file mode 100644 index 000000000..7627c719b --- /dev/null +++ b/src/main/java/org/traccar/ProcessingHandler.java @@ -0,0 +1,145 @@ +/* + * Copyright 2024 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar; + +import com.google.inject.Injector; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import org.traccar.database.NotificationManager; +import org.traccar.handler.BasePositionHandler; +import org.traccar.handler.ComputedAttributesHandler; +import org.traccar.handler.CopyAttributesHandler; +import org.traccar.handler.DatabaseHandler; +import org.traccar.handler.DistanceHandler; +import org.traccar.handler.EngineHoursHandler; +import org.traccar.handler.FilterHandler; +import org.traccar.handler.GeocoderHandler; +import org.traccar.handler.GeofenceHandler; +import org.traccar.handler.GeolocationHandler; +import org.traccar.handler.HemisphereHandler; +import org.traccar.handler.MotionHandler; +import org.traccar.handler.PositionForwardingHandler; +import org.traccar.handler.SpeedLimitHandler; +import org.traccar.handler.TimeHandler; +import org.traccar.handler.events.AlertEventHandler; +import org.traccar.handler.events.BaseEventHandler; +import org.traccar.handler.events.BehaviorEventHandler; +import org.traccar.handler.events.CommandResultEventHandler; +import org.traccar.handler.events.DriverEventHandler; +import org.traccar.handler.events.FuelEventHandler; +import org.traccar.handler.events.GeofenceEventHandler; +import org.traccar.handler.events.IgnitionEventHandler; +import org.traccar.handler.events.MaintenanceEventHandler; +import org.traccar.handler.events.MediaEventHandler; +import org.traccar.handler.events.MotionEventHandler; +import org.traccar.handler.events.OverspeedEventHandler; +import org.traccar.handler.network.AcknowledgementHandler; +import org.traccar.model.Position; + +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +@Singleton +@ChannelHandler.Sharable +public class ProcessingHandler extends ChannelInboundHandlerAdapter { + + private final NotificationManager notificationManager; + private final List positionHandlers; + private final List eventHandlers; + + @Inject + public ProcessingHandler(Injector injector, NotificationManager notificationManager) { + this.notificationManager = notificationManager; + + positionHandlers = Stream.of( + TimeHandler.class, + GeolocationHandler.class, + HemisphereHandler.class, + DistanceHandler.class, + FilterHandler.class, + GeofenceHandler.class, + GeocoderHandler.class, + SpeedLimitHandler.class, + MotionHandler.class, + EngineHoursHandler.class, + ComputedAttributesHandler.class, + CopyAttributesHandler.class, + PositionForwardingHandler.class, + DatabaseHandler.class) + .map((clazz) -> (BasePositionHandler) injector.getInstance(clazz)) + .filter(Objects::nonNull) + .collect(Collectors.toUnmodifiableList()); + + eventHandlers = Stream.of( + MediaEventHandler.class, + CommandResultEventHandler.class, + OverspeedEventHandler.class, + BehaviorEventHandler.class, + FuelEventHandler.class, + MotionEventHandler.class, + GeofenceEventHandler.class, + AlertEventHandler.class, + IgnitionEventHandler.class, + MaintenanceEventHandler.class, + DriverEventHandler.class) + .map((clazz) -> (BaseEventHandler) injector.getInstance(clazz)) + .filter(Objects::nonNull) + .collect(Collectors.toUnmodifiableList()); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + if (msg instanceof Position) { + processPositionHandlers(ctx, (Position) msg); + } + super.channelRead(ctx, msg); + } + + private void processPositionHandlers(ChannelHandlerContext ctx, Position position) { + var iterator = positionHandlers.iterator(); + iterator.next().handlePosition(position, new BasePositionHandler.Callback() { + @Override + public void processed(Position position) { + if (position != null) { + if (iterator.hasNext()) { + iterator.next().handlePosition(position, this); + } else { + processEventHandlers(ctx, position); + } + } else { + finishedProcessing(ctx, null); + } + } + }); + } + + private void processEventHandlers(ChannelHandlerContext ctx, Position position) { + eventHandlers.forEach(handler -> handler.analyzePosition( + position, (event) -> notificationManager.updateEvents(Map.of(event, position)))); + finishedProcessing(ctx, position); + } + + private void finishedProcessing(ChannelHandlerContext ctx, Position position) { + ctx.writeAndFlush(new AcknowledgementHandler.EventHandled(position)); + } + +} diff --git a/src/main/java/org/traccar/handler/AcknowledgementHandler.java b/src/main/java/org/traccar/handler/AcknowledgementHandler.java deleted file mode 100644 index 4c1085998..000000000 --- a/src/main/java/org/traccar/handler/AcknowledgementHandler.java +++ /dev/null @@ -1,121 +0,0 @@ -/* - * Copyright 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelOutboundHandlerAdapter; -import io.netty.channel.ChannelPromise; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Collection; -import java.util.HashSet; -import java.util.LinkedList; -import java.util.List; -import java.util.Set; - -public class AcknowledgementHandler extends ChannelOutboundHandlerAdapter { - - private static final Logger LOGGER = LoggerFactory.getLogger(AcknowledgementHandler.class); - - public interface Event { - } - - public static class EventReceived implements Event { - } - - public static class EventDecoded implements Event { - private final Collection objects; - - public EventDecoded(Collection objects) { - this.objects = objects; - } - - public Collection getObjects() { - return objects; - } - } - - public static class EventHandled implements Event { - private final Object object; - - public EventHandled(Object object) { - this.object = object; - } - - public Object getObject() { - return object; - } - } - - private static final class Entry { - private final Object message; - private final ChannelPromise promise; - - private Entry(Object message, ChannelPromise promise) { - this.message = message; - this.promise = promise; - } - - public Object getMessage() { - return message; - } - - public ChannelPromise getPromise() { - return promise; - } - } - - private List queue; - private final Set waiting = new HashSet<>(); - - @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { - List output = new LinkedList<>(); - synchronized (this) { - if (msg instanceof Event) { - if (msg instanceof EventReceived) { - LOGGER.debug("Event received"); - if (queue == null) { - queue = new LinkedList<>(); - } - } else if (msg instanceof EventDecoded) { - EventDecoded event = (EventDecoded) msg; - LOGGER.debug("Event decoded {}", event.getObjects().size()); - waiting.addAll(event.getObjects()); - } else if (msg instanceof EventHandled) { - EventHandled event = (EventHandled) msg; - LOGGER.debug("Event handled"); - waiting.remove(event.getObject()); - } - if (!(msg instanceof EventReceived) && waiting.isEmpty()) { - output.addAll(queue); - queue = null; - } - } else if (queue != null) { - LOGGER.debug("Message queued"); - queue.add(new Entry(msg, promise)); - } else { - LOGGER.debug("Message sent"); - output.add(new Entry(msg, promise)); - } - } - for (Entry entry : output) { - ctx.write(entry.getMessage(), entry.getPromise()); - } - } - -} diff --git a/src/main/java/org/traccar/handler/BasePositionHandler.java b/src/main/java/org/traccar/handler/BasePositionHandler.java new file mode 100644 index 000000000..2fee5c652 --- /dev/null +++ b/src/main/java/org/traccar/handler/BasePositionHandler.java @@ -0,0 +1,27 @@ +/* + * Copyright 2015 - 2024 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler; + +import org.traccar.model.Position; + +public abstract class BasePositionHandler { + + public interface Callback { + void processed(Position position); + } + + public abstract void handlePosition(Position position, Callback callback); +} diff --git a/src/main/java/org/traccar/handler/ComputedAttributesHandler.java b/src/main/java/org/traccar/handler/ComputedAttributesHandler.java index 8b010ceae..8d6fb39c3 100644 --- a/src/main/java/org/traccar/handler/ComputedAttributesHandler.java +++ b/src/main/java/org/traccar/handler/ComputedAttributesHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2017 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2017 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,26 +16,15 @@ */ package org.traccar.handler; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; -import java.util.List; - -import io.netty.channel.ChannelHandler; -import org.apache.commons.jexl3.JexlFeatures; -import org.apache.commons.jexl3.JexlEngine; +import jakarta.inject.Inject; import org.apache.commons.jexl3.JexlBuilder; -import org.apache.commons.jexl3.introspection.JexlSandbox; +import org.apache.commons.jexl3.JexlEngine; import org.apache.commons.jexl3.JexlException; +import org.apache.commons.jexl3.JexlFeatures; import org.apache.commons.jexl3.MapContext; +import org.apache.commons.jexl3.introspection.JexlSandbox; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.traccar.BaseDataHandler; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.model.Attribute; @@ -43,12 +32,17 @@ import org.traccar.model.Device; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; -@Singleton -@ChannelHandler.Sharable -public class ComputedAttributesHandler extends BaseDataHandler { +public class ComputedAttributesHandler extends BasePositionHandler { private static final Logger LOGGER = LoggerFactory.getLogger(ComputedAttributesHandler.class); @@ -144,7 +138,7 @@ public class ComputedAttributesHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { Collection attributes = cacheManager.getDeviceObjects(position.getDeviceId(), Attribute.class); for (Attribute attribute : attributes) { if (attribute.getAttribute() != null) { @@ -202,7 +196,7 @@ public class ComputedAttributesHandler extends BaseDataHandler { } } } - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/CopyAttributesHandler.java b/src/main/java/org/traccar/handler/CopyAttributesHandler.java index 42b438e41..9c31bf56e 100644 --- a/src/main/java/org/traccar/handler/CopyAttributesHandler.java +++ b/src/main/java/org/traccar/handler/CopyAttributesHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2016 - 2017 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,20 +16,14 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import org.traccar.BaseDataHandler; +import jakarta.inject.Inject; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.helper.model.AttributeUtil; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable -public class CopyAttributesHandler extends BaseDataHandler { +public class CopyAttributesHandler extends BasePositionHandler { private final boolean enabled; private final CacheManager cacheManager; @@ -41,7 +35,7 @@ public class CopyAttributesHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { if (enabled) { String attributesString = AttributeUtil.lookup( cacheManager, Keys.PROCESSING_COPY_ATTRIBUTES, position.getDeviceId()); @@ -54,7 +48,7 @@ public class CopyAttributesHandler extends BaseDataHandler { } } } - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/DatabaseHandler.java b/src/main/java/org/traccar/handler/DatabaseHandler.java new file mode 100644 index 000000000..b1f218a1e --- /dev/null +++ b/src/main/java/org/traccar/handler/DatabaseHandler.java @@ -0,0 +1,49 @@ +/* + * Copyright 2015 - 2024 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler; + +import jakarta.inject.Inject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.model.Position; +import org.traccar.storage.Storage; +import org.traccar.storage.query.Columns; +import org.traccar.storage.query.Request; + +public class DatabaseHandler extends BasePositionHandler { + + private static final Logger LOGGER = LoggerFactory.getLogger(DatabaseHandler.class); + + private final Storage storage; + + @Inject + public DatabaseHandler(Storage storage) { + this.storage = storage; + } + + @Override + public void handlePosition(Position position, Callback callback) { + + try { + position.setId(storage.addObject(position, new Request(new Columns.Exclude("id")))); + } catch (Exception error) { + LOGGER.warn("Failed to store position", error); + } + + callback.processed(position); + } + +} diff --git a/src/main/java/org/traccar/handler/DefaultDataHandler.java b/src/main/java/org/traccar/handler/DefaultDataHandler.java deleted file mode 100644 index cca6dcd0a..000000000 --- a/src/main/java/org/traccar/handler/DefaultDataHandler.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright 2015 - 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.channel.ChannelHandler; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.traccar.BaseDataHandler; -import org.traccar.model.Position; -import org.traccar.storage.Storage; -import org.traccar.storage.query.Columns; -import org.traccar.storage.query.Request; - -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable -public class DefaultDataHandler extends BaseDataHandler { - - private static final Logger LOGGER = LoggerFactory.getLogger(DefaultDataHandler.class); - - private final Storage storage; - - @Inject - public DefaultDataHandler(Storage storage) { - this.storage = storage; - } - - @Override - protected Position handlePosition(Position position) { - - try { - position.setId(storage.addObject(position, new Request(new Columns.Exclude("id")))); - } catch (Exception error) { - LOGGER.warn("Failed to store position", error); - } - - return position; - } - -} diff --git a/src/main/java/org/traccar/handler/DistanceHandler.java b/src/main/java/org/traccar/handler/DistanceHandler.java index db8c73779..ee5d64894 100644 --- a/src/main/java/org/traccar/handler/DistanceHandler.java +++ b/src/main/java/org/traccar/handler/DistanceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2015 Amila Silva * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,19 +16,14 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import org.traccar.BaseDataHandler; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.helper.DistanceCalculator; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -@Singleton -@ChannelHandler.Sharable -public class DistanceHandler extends BaseDataHandler { +public class DistanceHandler extends BasePositionHandler { private final CacheManager cacheManager; @@ -45,7 +40,7 @@ public class DistanceHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { double distance = 0.0; if (position.hasAttribute(Position.KEY_DISTANCE)) { @@ -76,7 +71,7 @@ public class DistanceHandler extends BaseDataHandler { position.set(Position.KEY_DISTANCE, distance); position.set(Position.KEY_TOTAL_DISTANCE, totalDistance + distance); - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/EngineHoursHandler.java b/src/main/java/org/traccar/handler/EngineHoursHandler.java index 621205b34..ed5f9b509 100644 --- a/src/main/java/org/traccar/handler/EngineHoursHandler.java +++ b/src/main/java/org/traccar/handler/EngineHoursHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2018 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2018 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,17 +16,11 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import org.traccar.BaseDataHandler; +import jakarta.inject.Inject; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable -public class EngineHoursHandler extends BaseDataHandler { +public class EngineHoursHandler extends BasePositionHandler { private final CacheManager cacheManager; @@ -36,7 +30,7 @@ public class EngineHoursHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { if (!position.hasAttribute(Position.KEY_HOURS)) { Position last = cacheManager.getPosition(position.getDeviceId()); if (last != null) { @@ -49,7 +43,7 @@ public class EngineHoursHandler extends BaseDataHandler { } } } - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/FilterHandler.java b/src/main/java/org/traccar/handler/FilterHandler.java index a15d3ffad..4cc6233d0 100644 --- a/src/main/java/org/traccar/handler/FilterHandler.java +++ b/src/main/java/org/traccar/handler/FilterHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2014 - 2024 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,9 +15,7 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; +import jakarta.inject.Inject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Config; @@ -36,13 +34,9 @@ import org.traccar.storage.query.Condition; import org.traccar.storage.query.Order; import org.traccar.storage.query.Request; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; import java.util.Date; -@Singleton -@ChannelHandler.Sharable -public class FilterHandler extends ChannelInboundHandlerAdapter { +public class FilterHandler extends BasePositionHandler { private static final Logger LOGGER = LoggerFactory.getLogger(FilterHandler.class); @@ -277,16 +271,11 @@ public class FilterHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (msg instanceof Position) { - Position position = (Position) msg; - if (enabled && filter(position)) { - ctx.writeAndFlush(new AcknowledgementHandler.EventHandled(position)); - } else { - ctx.fireChannelRead(position); - } + public void handlePosition(Position position, Callback callback) { + if (enabled && filter(position)) { + callback.processed(null); } else { - super.channelRead(ctx, msg); + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/GeocoderHandler.java b/src/main/java/org/traccar/handler/GeocoderHandler.java index e4f240a90..c62bcb6f8 100644 --- a/src/main/java/org/traccar/handler/GeocoderHandler.java +++ b/src/main/java/org/traccar/handler/GeocoderHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2012 - 2021 Anton Tananaev (anton@traccar.org) + * Copyright 2012 - 2024 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,9 +15,6 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Config; @@ -26,8 +23,7 @@ import org.traccar.geocoder.Geocoder; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -@ChannelHandler.Sharable -public class GeocoderHandler extends ChannelInboundHandlerAdapter { +public class GeocoderHandler extends BasePositionHandler { private static final Logger LOGGER = LoggerFactory.getLogger(GeocoderHandler.class); @@ -46,39 +42,38 @@ public class GeocoderHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(final ChannelHandlerContext ctx, Object message) { - if (message instanceof Position && !ignorePositions) { - final Position position = (Position) message; - if (processInvalidPositions || position.getValid()) { - if (reuseDistance != 0) { - Position lastPosition = cacheManager.getPosition(position.getDeviceId()); - if (lastPosition != null && lastPosition.getAddress() != null - && position.getDouble(Position.KEY_DISTANCE) <= reuseDistance) { - position.setAddress(lastPosition.getAddress()); - ctx.fireChannelRead(position); - return; - } + public void handlePosition(Position position, Callback callback) { + if (!ignorePositions) { + callback.processed(position); + } + + if (processInvalidPositions || position.getValid()) { + if (reuseDistance != 0) { + Position lastPosition = cacheManager.getPosition(position.getDeviceId()); + if (lastPosition != null && lastPosition.getAddress() != null + && position.getDouble(Position.KEY_DISTANCE) <= reuseDistance) { + position.setAddress(lastPosition.getAddress()); + callback.processed(position); + return; } + } - geocoder.getAddress(position.getLatitude(), position.getLongitude(), - new Geocoder.ReverseGeocoderCallback() { - @Override - public void onSuccess(String address) { - position.setAddress(address); - ctx.fireChannelRead(position); - } + geocoder.getAddress(position.getLatitude(), position.getLongitude(), + new Geocoder.ReverseGeocoderCallback() { + @Override + public void onSuccess(String address) { + position.setAddress(address); + callback.processed(position); + } - @Override - public void onFailure(Throwable e) { - LOGGER.warn("Geocoding failed", e); - ctx.fireChannelRead(position); - } - }); - } else { - ctx.fireChannelRead(position); - } + @Override + public void onFailure(Throwable e) { + LOGGER.warn("Geocoding failed", e); + callback.processed(position); + } + }); } else { - ctx.fireChannelRead(message); + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/GeofenceHandler.java b/src/main/java/org/traccar/handler/GeofenceHandler.java index 68bc6dbf0..33b46f058 100644 --- a/src/main/java/org/traccar/handler/GeofenceHandler.java +++ b/src/main/java/org/traccar/handler/GeofenceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2023 - 2024 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,20 +15,15 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import org.traccar.BaseDataHandler; +import jakarta.inject.Inject; import org.traccar.config.Config; import org.traccar.helper.model.GeofenceUtil; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; import java.util.List; -@Singleton -@ChannelHandler.Sharable -public class GeofenceHandler extends BaseDataHandler { +public class GeofenceHandler extends BasePositionHandler { private final Config config; private final CacheManager cacheManager; @@ -40,13 +35,13 @@ public class GeofenceHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { List geofenceIds = GeofenceUtil.getCurrentGeofences(config, cacheManager, position); if (!geofenceIds.isEmpty()) { position.setGeofenceIds(geofenceIds); } - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/GeolocationHandler.java b/src/main/java/org/traccar/handler/GeolocationHandler.java index a54ea03e3..cb9c04808 100644 --- a/src/main/java/org/traccar/handler/GeolocationHandler.java +++ b/src/main/java/org/traccar/handler/GeolocationHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2015 - 2024 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,9 +15,6 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Config; @@ -27,8 +24,7 @@ import org.traccar.geolocation.GeolocationProvider; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -@ChannelHandler.Sharable -public class GeolocationHandler extends ChannelInboundHandlerAdapter { +public class GeolocationHandler extends BasePositionHandler { private static final Logger LOGGER = LoggerFactory.getLogger(GeolocationHandler.class); @@ -51,46 +47,41 @@ public class GeolocationHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(final ChannelHandlerContext ctx, Object message) { - if (message instanceof Position) { - final Position position = (Position) message; - if ((position.getOutdated() || processInvalidPositions && !position.getValid()) - && position.getNetwork() != null - && (!requireWifi || position.getNetwork().getWifiAccessPoints() != null)) { - if (reuse) { - Position lastPosition = cacheManager.getPosition(position.getDeviceId()); - if (lastPosition != null && position.getNetwork().equals(lastPosition.getNetwork())) { - updatePosition( - position, lastPosition.getLatitude(), lastPosition.getLongitude(), - lastPosition.getAccuracy()); - ctx.fireChannelRead(position); - return; - } + public void handlePosition(Position position, Callback callback) { + if ((position.getOutdated() || processInvalidPositions && !position.getValid()) + && position.getNetwork() != null + && (!requireWifi || position.getNetwork().getWifiAccessPoints() != null)) { + if (reuse) { + Position lastPosition = cacheManager.getPosition(position.getDeviceId()); + if (lastPosition != null && position.getNetwork().equals(lastPosition.getNetwork())) { + updatePosition( + position, lastPosition.getLatitude(), lastPosition.getLongitude(), + lastPosition.getAccuracy()); + callback.processed(position); + return; } + } - if (statisticsManager != null) { - statisticsManager.registerGeolocationRequest(); - } + if (statisticsManager != null) { + statisticsManager.registerGeolocationRequest(); + } - geolocationProvider.getLocation(position.getNetwork(), - new GeolocationProvider.LocationProviderCallback() { - @Override - public void onSuccess(double latitude, double longitude, double accuracy) { - updatePosition(position, latitude, longitude, accuracy); - ctx.fireChannelRead(position); - } + geolocationProvider.getLocation(position.getNetwork(), + new GeolocationProvider.LocationProviderCallback() { + @Override + public void onSuccess(double latitude, double longitude, double accuracy) { + updatePosition(position, latitude, longitude, accuracy); + callback.processed(position); + } - @Override - public void onFailure(Throwable e) { - LOGGER.warn("Geolocation network error", e); - ctx.fireChannelRead(position); - } - }); - } else { - ctx.fireChannelRead(position); - } + @Override + public void onFailure(Throwable e) { + LOGGER.warn("Geolocation network error", e); + callback.processed(position); + } + }); } else { - ctx.fireChannelRead(message); + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/HemisphereHandler.java b/src/main/java/org/traccar/handler/HemisphereHandler.java index 294e449db..6b64177e4 100644 --- a/src/main/java/org/traccar/handler/HemisphereHandler.java +++ b/src/main/java/org/traccar/handler/HemisphereHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 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,18 +15,12 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import org.traccar.BaseDataHandler; +import jakarta.inject.Inject; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.model.Position; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable -public class HemisphereHandler extends BaseDataHandler { +public class HemisphereHandler extends BasePositionHandler { private int latitudeFactor; private int longitudeFactor; @@ -52,14 +46,14 @@ public class HemisphereHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { if (latitudeFactor != 0) { position.setLatitude(Math.abs(position.getLatitude()) * latitudeFactor); } if (longitudeFactor != 0) { position.setLongitude(Math.abs(position.getLongitude()) * longitudeFactor); } - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/MotionHandler.java b/src/main/java/org/traccar/handler/MotionHandler.java index 68a31a16a..bb7ff2a65 100644 --- a/src/main/java/org/traccar/handler/MotionHandler.java +++ b/src/main/java/org/traccar/handler/MotionHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2017 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2017 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,19 +16,13 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import org.traccar.BaseDataHandler; +import jakarta.inject.Inject; import org.traccar.config.Keys; import org.traccar.helper.model.AttributeUtil; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable -public class MotionHandler extends BaseDataHandler { +public class MotionHandler extends BasePositionHandler { private final CacheManager cacheManager; @@ -38,13 +32,13 @@ public class MotionHandler extends BaseDataHandler { } @Override - protected Position handlePosition(Position position) { + public void handlePosition(Position position, Callback callback) { if (!position.hasAttribute(Position.KEY_MOTION)) { double threshold = AttributeUtil.lookup( cacheManager, Keys.EVENT_MOTION_SPEED_THRESHOLD, position.getDeviceId()); position.set(Position.KEY_MOTION, position.getSpeed() > threshold); } - return position; + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/NetworkForwarderHandler.java b/src/main/java/org/traccar/handler/NetworkForwarderHandler.java deleted file mode 100644 index 470e175ca..000000000 --- a/src/main/java/org/traccar/handler/NetworkForwarderHandler.java +++ /dev/null @@ -1,72 +0,0 @@ -/* - * Copyright 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.channel.socket.DatagramChannel; -import io.netty.channel.socket.DatagramPacket; -import org.traccar.forward.NetworkForwarder; - -import jakarta.inject.Inject; -import java.net.InetSocketAddress; -import java.net.SocketAddress; - -public class NetworkForwarderHandler extends ChannelInboundHandlerAdapter { - - private final int port; - - private NetworkForwarder networkForwarder; - - public NetworkForwarderHandler(int port) { - this.port = port; - } - - @Inject - public void setNetworkForwarder(NetworkForwarder networkForwarder) { - this.networkForwarder = networkForwarder; - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - boolean datagram = ctx.channel() instanceof DatagramChannel; - SocketAddress remoteAddress; - ByteBuf buffer; - if (datagram) { - DatagramPacket message = (DatagramPacket) msg; - remoteAddress = message.recipient(); - buffer = message.content(); - } else { - remoteAddress = ctx.channel().remoteAddress(); - buffer = (ByteBuf) msg; - } - - byte[] data = new byte[buffer.readableBytes()]; - buffer.getBytes(buffer.readerIndex(), data); - networkForwarder.forward((InetSocketAddress) remoteAddress, port, datagram, data); - super.channelRead(ctx, msg); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - if (!(ctx.channel() instanceof DatagramChannel)) { - networkForwarder.disconnect((InetSocketAddress) ctx.channel().remoteAddress()); - } - super.channelInactive(ctx); - } - -} diff --git a/src/main/java/org/traccar/handler/NetworkMessageHandler.java b/src/main/java/org/traccar/handler/NetworkMessageHandler.java deleted file mode 100644 index b1d926bfa..000000000 --- a/src/main/java/org/traccar/handler/NetworkMessageHandler.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * Copyright 2019 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelDuplexHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelPromise; -import io.netty.channel.socket.DatagramChannel; -import io.netty.channel.socket.DatagramPacket; -import org.traccar.NetworkMessage; - -import java.net.InetSocketAddress; - -public class NetworkMessageHandler extends ChannelDuplexHandler { - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { - if (ctx.channel() instanceof DatagramChannel) { - DatagramPacket packet = (DatagramPacket) msg; - ctx.fireChannelRead(new NetworkMessage(packet.content(), packet.sender())); - } else if (msg instanceof ByteBuf) { - ByteBuf buffer = (ByteBuf) msg; - ctx.fireChannelRead(new NetworkMessage(buffer, ctx.channel().remoteAddress())); - } - } - - @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { - if (msg instanceof NetworkMessage) { - NetworkMessage message = (NetworkMessage) msg; - if (ctx.channel() instanceof DatagramChannel) { - InetSocketAddress recipient = (InetSocketAddress) message.getRemoteAddress(); - InetSocketAddress sender = (InetSocketAddress) ctx.channel().localAddress(); - ctx.write(new DatagramPacket((ByteBuf) message.getMessage(), recipient, sender), promise); - } else { - ctx.write(message.getMessage(), promise); - } - } else { - ctx.write(msg, promise); - } - } - -} diff --git a/src/main/java/org/traccar/handler/OpenChannelHandler.java b/src/main/java/org/traccar/handler/OpenChannelHandler.java deleted file mode 100644 index e416f35ae..000000000 --- a/src/main/java/org/traccar/handler/OpenChannelHandler.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2019 - 2021 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.channel.ChannelDuplexHandler; -import io.netty.channel.ChannelHandlerContext; -import org.traccar.TrackerConnector; - -public class OpenChannelHandler extends ChannelDuplexHandler { - - private final TrackerConnector connector; - - public OpenChannelHandler(TrackerConnector connector) { - this.connector = connector; - } - - @Override - public void channelActive(ChannelHandlerContext ctx) throws Exception { - super.channelActive(ctx); - connector.getChannelGroup().add(ctx.channel()); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - super.channelInactive(ctx); - connector.getChannelGroup().remove(ctx.channel()); - } - -} diff --git a/src/main/java/org/traccar/handler/PositionForwardingHandler.java b/src/main/java/org/traccar/handler/PositionForwardingHandler.java new file mode 100644 index 000000000..be62fff37 --- /dev/null +++ b/src/main/java/org/traccar/handler/PositionForwardingHandler.java @@ -0,0 +1,137 @@ +/* + * Copyright 2015 - 2024 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler; + +import io.netty.util.Timeout; +import io.netty.util.Timer; +import io.netty.util.TimerTask; +import jakarta.annotation.Nullable; +import jakarta.inject.Inject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.config.Config; +import org.traccar.config.Keys; +import org.traccar.forward.PositionData; +import org.traccar.forward.PositionForwarder; +import org.traccar.forward.ResultHandler; +import org.traccar.model.Device; +import org.traccar.model.Position; +import org.traccar.session.cache.CacheManager; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +public class PositionForwardingHandler extends BasePositionHandler { + + private static final Logger LOGGER = LoggerFactory.getLogger(PositionForwardingHandler.class); + + private final CacheManager cacheManager; + private final Timer timer; + + private final PositionForwarder positionForwarder; + + private final boolean retryEnabled; + private final int retryDelay; + private final int retryCount; + private final int retryLimit; + + private final AtomicInteger deliveryPending; + + @Inject + public PositionForwardingHandler( + Config config, CacheManager cacheManager, Timer timer, @Nullable PositionForwarder positionForwarder) { + + this.cacheManager = cacheManager; + this.timer = timer; + this.positionForwarder = positionForwarder; + + this.retryEnabled = config.getBoolean(Keys.FORWARD_RETRY_ENABLE); + this.retryDelay = config.getInteger(Keys.FORWARD_RETRY_DELAY); + this.retryCount = config.getInteger(Keys.FORWARD_RETRY_COUNT); + this.retryLimit = config.getInteger(Keys.FORWARD_RETRY_LIMIT); + + this.deliveryPending = new AtomicInteger(); + } + + class AsyncRequestAndCallback implements ResultHandler, TimerTask { + + private final PositionData positionData; + + private int retries = 0; + + AsyncRequestAndCallback(PositionData positionData) { + this.positionData = positionData; + deliveryPending.incrementAndGet(); + } + + private void send() { + positionForwarder.forward(positionData, this); + } + + private void retry(Throwable throwable) { + boolean scheduled = false; + try { + if (retryEnabled && deliveryPending.get() <= retryLimit && retries < retryCount) { + schedule(); + scheduled = true; + } + } finally { + int pending = scheduled ? deliveryPending.get() : deliveryPending.decrementAndGet(); + LOGGER.warn("Position forwarding failed: " + pending + " pending", throwable); + } + } + + private void schedule() { + timer.newTimeout(this, retryDelay * (long) Math.pow(2, retries++), TimeUnit.MILLISECONDS); + } + + @Override + public void onResult(boolean success, Throwable throwable) { + if (success) { + deliveryPending.decrementAndGet(); + } else { + retry(throwable); + } + } + + @Override + public void run(Timeout timeout) { + boolean sent = false; + try { + if (!timeout.isCancelled()) { + send(); + sent = true; + } + } finally { + if (!sent) { + deliveryPending.decrementAndGet(); + } + } + } + } + + @Override + public void handlePosition(Position position, Callback callback) { + if (positionForwarder != null) { + PositionData positionData = new PositionData(); + positionData.setPosition(position); + positionData.setDevice(cacheManager.getObject(Device.class, position.getDeviceId())); + new AsyncRequestAndCallback(positionData).send(); + } + callback.processed(position); + } + +} diff --git a/src/main/java/org/traccar/handler/RemoteAddressHandler.java b/src/main/java/org/traccar/handler/RemoteAddressHandler.java deleted file mode 100644 index 61ada5b91..000000000 --- a/src/main/java/org/traccar/handler/RemoteAddressHandler.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * Copyright 2015 - 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import org.traccar.config.Config; -import org.traccar.config.Keys; -import org.traccar.model.Position; - -import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import java.net.InetSocketAddress; - -@Singleton -@ChannelHandler.Sharable -public class RemoteAddressHandler extends ChannelInboundHandlerAdapter { - - private final boolean enabled; - - @Inject - public RemoteAddressHandler(Config config) { - enabled = config.getBoolean(Keys.PROCESSING_REMOTE_ADDRESS_ENABLE); - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { - - if (enabled) { - InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress(); - String hostAddress = remoteAddress != null ? remoteAddress.getAddress().getHostAddress() : null; - - if (msg instanceof Position) { - Position position = (Position) msg; - position.set(Position.KEY_IP, hostAddress); - } - } - - ctx.fireChannelRead(msg); - } - -} diff --git a/src/main/java/org/traccar/handler/SpeedLimitHandler.java b/src/main/java/org/traccar/handler/SpeedLimitHandler.java index 6edb6e912..604c10ca7 100644 --- a/src/main/java/org/traccar/handler/SpeedLimitHandler.java +++ b/src/main/java/org/traccar/handler/SpeedLimitHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2020 - 2024 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,20 +15,13 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; +import jakarta.inject.Inject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.model.Position; import org.traccar.speedlimit.SpeedLimitProvider; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable -public class SpeedLimitHandler extends ChannelInboundHandlerAdapter { +public class SpeedLimitHandler extends BasePositionHandler { private static final Logger LOGGER = LoggerFactory.getLogger(SpeedLimitHandler.class); @@ -40,26 +33,22 @@ public class SpeedLimitHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(final ChannelHandlerContext ctx, Object message) { - if (message instanceof Position) { - final Position position = (Position) message; - speedLimitProvider.getSpeedLimit(position.getLatitude(), position.getLongitude(), - new SpeedLimitProvider.SpeedLimitProviderCallback() { - @Override - public void onSuccess(double speedLimit) { - position.set(Position.KEY_SPEED_LIMIT, speedLimit); - ctx.fireChannelRead(position); - } - - @Override - public void onFailure(Throwable e) { - LOGGER.warn("Speed limit provider failed", e); - ctx.fireChannelRead(position); - } - }); - } else { - ctx.fireChannelRead(message); - } + public void handlePosition(Position position, Callback callback) { + + speedLimitProvider.getSpeedLimit(position.getLatitude(), position.getLongitude(), + new SpeedLimitProvider.SpeedLimitProviderCallback() { + @Override + public void onSuccess(double speedLimit) { + position.set(Position.KEY_SPEED_LIMIT, speedLimit); + callback.processed(position); + } + + @Override + public void onFailure(Throwable e) { + LOGGER.warn("Speed limit provider failed", e); + callback.processed(position); + } + }); } } diff --git a/src/main/java/org/traccar/handler/StandardLoggingHandler.java b/src/main/java/org/traccar/handler/StandardLoggingHandler.java deleted file mode 100644 index 513602dd8..000000000 --- a/src/main/java/org/traccar/handler/StandardLoggingHandler.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Copyright 2019 - 2024 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.traccar.handler; - -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufUtil; -import io.netty.channel.ChannelDuplexHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelPromise; -import jakarta.inject.Inject; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.traccar.NetworkMessage; -import org.traccar.helper.NetworkUtil; -import org.traccar.model.LogRecord; -import org.traccar.session.ConnectionManager; - -public class StandardLoggingHandler extends ChannelDuplexHandler { - - private static final Logger LOGGER = LoggerFactory.getLogger(StandardLoggingHandler.class); - - private final String protocol; - private ConnectionManager connectionManager; - - public StandardLoggingHandler(String protocol) { - this.protocol = protocol; - } - - @Inject - public void setConnectionManager(ConnectionManager connectionManager) { - this.connectionManager = connectionManager; - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - LogRecord record = createLogRecord(ctx, msg); - log(ctx, false, record); - super.channelRead(ctx, msg); - if (record != null) { - connectionManager.updateLog(record); - } - } - - @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { - log(ctx, true, createLogRecord(ctx, msg)); - super.write(ctx, msg, promise); - } - - private LogRecord createLogRecord(ChannelHandlerContext ctx, Object msg) { - if (msg instanceof NetworkMessage) { - NetworkMessage networkMessage = (NetworkMessage) msg; - if (networkMessage.getMessage() instanceof ByteBuf) { - LogRecord record = new LogRecord(ctx.channel().localAddress(), networkMessage.getRemoteAddress()); - record.setProtocol(protocol); - record.setData(ByteBufUtil.hexDump((ByteBuf) networkMessage.getMessage())); - return record; - } - } - return null; - } - - private void log(ChannelHandlerContext ctx, boolean downstream, LogRecord record) { - if (record != null) { - StringBuilder message = new StringBuilder(); - message.append("[").append(NetworkUtil.session(ctx.channel())).append(": "); - message.append(protocol); - message.append(downstream ? " > " : " < "); - message.append(record.getAddress().getHostString()); - message.append("] "); - message.append(record.getData()); - LOGGER.info(message.toString()); - } - } - -} diff --git a/src/main/java/org/traccar/handler/TimeHandler.java b/src/main/java/org/traccar/handler/TimeHandler.java index ef0c3445d..2e703c681 100644 --- a/src/main/java/org/traccar/handler/TimeHandler.java +++ b/src/main/java/org/traccar/handler/TimeHandler.java @@ -15,11 +15,7 @@ */ package org.traccar.handler; -import io.netty.channel.ChannelHandler; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import jakarta.inject.Inject; -import jakarta.inject.Singleton; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.model.Position; @@ -28,9 +24,7 @@ import java.util.Arrays; import java.util.HashSet; import java.util.Set; -@Singleton -@ChannelHandler.Sharable -public class TimeHandler extends ChannelInboundHandlerAdapter { +public class TimeHandler extends BasePositionHandler { private final boolean enabled; private final boolean useServerTime; @@ -53,10 +47,9 @@ public class TimeHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { + public void handlePosition(Position position, Callback callback) { - if (enabled && msg instanceof Position) { - Position position = (Position) msg; + if (enabled) { if (protocols == null || protocols.contains(position.getProtocol())) { if (useServerTime) { position.setDeviceTime(position.getServerTime()); @@ -66,7 +59,7 @@ public class TimeHandler extends ChannelInboundHandlerAdapter { } } } - ctx.fireChannelRead(msg); + callback.processed(position); } } diff --git a/src/main/java/org/traccar/handler/events/AlertEventHandler.java b/src/main/java/org/traccar/handler/events/AlertEventHandler.java index 531a0f957..ca580b60d 100644 --- a/src/main/java/org/traccar/handler/events/AlertEventHandler.java +++ b/src/main/java/org/traccar/handler/events/AlertEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 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,13 @@ */ package org.traccar.handler.events; -import java.util.Collections; -import java.util.Map; - -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable public class AlertEventHandler extends BaseEventHandler { private final CacheManager cacheManager; @@ -42,7 +34,7 @@ public class AlertEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { Object alarm = position.getAttributes().get(Position.KEY_ALARM); if (alarm != null) { boolean ignoreAlert = false; @@ -55,10 +47,9 @@ public class AlertEventHandler extends BaseEventHandler { if (!ignoreAlert) { Event event = new Event(Event.TYPE_ALARM, position); event.set(Position.KEY_ALARM, (String) alarm); - return Collections.singletonMap(event, position); + callback.eventDetected(event); } } - return null; } } diff --git a/src/main/java/org/traccar/handler/events/BaseEventHandler.java b/src/main/java/org/traccar/handler/events/BaseEventHandler.java index 4a4fb40ff..009c83145 100644 --- a/src/main/java/org/traccar/handler/events/BaseEventHandler.java +++ b/src/main/java/org/traccar/handler/events/BaseEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 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,33 +15,17 @@ */ package org.traccar.handler.events; -import java.util.Map; - -import org.traccar.BaseDataHandler; -import org.traccar.database.NotificationManager; import org.traccar.model.Event; import org.traccar.model.Position; -import jakarta.inject.Inject; - -public abstract class BaseEventHandler extends BaseDataHandler { - - private NotificationManager notificationManager; +public abstract class BaseEventHandler { - @Inject - public void setNotificationManager(NotificationManager notificationManager) { - this.notificationManager = notificationManager; + public interface Callback { + void eventDetected(Event event); } - @Override - protected Position handlePosition(Position position) { - Map events = analyzePosition(position); - if (events != null && !events.isEmpty()) { - notificationManager.updateEvents(events); - } - return position; - } - - protected abstract Map analyzePosition(Position position); - + /** + * Event handlers should be processed synchronously. + */ + public abstract void analyzePosition(Position position, Callback callback); } diff --git a/src/main/java/org/traccar/handler/events/BehaviorEventHandler.java b/src/main/java/org/traccar/handler/events/BehaviorEventHandler.java index 08ae35fcd..d654e18ce 100644 --- a/src/main/java/org/traccar/handler/events/BehaviorEventHandler.java +++ b/src/main/java/org/traccar/handler/events/BehaviorEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2021 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2021 - 2024 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,7 +15,7 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.helper.UnitsConverter; @@ -23,13 +23,6 @@ import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import java.util.Collections; -import java.util.Map; - -@Singleton -@ChannelHandler.Sharable public class BehaviorEventHandler extends BaseEventHandler { private final double accelerationThreshold; @@ -45,7 +38,7 @@ public class BehaviorEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { Position lastPosition = cacheManager.getPosition(position.getDeviceId()); if (lastPosition != null && position.getFixTime().equals(lastPosition.getFixTime())) { @@ -54,14 +47,13 @@ public class BehaviorEventHandler extends BaseEventHandler { if (accelerationThreshold != 0 && acceleration >= accelerationThreshold) { Event event = new Event(Event.TYPE_ALARM, position); event.set(Position.KEY_ALARM, Position.ALARM_ACCELERATION); - return Collections.singletonMap(event, position); + callback.eventDetected(event); } else if (brakingThreshold != 0 && acceleration <= -brakingThreshold) { Event event = new Event(Event.TYPE_ALARM, position); event.set(Position.KEY_ALARM, Position.ALARM_BRAKING); - return Collections.singletonMap(event, position); + callback.eventDetected(event); } } - return null; } } diff --git a/src/main/java/org/traccar/handler/events/CommandResultEventHandler.java b/src/main/java/org/traccar/handler/events/CommandResultEventHandler.java index b70f8f33b..b98807b23 100644 --- a/src/main/java/org/traccar/handler/events/CommandResultEventHandler.java +++ b/src/main/java/org/traccar/handler/events/CommandResultEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 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,18 +15,10 @@ */ package org.traccar.handler.events; -import java.util.Collections; -import java.util.Map; - -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.model.Event; import org.traccar.model.Position; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable public class CommandResultEventHandler extends BaseEventHandler { @Inject @@ -34,14 +26,13 @@ public class CommandResultEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { Object commandResult = position.getAttributes().get(Position.KEY_RESULT); if (commandResult != null) { Event event = new Event(Event.TYPE_COMMAND_RESULT, position); event.set(Position.KEY_RESULT, (String) commandResult); - return Collections.singletonMap(event, position); + callback.eventDetected(event); } - return null; } } diff --git a/src/main/java/org/traccar/handler/events/DriverEventHandler.java b/src/main/java/org/traccar/handler/events/DriverEventHandler.java index b68327983..31f8d2b4b 100644 --- a/src/main/java/org/traccar/handler/events/DriverEventHandler.java +++ b/src/main/java/org/traccar/handler/events/DriverEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2017 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2017 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,19 +16,12 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.helper.model.PositionUtil; import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import java.util.Collections; -import java.util.Map; - -@Singleton -@ChannelHandler.Sharable public class DriverEventHandler extends BaseEventHandler { private final CacheManager cacheManager; @@ -39,9 +32,9 @@ public class DriverEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { if (!PositionUtil.isLatest(cacheManager, position)) { - return null; + return; } String driverUniqueId = position.getString(Position.KEY_DRIVER_UNIQUE_ID); if (driverUniqueId != null) { @@ -53,10 +46,9 @@ public class DriverEventHandler extends BaseEventHandler { if (!driverUniqueId.equals(oldDriverUniqueId)) { Event event = new Event(Event.TYPE_DRIVER_CHANGED, position); event.set(Position.KEY_DRIVER_UNIQUE_ID, driverUniqueId); - return Collections.singletonMap(event, position); + callback.eventDetected(event); } } - return null; } } diff --git a/src/main/java/org/traccar/handler/events/FuelEventHandler.java b/src/main/java/org/traccar/handler/events/FuelEventHandler.java index e5085ecc2..c5675f51d 100644 --- a/src/main/java/org/traccar/handler/events/FuelEventHandler.java +++ b/src/main/java/org/traccar/handler/events/FuelEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2017 - 2024 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,7 +15,7 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.config.Keys; import org.traccar.helper.model.AttributeUtil; import org.traccar.helper.model.PositionUtil; @@ -24,12 +24,6 @@ import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import java.util.Map; - -@Singleton -@ChannelHandler.Sharable public class FuelEventHandler extends BaseEventHandler { private final CacheManager cacheManager; @@ -40,14 +34,14 @@ public class FuelEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { Device device = cacheManager.getObject(Device.class, position.getDeviceId()); if (device == null) { - return null; + return; } if (!PositionUtil.isLatest(cacheManager, position)) { - return null; + return; } if (position.hasAttribute(Position.KEY_FUEL_LEVEL)) { @@ -61,19 +55,17 @@ public class FuelEventHandler extends BaseEventHandler { double threshold = AttributeUtil.lookup( cacheManager, Keys.EVENT_FUEL_INCREASE_THRESHOLD, position.getDeviceId()); if (threshold > 0 && change >= threshold) { - return Map.of(new Event(Event.TYPE_DEVICE_FUEL_INCREASE, position), position); + callback.eventDetected(new Event(Event.TYPE_DEVICE_FUEL_INCREASE, position)); } } else if (change < 0) { double threshold = AttributeUtil.lookup( cacheManager, Keys.EVENT_FUEL_DROP_THRESHOLD, position.getDeviceId()); if (threshold > 0 && Math.abs(change) >= threshold) { - return Map.of(new Event(Event.TYPE_DEVICE_FUEL_DROP, position), position); + callback.eventDetected(new Event(Event.TYPE_DEVICE_FUEL_DROP, position)); } } } } - - return null; } } diff --git a/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java b/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java index dbe2b8118..c8ecfb1ed 100644 --- a/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java +++ b/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 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,7 +15,7 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.helper.model.PositionUtil; import org.traccar.model.Calendar; import org.traccar.model.Event; @@ -23,15 +23,9 @@ import org.traccar.model.Geofence; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; -@Singleton -@ChannelHandler.Sharable public class GeofenceEventHandler extends BaseEventHandler { private final CacheManager cacheManager; @@ -42,9 +36,9 @@ public class GeofenceEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { if (!PositionUtil.isLatest(cacheManager, position)) { - return null; + return; } List oldGeofences = new ArrayList<>(); @@ -60,7 +54,6 @@ public class GeofenceEventHandler extends BaseEventHandler { oldGeofences.removeAll(position.getGeofenceIds()); } - Map events = new HashMap<>(); for (long geofenceId : oldGeofences) { Geofence geofence = cacheManager.getObject(Geofence.class, geofenceId); if (geofence != null) { @@ -69,7 +62,7 @@ public class GeofenceEventHandler extends BaseEventHandler { if (calendar == null || calendar.checkMoment(position.getFixTime())) { Event event = new Event(Event.TYPE_GEOFENCE_EXIT, position); event.setGeofenceId(geofenceId); - events.put(event, position); + callback.eventDetected(event); } } } @@ -79,10 +72,8 @@ public class GeofenceEventHandler extends BaseEventHandler { if (calendar == null || calendar.checkMoment(position.getFixTime())) { Event event = new Event(Event.TYPE_GEOFENCE_ENTER, position); event.setGeofenceId(geofenceId); - events.put(event, position); + callback.eventDetected(event); } } - return events; } - } diff --git a/src/main/java/org/traccar/handler/events/IgnitionEventHandler.java b/src/main/java/org/traccar/handler/events/IgnitionEventHandler.java index ba4159a42..bbf9fadd1 100644 --- a/src/main/java/org/traccar/handler/events/IgnitionEventHandler.java +++ b/src/main/java/org/traccar/handler/events/IgnitionEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2016 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,21 +16,13 @@ */ package org.traccar.handler.events; -import java.util.Collections; -import java.util.Map; - -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.helper.model.PositionUtil; import org.traccar.model.Device; import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable public class IgnitionEventHandler extends BaseEventHandler { private final CacheManager cacheManager; @@ -41,14 +33,12 @@ public class IgnitionEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { Device device = cacheManager.getObject(Device.class, position.getDeviceId()); if (device == null || !PositionUtil.isLatest(cacheManager, position)) { - return null; + return; } - Map result = null; - if (position.hasAttribute(Position.KEY_IGNITION)) { boolean ignition = position.getBoolean(Position.KEY_IGNITION); @@ -57,15 +47,12 @@ public class IgnitionEventHandler extends BaseEventHandler { boolean oldIgnition = lastPosition.getBoolean(Position.KEY_IGNITION); if (ignition && !oldIgnition) { - result = Collections.singletonMap( - new Event(Event.TYPE_IGNITION_ON, position), position); + callback.eventDetected(new Event(Event.TYPE_IGNITION_ON, position)); } else if (!ignition && oldIgnition) { - result = Collections.singletonMap( - new Event(Event.TYPE_IGNITION_OFF, position), position); + callback.eventDetected(new Event(Event.TYPE_IGNITION_OFF, position)); } } } - return result; } } diff --git a/src/main/java/org/traccar/handler/events/MaintenanceEventHandler.java b/src/main/java/org/traccar/handler/events/MaintenanceEventHandler.java index 2fa2e8869..573ad4ad6 100644 --- a/src/main/java/org/traccar/handler/events/MaintenanceEventHandler.java +++ b/src/main/java/org/traccar/handler/events/MaintenanceEventHandler.java @@ -16,20 +16,12 @@ */ package org.traccar.handler.events; -import java.util.HashMap; -import java.util.Map; - -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.model.Event; import org.traccar.model.Maintenance; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; - -@Singleton -@ChannelHandler.Sharable public class MaintenanceEventHandler extends BaseEventHandler { private final CacheManager cacheManager; @@ -40,13 +32,12 @@ public class MaintenanceEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { Position lastPosition = cacheManager.getPosition(position.getDeviceId()); if (lastPosition == null || position.getFixTime().compareTo(lastPosition.getFixTime()) < 0) { - return null; + return; } - Map events = new HashMap<>(); for (Maintenance maintenance : cacheManager.getDeviceObjects(position.getDeviceId(), Maintenance.class)) { if (maintenance.getPeriod() != 0) { double oldValue = getValue(lastPosition, maintenance.getType()); @@ -58,13 +49,11 @@ public class MaintenanceEventHandler extends BaseEventHandler { Event event = new Event(Event.TYPE_MAINTENANCE, position); event.setMaintenanceId(maintenance.getId()); event.set(maintenance.getType(), newValue); - events.put(event, position); + callback.eventDetected(event); } } } } - - return events; } private double getValue(Position position, String type) { diff --git a/src/main/java/org/traccar/handler/events/MediaEventHandler.java b/src/main/java/org/traccar/handler/events/MediaEventHandler.java index 52d8e6961..2745296c4 100644 --- a/src/main/java/org/traccar/handler/events/MediaEventHandler.java +++ b/src/main/java/org/traccar/handler/events/MediaEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2022 - 2024 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,18 +15,12 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.traccar.model.Event; import org.traccar.model.Position; -import jakarta.inject.Inject; -import jakarta.inject.Singleton; -import java.util.Map; -import java.util.stream.Collectors; import java.util.stream.Stream; -@Singleton -@ChannelHandler.Sharable public class MediaEventHandler extends BaseEventHandler { @Inject @@ -34,8 +28,8 @@ public class MediaEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { - return Stream.of(Position.KEY_IMAGE, Position.KEY_VIDEO, Position.KEY_AUDIO) + public void analyzePosition(Position position, Callback callback) { + Stream.of(Position.KEY_IMAGE, Position.KEY_VIDEO, Position.KEY_AUDIO) .filter(position::hasAttribute) .map(type -> { Event event = new Event(Event.TYPE_MEDIA, position); @@ -43,7 +37,7 @@ public class MediaEventHandler extends BaseEventHandler { event.set("file", position.getString(type)); return event; }) - .collect(Collectors.toMap(event -> event, event -> position)); + .forEach(callback::eventDetected); } } diff --git a/src/main/java/org/traccar/handler/events/MotionEventHandler.java b/src/main/java/org/traccar/handler/events/MotionEventHandler.java index 15902d6d4..41d68985b 100644 --- a/src/main/java/org/traccar/handler/events/MotionEventHandler.java +++ b/src/main/java/org/traccar/handler/events/MotionEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2017 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,14 +16,13 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Keys; import org.traccar.helper.model.AttributeUtil; import org.traccar.helper.model.PositionUtil; import org.traccar.model.Device; -import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.reports.common.TripsConfig; import org.traccar.session.cache.CacheManager; @@ -35,13 +34,6 @@ 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.Collections; -import java.util.Map; - -@Singleton -@ChannelHandler.Sharable public class MotionEventHandler extends BaseEventHandler { private static final Logger LOGGER = LoggerFactory.getLogger(MotionEventHandler.class); @@ -56,17 +48,17 @@ public class MotionEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { long deviceId = position.getDeviceId(); Device device = cacheManager.getObject(Device.class, deviceId); if (device == null || !PositionUtil.isLatest(cacheManager, position)) { - return null; + return; } boolean processInvalid = AttributeUtil.lookup( cacheManager, Keys.EVENT_MOTION_PROCESS_INVALID_POSITIONS, deviceId); if (!processInvalid && !position.getValid()) { - return null; + return; } TripsConfig tripsConfig = new TripsConfig(new AttributeUtil.CacheProvider(cacheManager, deviceId)); @@ -82,7 +74,9 @@ public class MotionEventHandler extends BaseEventHandler { LOGGER.warn("Update device motion error", e); } } - return state.getEvent() != null ? Collections.singletonMap(state.getEvent(), position) : null; + if (state.getEvent() != null) { + callback.eventDetected(state.getEvent()); + } } } diff --git a/src/main/java/org/traccar/handler/events/OverspeedEventHandler.java b/src/main/java/org/traccar/handler/events/OverspeedEventHandler.java index 3bb5f713c..9598581e6 100644 --- a/src/main/java/org/traccar/handler/events/OverspeedEventHandler.java +++ b/src/main/java/org/traccar/handler/events/OverspeedEventHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 - 2023 Anton Tananaev (anton@traccar.org) + * Copyright 2016 - 2024 Anton Tananaev (anton@traccar.org) * Copyright 2018 Andrey Kunitsyn (andrey@traccar.org) * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -16,7 +16,7 @@ */ package org.traccar.handler.events; -import io.netty.channel.ChannelHandler; +import jakarta.inject.Inject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Config; @@ -24,7 +24,6 @@ import org.traccar.config.Keys; import org.traccar.helper.model.AttributeUtil; import org.traccar.helper.model.PositionUtil; import org.traccar.model.Device; -import org.traccar.model.Event; import org.traccar.model.Geofence; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; @@ -36,13 +35,6 @@ 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.Collections; -import java.util.Map; - -@Singleton -@ChannelHandler.Sharable public class OverspeedEventHandler extends BaseEventHandler { private static final Logger LOGGER = LoggerFactory.getLogger(OverspeedEventHandler.class); @@ -55,8 +47,7 @@ public class OverspeedEventHandler extends BaseEventHandler { private final double multiplier; @Inject - public OverspeedEventHandler( - Config config, CacheManager cacheManager, Storage storage) { + public OverspeedEventHandler(Config config, CacheManager cacheManager, Storage storage) { this.cacheManager = cacheManager; this.storage = storage; minimalDuration = config.getLong(Keys.EVENT_OVERSPEED_MINIMAL_DURATION) * 1000; @@ -65,15 +56,15 @@ public class OverspeedEventHandler extends BaseEventHandler { } @Override - protected Map analyzePosition(Position position) { + public void analyzePosition(Position position, Callback callback) { long deviceId = position.getDeviceId(); Device device = cacheManager.getObject(Device.class, position.getDeviceId()); if (device == null) { - return null; + return; } if (!PositionUtil.isLatest(cacheManager, position) || !position.getValid()) { - return null; + return; } double speedLimit = AttributeUtil.lookup(cacheManager, Keys.EVENT_OVERSPEED_LIMIT, deviceId); @@ -105,7 +96,7 @@ public class OverspeedEventHandler extends BaseEventHandler { } if (speedLimit == 0) { - return null; + return; } OverspeedState state = OverspeedState.fromDevice(device); @@ -120,7 +111,9 @@ public class OverspeedEventHandler extends BaseEventHandler { LOGGER.warn("Update device overspeed error", e); } } - return state.getEvent() != null ? Collections.singletonMap(state.getEvent(), position) : null; + if (state.getEvent() != null) { + callback.eventDetected(state.getEvent()); + } } } diff --git a/src/main/java/org/traccar/handler/network/AcknowledgementHandler.java b/src/main/java/org/traccar/handler/network/AcknowledgementHandler.java new file mode 100644 index 000000000..e87f5d34c --- /dev/null +++ b/src/main/java/org/traccar/handler/network/AcknowledgementHandler.java @@ -0,0 +1,121 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collection; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Set; + +public class AcknowledgementHandler extends ChannelOutboundHandlerAdapter { + + private static final Logger LOGGER = LoggerFactory.getLogger(AcknowledgementHandler.class); + + public interface Event { + } + + public static class EventReceived implements Event { + } + + public static class EventDecoded implements Event { + private final Collection objects; + + public EventDecoded(Collection objects) { + this.objects = objects; + } + + public Collection getObjects() { + return objects; + } + } + + public static class EventHandled implements Event { + private final Object object; + + public EventHandled(Object object) { + this.object = object; + } + + public Object getObject() { + return object; + } + } + + private static final class Entry { + private final Object message; + private final ChannelPromise promise; + + private Entry(Object message, ChannelPromise promise) { + this.message = message; + this.promise = promise; + } + + public Object getMessage() { + return message; + } + + public ChannelPromise getPromise() { + return promise; + } + } + + private List queue; + private final Set waiting = new HashSet<>(); + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + List output = new LinkedList<>(); + synchronized (this) { + if (msg instanceof Event) { + if (msg instanceof EventReceived) { + LOGGER.debug("Event received"); + if (queue == null) { + queue = new LinkedList<>(); + } + } else if (msg instanceof EventDecoded) { + EventDecoded event = (EventDecoded) msg; + LOGGER.debug("Event decoded {}", event.getObjects().size()); + waiting.addAll(event.getObjects()); + } else if (msg instanceof EventHandled) { + EventHandled event = (EventHandled) msg; + LOGGER.debug("Event handled"); + waiting.remove(event.getObject()); + } + if (!(msg instanceof EventReceived) && waiting.isEmpty()) { + output.addAll(queue); + queue = null; + } + } else if (queue != null) { + LOGGER.debug("Message queued"); + queue.add(new Entry(msg, promise)); + } else { + LOGGER.debug("Message sent"); + output.add(new Entry(msg, promise)); + } + } + for (Entry entry : output) { + ctx.write(entry.getMessage(), entry.getPromise()); + } + } + +} diff --git a/src/main/java/org/traccar/handler/network/MainEventHandler.java b/src/main/java/org/traccar/handler/network/MainEventHandler.java new file mode 100644 index 000000000..901036d44 --- /dev/null +++ b/src/main/java/org/traccar/handler/network/MainEventHandler.java @@ -0,0 +1,195 @@ +/* + * 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +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; +import io.netty.handler.codec.http.HttpRequestDecoder; +import io.netty.handler.timeout.IdleStateEvent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.BasePipelineFactory; +import org.traccar.BaseProtocolDecoder; +import org.traccar.config.Config; +import org.traccar.config.Keys; +import org.traccar.database.StatisticsManager; +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); + + private final Set connectionlessProtocols = new HashSet<>(); + private final Set logAttributes = new LinkedHashSet<>(); + + 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(config.getString(Keys.LOGGER_ATTRIBUTES).split("[, ]"))); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (msg instanceof Position) { + + Position position = (Position) msg; + Device device = cacheManager.getObject(Device.class, position.getDeviceId()); + + try { + 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); + } + + StringBuilder builder = new StringBuilder(); + builder.append("[").append(NetworkUtil.session(ctx.channel())).append("] "); + builder.append("id: ").append(device.getUniqueId()); + for (String attribute : logAttributes) { + switch (attribute) { + case "time": + builder.append(", time: ").append(DateUtil.formatDate(position.getFixTime(), false)); + break; + case "position": + builder.append(", lat: ").append(String.format("%.5f", position.getLatitude())); + builder.append(", lon: ").append(String.format("%.5f", position.getLongitude())); + break; + case "speed": + if (position.getSpeed() > 0) { + builder.append(", speed: ").append(String.format("%.1f", position.getSpeed())); + } + break; + case "course": + builder.append(", course: ").append(String.format("%.1f", position.getCourse())); + break; + case "accuracy": + if (position.getAccuracy() > 0) { + builder.append(", accuracy: ").append(String.format("%.1f", position.getAccuracy())); + } + break; + case "outdated": + if (position.getOutdated()) { + builder.append(", outdated"); + } + break; + case "invalid": + if (!position.getValid()) { + builder.append(", invalid"); + } + break; + default: + Object value = position.getAttributes().get(attribute); + if (value != null) { + builder.append(", ").append(attribute).append(": ").append(value); + } + break; + } + } + LOGGER.info(builder.toString()); + + statisticsManager.registerMessageStored(position.getDeviceId(), position.getProtocol()); + + ctx.writeAndFlush(new AcknowledgementHandler.EventHandled(position)); + } + } + + @Override + public void channelActive(ChannelHandlerContext ctx) { + if (!(ctx.channel() instanceof DatagramChannel)) { + LOGGER.info("[{}] connected", NetworkUtil.session(ctx.channel())); + } + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) { + LOGGER.info("[{}] disconnected", NetworkUtil.session(ctx.channel())); + closeChannel(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 + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + while (cause.getCause() != null && cause.getCause() != cause) { + cause = cause.getCause(); + } + 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("[{}] timed out", NetworkUtil.session(ctx.channel())); + closeChannel(ctx.channel()); + } + } + + private void closeChannel(Channel channel) { + if (!(channel instanceof DatagramChannel)) { + channel.close(); + } + } + +} diff --git a/src/main/java/org/traccar/handler/network/NetworkForwarderHandler.java b/src/main/java/org/traccar/handler/network/NetworkForwarderHandler.java new file mode 100644 index 000000000..5f07ce355 --- /dev/null +++ b/src/main/java/org/traccar/handler/network/NetworkForwarderHandler.java @@ -0,0 +1,72 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.socket.DatagramChannel; +import io.netty.channel.socket.DatagramPacket; +import org.traccar.forward.NetworkForwarder; + +import jakarta.inject.Inject; +import java.net.InetSocketAddress; +import java.net.SocketAddress; + +public class NetworkForwarderHandler extends ChannelInboundHandlerAdapter { + + private final int port; + + private NetworkForwarder networkForwarder; + + public NetworkForwarderHandler(int port) { + this.port = port; + } + + @Inject + public void setNetworkForwarder(NetworkForwarder networkForwarder) { + this.networkForwarder = networkForwarder; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + boolean datagram = ctx.channel() instanceof DatagramChannel; + SocketAddress remoteAddress; + ByteBuf buffer; + if (datagram) { + DatagramPacket message = (DatagramPacket) msg; + remoteAddress = message.recipient(); + buffer = message.content(); + } else { + remoteAddress = ctx.channel().remoteAddress(); + buffer = (ByteBuf) msg; + } + + byte[] data = new byte[buffer.readableBytes()]; + buffer.getBytes(buffer.readerIndex(), data); + networkForwarder.forward((InetSocketAddress) remoteAddress, port, datagram, data); + super.channelRead(ctx, msg); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + if (!(ctx.channel() instanceof DatagramChannel)) { + networkForwarder.disconnect((InetSocketAddress) ctx.channel().remoteAddress()); + } + super.channelInactive(ctx); + } + +} diff --git a/src/main/java/org/traccar/handler/network/NetworkMessageHandler.java b/src/main/java/org/traccar/handler/network/NetworkMessageHandler.java new file mode 100644 index 000000000..c2fb9016a --- /dev/null +++ b/src/main/java/org/traccar/handler/network/NetworkMessageHandler.java @@ -0,0 +1,57 @@ +/* + * Copyright 2019 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import io.netty.channel.socket.DatagramChannel; +import io.netty.channel.socket.DatagramPacket; +import org.traccar.NetworkMessage; + +import java.net.InetSocketAddress; + +public class NetworkMessageHandler extends ChannelDuplexHandler { + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (ctx.channel() instanceof DatagramChannel) { + DatagramPacket packet = (DatagramPacket) msg; + ctx.fireChannelRead(new NetworkMessage(packet.content(), packet.sender())); + } else if (msg instanceof ByteBuf) { + ByteBuf buffer = (ByteBuf) msg; + ctx.fireChannelRead(new NetworkMessage(buffer, ctx.channel().remoteAddress())); + } + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + if (msg instanceof NetworkMessage) { + NetworkMessage message = (NetworkMessage) msg; + if (ctx.channel() instanceof DatagramChannel) { + InetSocketAddress recipient = (InetSocketAddress) message.getRemoteAddress(); + InetSocketAddress sender = (InetSocketAddress) ctx.channel().localAddress(); + ctx.write(new DatagramPacket((ByteBuf) message.getMessage(), recipient, sender), promise); + } else { + ctx.write(message.getMessage(), promise); + } + } else { + ctx.write(msg, promise); + } + } + +} diff --git a/src/main/java/org/traccar/handler/network/OpenChannelHandler.java b/src/main/java/org/traccar/handler/network/OpenChannelHandler.java new file mode 100644 index 000000000..21aaae676 --- /dev/null +++ b/src/main/java/org/traccar/handler/network/OpenChannelHandler.java @@ -0,0 +1,42 @@ +/* + * Copyright 2019 - 2021 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import org.traccar.TrackerConnector; + +public class OpenChannelHandler extends ChannelDuplexHandler { + + private final TrackerConnector connector; + + public OpenChannelHandler(TrackerConnector connector) { + this.connector = connector; + } + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + super.channelActive(ctx); + connector.getChannelGroup().add(ctx.channel()); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + super.channelInactive(ctx); + connector.getChannelGroup().remove(ctx.channel()); + } + +} diff --git a/src/main/java/org/traccar/handler/network/RemoteAddressHandler.java b/src/main/java/org/traccar/handler/network/RemoteAddressHandler.java new file mode 100644 index 000000000..d545ed799 --- /dev/null +++ b/src/main/java/org/traccar/handler/network/RemoteAddressHandler.java @@ -0,0 +1,54 @@ +/* + * Copyright 2015 - 2024 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import jakarta.inject.Inject; +import org.traccar.config.Config; +import org.traccar.config.Keys; +import org.traccar.model.Position; + +import java.net.InetSocketAddress; + +@ChannelHandler.Sharable +public class RemoteAddressHandler extends ChannelInboundHandlerAdapter { + + private final boolean enabled; + + @Inject + public RemoteAddressHandler(Config config) { + enabled = config.getBoolean(Keys.PROCESSING_REMOTE_ADDRESS_ENABLE); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + + if (enabled) { + InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress(); + String hostAddress = remoteAddress != null ? remoteAddress.getAddress().getHostAddress() : null; + + if (msg instanceof Position) { + Position position = (Position) msg; + position.set(Position.KEY_IP, hostAddress); + } + } + + ctx.fireChannelRead(msg); + } + +} diff --git a/src/main/java/org/traccar/handler/network/StandardLoggingHandler.java b/src/main/java/org/traccar/handler/network/StandardLoggingHandler.java new file mode 100644 index 000000000..dae93655d --- /dev/null +++ b/src/main/java/org/traccar/handler/network/StandardLoggingHandler.java @@ -0,0 +1,89 @@ +/* + * Copyright 2019 - 2024 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.handler.network; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import jakarta.inject.Inject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.NetworkMessage; +import org.traccar.helper.NetworkUtil; +import org.traccar.model.LogRecord; +import org.traccar.session.ConnectionManager; + +public class StandardLoggingHandler extends ChannelDuplexHandler { + + private static final Logger LOGGER = LoggerFactory.getLogger(StandardLoggingHandler.class); + + private final String protocol; + private ConnectionManager connectionManager; + + public StandardLoggingHandler(String protocol) { + this.protocol = protocol; + } + + @Inject + public void setConnectionManager(ConnectionManager connectionManager) { + this.connectionManager = connectionManager; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + LogRecord record = createLogRecord(ctx, msg); + log(ctx, false, record); + super.channelRead(ctx, msg); + if (record != null) { + connectionManager.updateLog(record); + } + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + log(ctx, true, createLogRecord(ctx, msg)); + super.write(ctx, msg, promise); + } + + private LogRecord createLogRecord(ChannelHandlerContext ctx, Object msg) { + if (msg instanceof NetworkMessage) { + NetworkMessage networkMessage = (NetworkMessage) msg; + if (networkMessage.getMessage() instanceof ByteBuf) { + LogRecord record = new LogRecord(ctx.channel().localAddress(), networkMessage.getRemoteAddress()); + record.setProtocol(protocol); + record.setData(ByteBufUtil.hexDump((ByteBuf) networkMessage.getMessage())); + return record; + } + } + return null; + } + + private void log(ChannelHandlerContext ctx, boolean downstream, LogRecord record) { + if (record != null) { + StringBuilder message = new StringBuilder(); + message.append("[").append(NetworkUtil.session(ctx.channel())).append(": "); + message.append(protocol); + message.append(downstream ? " > " : " < "); + message.append(record.getAddress().getHostString()); + message.append("] "); + message.append(record.getData()); + LOGGER.info(message.toString()); + } + } + +} diff --git a/src/test/java/org/traccar/handler/DistanceHandlerTest.java b/src/test/java/org/traccar/handler/DistanceHandlerTest.java index 7d2f1e2e3..04a27ad79 100644 --- a/src/test/java/org/traccar/handler/DistanceHandlerTest.java +++ b/src/test/java/org/traccar/handler/DistanceHandlerTest.java @@ -15,14 +15,15 @@ public class DistanceHandlerTest { DistanceHandler distanceHandler = new DistanceHandler(new Config(), mock(CacheManager.class)); - Position position = distanceHandler.handlePosition(new Position()); + Position position = new Position(); + distanceHandler.handlePosition(position, p -> {}); assertEquals(0.0, position.getAttributes().get(Position.KEY_DISTANCE)); assertEquals(0.0, position.getAttributes().get(Position.KEY_TOTAL_DISTANCE)); position.set(Position.KEY_DISTANCE, 100); - position = distanceHandler.handlePosition(position); + distanceHandler.handlePosition(position, p -> {}); assertEquals(100.0, position.getAttributes().get(Position.KEY_DISTANCE)); assertEquals(100.0, position.getAttributes().get(Position.KEY_TOTAL_DISTANCE)); diff --git a/src/test/java/org/traccar/handler/MotionHandlerTest.java b/src/test/java/org/traccar/handler/MotionHandlerTest.java index 10cdf6a90..927c803d9 100644 --- a/src/test/java/org/traccar/handler/MotionHandlerTest.java +++ b/src/test/java/org/traccar/handler/MotionHandlerTest.java @@ -26,7 +26,8 @@ public class MotionHandlerTest { MotionHandler motionHandler = new MotionHandler(cacheManager); - Position position = motionHandler.handlePosition(new Position()); + Position position = new Position(); + motionHandler.handlePosition(position, p -> {}); assertEquals(false, position.getAttributes().get(Position.KEY_MOTION)); diff --git a/src/test/java/org/traccar/handler/events/AlertEventHandlerTest.java b/src/test/java/org/traccar/handler/events/AlertEventHandlerTest.java index 66dc55c85..7b1991553 100644 --- a/src/test/java/org/traccar/handler/events/AlertEventHandlerTest.java +++ b/src/test/java/org/traccar/handler/events/AlertEventHandlerTest.java @@ -7,10 +7,11 @@ import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import java.util.Map; +import java.util.ArrayList; +import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.Mockito.mock; public class AlertEventHandlerTest extends BaseTest { @@ -22,9 +23,10 @@ public class AlertEventHandlerTest extends BaseTest { Position position = new Position(); position.set(Position.KEY_ALARM, Position.ALARM_GENERAL); - Map events = alertEventHandler.analyzePosition(position); - assertNotNull(events); - Event event = events.keySet().iterator().next(); + List events = new ArrayList<>(); + alertEventHandler.analyzePosition(position, events::add); + assertFalse(events.isEmpty()); + Event event = events.iterator().next(); assertEquals(Event.TYPE_ALARM, event.getType()); } diff --git a/src/test/java/org/traccar/handler/events/CommandResultEventHandlerTest.java b/src/test/java/org/traccar/handler/events/CommandResultEventHandlerTest.java index bc24e42f5..58e198b7f 100644 --- a/src/test/java/org/traccar/handler/events/CommandResultEventHandlerTest.java +++ b/src/test/java/org/traccar/handler/events/CommandResultEventHandlerTest.java @@ -5,9 +5,12 @@ import org.traccar.BaseTest; import org.traccar.model.Event; import org.traccar.model.Position; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; public class CommandResultEventHandlerTest extends BaseTest { @@ -19,9 +22,10 @@ public class CommandResultEventHandlerTest extends BaseTest { Position position = new Position(); position.set(Position.KEY_RESULT, "Test Result"); - Map events = commandResultEventHandler.analyzePosition(position); - assertNotNull(events); - Event event = events.keySet().iterator().next(); + List events = new ArrayList<>(); + commandResultEventHandler.analyzePosition(position, events::add); + assertFalse(events.isEmpty()); + Event event = events.iterator().next(); assertEquals(Event.TYPE_COMMAND_RESULT, event.getType()); } diff --git a/src/test/java/org/traccar/handler/events/IgnitionEventHandlerTest.java b/src/test/java/org/traccar/handler/events/IgnitionEventHandlerTest.java index 972932df4..61ff50efa 100644 --- a/src/test/java/org/traccar/handler/events/IgnitionEventHandlerTest.java +++ b/src/test/java/org/traccar/handler/events/IgnitionEventHandlerTest.java @@ -1,14 +1,11 @@ package org.traccar.handler.events; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.traccar.BaseTest; -import org.traccar.model.Event; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; -import java.util.Map; - -import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.Mockito.mock; public class IgnitionEventHandlerTest extends BaseTest { @@ -21,8 +18,7 @@ public class IgnitionEventHandlerTest extends BaseTest { Position position = new Position(); position.set(Position.KEY_IGNITION, true); position.setValid(true); - Map events = ignitionEventHandler.analyzePosition(position); - assertNull(events); + ignitionEventHandler.analyzePosition(position, Assertions::assertNull); } } diff --git a/src/test/java/org/traccar/handler/events/MaintenanceEventHandlerTest.java b/src/test/java/org/traccar/handler/events/MaintenanceEventHandlerTest.java index 661336d76..f0bd4a3f9 100644 --- a/src/test/java/org/traccar/handler/events/MaintenanceEventHandlerTest.java +++ b/src/test/java/org/traccar/handler/events/MaintenanceEventHandlerTest.java @@ -2,11 +2,14 @@ package org.traccar.handler.events; import org.junit.jupiter.api.Test; import org.traccar.BaseTest; +import org.traccar.model.Event; import org.traccar.model.Maintenance; import org.traccar.model.Position; import org.traccar.session.cache.CacheManager; +import java.util.ArrayList; import java.util.Date; +import java.util.List; import java.util.Set; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -38,24 +41,30 @@ public class MaintenanceEventHandlerTest extends BaseTest { MaintenanceEventHandler eventHandler = new MaintenanceEventHandler(cacheManager); when(maintenance.getStart()).thenReturn(10000.0); - when(maintenance.getPeriod()).thenReturn(2000.0); + when(maintenance.getPeriod()).thenReturn(2000.0); + + List events = new ArrayList<>(); lastPosition.set(Position.KEY_TOTAL_DISTANCE, 1999); - position.set(Position.KEY_TOTAL_DISTANCE, 2001); - assertTrue(eventHandler.analyzePosition(position).isEmpty()); + position.set(Position.KEY_TOTAL_DISTANCE, 2001); + eventHandler.analyzePosition(position, events::add); + assertTrue(events.isEmpty()); lastPosition.set(Position.KEY_TOTAL_DISTANCE, 3999); - position.set(Position.KEY_TOTAL_DISTANCE, 4001); - assertTrue(eventHandler.analyzePosition(position).isEmpty()); + position.set(Position.KEY_TOTAL_DISTANCE, 4001); + eventHandler.analyzePosition(position, events::add); + assertTrue(events.isEmpty()); lastPosition.set(Position.KEY_TOTAL_DISTANCE, 9999); position.set(Position.KEY_TOTAL_DISTANCE, 10001); - assertEquals(1, eventHandler.analyzePosition(position).size()); + eventHandler.analyzePosition(position, events::add); + assertEquals(1, events.size()); lastPosition.set(Position.KEY_TOTAL_DISTANCE, 11999); position.set(Position.KEY_TOTAL_DISTANCE, 12001); - assertEquals(1, eventHandler.analyzePosition(position).size()); - + eventHandler.analyzePosition(position, events::add); + assertEquals(2, events.size()); + } } -- cgit v1.2.3