aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/main/java/org/traccar/BaseMqttProtocolDecoder.java96
-rw-r--r--src/main/java/org/traccar/protocol/IotmProtocolDecoder.java147
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;
}
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-04 22:32:42 +0000