Extract base MQTT decoder

2 files changed, 141 insertions, 102 deletions

diff --git a/src/main/java/org/traccar/BaseMqttProtocolDecoder.java b/src/main/java/org/traccar/BaseMqttProtocolDecoder.java
new file mode 100644
index 000000000..0388563f5
--- /dev/null
+++ b/src/main/java/org/traccar/BaseMqttProtocolDecoder.java
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import io.netty.channel.Channel;
+import io.netty.handler.codec.mqtt.MqttConnectMessage;
+import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageBuilders;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
+import org.traccar.session.DeviceSession;
+
+import java.net.SocketAddress;
+
+public abstract class BaseMqttProtocolDecoder extends BaseProtocolDecoder {
+
+    public BaseMqttProtocolDecoder(Protocol protocol) {
+        super(protocol);
+    }
+
+    protected abstract Object decode(DeviceSession deviceSession, MqttPublishMessage message) throws Exception;
+
+    @Override
+    protected final Object decode(
+            Channel channel, SocketAddress remoteAddress, Object msg) throws Exception {
+
+        if (msg instanceof MqttConnectMessage) {
+
+            MqttConnectMessage message = (MqttConnectMessage) msg;
+
+            DeviceSession deviceSession = getDeviceSession(
+                    channel, remoteAddress, message.payload().clientIdentifier());
+
+            MqttConnectReturnCode returnCode = deviceSession != null
+                    ? MqttConnectReturnCode.CONNECTION_ACCEPTED
+                    : MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
+
+            MqttMessage response = MqttMessageBuilders.connAck().returnCode(returnCode).build();
+
+            if (channel != null) {
+                channel.writeAndFlush(new NetworkMessage(response, remoteAddress));
+            }
+
+        } else if (msg instanceof MqttSubscribeMessage) {
+
+            MqttSubscribeMessage message = (MqttSubscribeMessage) msg;
+
+            MqttMessage response = MqttMessageBuilders.subAck()
+                    .packetId(message.variableHeader().messageId())
+                    .build();
+
+            if (channel != null) {
+                channel.writeAndFlush(new NetworkMessage(response, remoteAddress));
+            }
+
+        } else if (msg instanceof MqttPublishMessage) {
+
+            DeviceSession deviceSession = getDeviceSession(channel, remoteAddress);
+            if (deviceSession == null) {
+                return null;
+            }
+
+            MqttPublishMessage message = (MqttPublishMessage) msg;
+
+            Object result = decode(deviceSession, message);
+
+            MqttMessage response = MqttMessageBuilders.pubAck()
+                    .packetId(message.variableHeader().packetId())
+                    .build();
+
+            if (channel != null) {
+                channel.writeAndFlush(new NetworkMessage(response, remoteAddress));
+            }
+
+            return result;
+
+        }
+
+        return null;
+    }
+
+}
diff --git a/src/main/java/org/traccar/protocol/IotmProtocolDecoder.java b/src/main/java/org/traccar/protocol/IotmProtocolDecoder.java
index 7bbe6c8de..d9e6670c6 100644
--- a/src/main/java/org/traccar/protocol/IotmProtocolDecoder.java
+++ b/src/main/java/org/traccar/protocol/IotmProtocolDecoder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 - 2022 Anton Tananaev (anton@traccar.org)
+ * Copyright 2020 - 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.
@@ -17,27 +17,19 @@ package org.traccar.protocol;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
-import io.netty.channel.Channel;
-import io.netty.handler.codec.mqtt.MqttConnectMessage;
-import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
-import io.netty.handler.codec.mqtt.MqttMessage;
-import io.netty.handler.codec.mqtt.MqttMessageBuilders;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
-import org.traccar.BaseProtocolDecoder;
-import org.traccar.session.DeviceSession;
-import org.traccar.NetworkMessage;
+import org.traccar.BaseMqttProtocolDecoder;
 import org.traccar.Protocol;
 import org.traccar.helper.UnitsConverter;
 import org.traccar.model.Position;
+import org.traccar.session.DeviceSession;
 
-import java.net.SocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.Date;
 import java.util.LinkedList;
 import java.util.List;
 
-public class IotmProtocolDecoder extends BaseProtocolDecoder {
+public class IotmProtocolDecoder extends BaseMqttProtocolDecoder {
 
     public IotmProtocolDecoder(Protocol protocol) {
         super(protocol);
@@ -236,121 +228,72 @@ public class IotmProtocolDecoder extends BaseProtocolDecoder {
 
     @Override
     protected Object decode(
-            Channel channel, SocketAddress remoteAddress, Object msg) throws Exception {
-
-        if (msg instanceof MqttConnectMessage) {
-
-            MqttConnectMessage message = (MqttConnectMessage) msg;
+            DeviceSession deviceSession, MqttPublishMessage message) throws Exception {
 
-            DeviceSession deviceSession = getDeviceSession(
-                    channel, remoteAddress, message.payload().clientIdentifier());
-
-            MqttConnectReturnCode returnCode = deviceSession != null
-                    ? MqttConnectReturnCode.CONNECTION_ACCEPTED
-                    : MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
-
-            MqttMessage response = MqttMessageBuilders.connAck().returnCode(returnCode).build();
-
-            if (channel != null) {
-                channel.writeAndFlush(new NetworkMessage(response, remoteAddress));
-            }
+        List<Position> positions = new LinkedList<>();
 
-        } else if (msg instanceof MqttSubscribeMessage) {
+        ByteBuf buf = message.payload();
 
-            MqttSubscribeMessage message = (MqttSubscribeMessage) msg;
-
-            MqttMessage response = MqttMessageBuilders.subAck()
-                    .packetId(message.variableHeader().messageId())
-                    .build();
-
-            if (channel != null) {
-                channel.writeAndFlush(new NetworkMessage(response, remoteAddress));
-            }
-
-        } else if (msg instanceof MqttPublishMessage) {
-
-            DeviceSession deviceSession = getDeviceSession(channel, remoteAddress);
-            if (deviceSession == null) {
-                return null;
-            }
+        buf.readUnsignedByte(); // structure version
 
-            List<Position> positions = new LinkedList<>();
+        while (buf.readableBytes() > 1) {
+            int type = buf.readUnsignedByte();
+            int length = buf.readUnsignedShortLE();
+            ByteBuf record = buf.readSlice(length);
+            if (type == 1) {
 
-            MqttPublishMessage message = (MqttPublishMessage) msg;
-            ByteBuf buf = message.payload();
+                Position position = new Position(getProtocolName());
+                position.setDeviceId(deviceSession.getDeviceId());
+                position.setTime(new Date(record.readUnsignedIntLE() * 1000));
 
-            buf.readUnsignedByte(); // structure version
+                while (record.readableBytes() > 0) {
+                    int sensorType = record.readUnsignedByte();
+                    int sensorId = record.readUnsignedShortLE();
+                    if (sensorType == 14) {
 
-            while (buf.readableBytes() > 1) {
-                int type = buf.readUnsignedByte();
-                int length = buf.readUnsignedShortLE();
-                ByteBuf record = buf.readSlice(length);
-                if (type == 1) {
+                        position.setValid(true);
+                        position.setLatitude(record.readFloatLE());
+                        position.setLongitude(record.readFloatLE());
+                        position.setSpeed(UnitsConverter.knotsFromKph(record.readUnsignedShortLE()));
 
-                    Position position = new Position(getProtocolName());
-                    position.setDeviceId(deviceSession.getDeviceId());
-                    position.setTime(new Date(record.readUnsignedIntLE() * 1000));
+                        position.set(Position.KEY_HDOP, record.readUnsignedByte());
+                        position.set(Position.KEY_SATELLITES, record.readUnsignedByte());
 
-                    while (record.readableBytes() > 0) {
-                        int sensorType = record.readUnsignedByte();
-                        int sensorId = record.readUnsignedShortLE();
-                        if (sensorType == 14) {
+                        position.setCourse(record.readUnsignedShortLE());
+                        position.setAltitude(record.readShortLE());
 
-                            position.setValid(true);
-                            position.setLatitude(record.readFloatLE());
-                            position.setLongitude(record.readFloatLE());
-                            position.setSpeed(UnitsConverter.knotsFromKph(record.readUnsignedShortLE()));
-
-                            position.set(Position.KEY_HDOP, record.readUnsignedByte());
-                            position.set(Position.KEY_SATELLITES, record.readUnsignedByte());
-
-                            position.setCourse(record.readUnsignedShortLE());
-                            position.setAltitude(record.readShortLE());
-
-                        } else {
-
-                            if (sensorType == 3) {
-                                continue;
-                            }
-
-                            decodeSensor(position, record, sensorType, sensorId);
+                    } else {
 
+                        if (sensorType == 3) {
+                            continue;
                         }
-                    }
-
-                    positions.add(position);
 
-                } else if (type == 3) {
+                        decodeSensor(position, record, sensorType, sensorId);
 
-                    Position position = new Position(getProtocolName());
-                    position.setDeviceId(deviceSession.getDeviceId());
+                    }
+                }
 
-                    getLastLocation(position, new Date(record.readUnsignedIntLE() * 1000));
+                positions.add(position);
 
-                    record.readUnsignedByte(); // function identifier
+            } else if (type == 3) {
 
-                    position.set(Position.KEY_EVENT, record.readUnsignedByte());
+                Position position = new Position(getProtocolName());
+                position.setDeviceId(deviceSession.getDeviceId());
 
-                    positions.add(position);
+                getLastLocation(position, new Date(record.readUnsignedIntLE() * 1000));
 
-                }
-            }
+                record.readUnsignedByte(); // function identifier
 
-            buf.readUnsignedByte(); // checksum
+                position.set(Position.KEY_EVENT, record.readUnsignedByte());
 
-            MqttMessage response = MqttMessageBuilders.pubAck()
-                    .packetId(message.variableHeader().packetId())
-                    .build();
+                positions.add(position);
 
-            if (channel != null) {
-                channel.writeAndFlush(new NetworkMessage(response, remoteAddress));
             }
-
-            return positions.isEmpty() ? null : positions;
-
         }
 
-        return null;
+        buf.readUnsignedByte(); // checksum
+
+        return positions.isEmpty() ? null : positions;
     }
 
 }