Improve progress protocol

2 files changed, 164 insertions, 102 deletions

diff --git a/src/org/traccar/ServerManager.java b/src/org/traccar/ServerManager.java
index b1cd51040..5a9a4cc8b 100644
--- a/src/org/traccar/ServerManager.java
+++ b/src/org/traccar/ServerManager.java
@@ -17,6 +17,7 @@ package org.traccar;
 
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.nio.ByteOrder;
 import java.sql.SQLException;
 import java.text.DateFormat;
 import java.text.FieldPosition;
@@ -407,13 +408,15 @@ public class ServerManager {
 
     private void initProgressServer(String protocol) throws SQLException {
         if (isProtocolEnabled(properties, protocol)) {
-            serverList.add(new TrackerServer(this, new ServerBootstrap(), protocol) {
+            TrackerServer server = new TrackerServer(this, new ServerBootstrap(), protocol) {
                 @Override
                 protected void addSpecificHandlers(ChannelPipeline pipeline) {
                     pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1024, 2, 2, 0, 0));
                     pipeline.addLast("objectDecoder", new ProgressProtocolDecoder(ServerManager.this));
                 }
-            });
+            };
+            server.setEndianness(ByteOrder.LITTLE_ENDIAN);
+            serverList.add(server);
         }
     }
 
@@ -508,13 +511,15 @@ public class ServerManager {
 
     private void initNavisServer(String protocol) throws SQLException {
         if (isProtocolEnabled(properties, protocol)) {
-            serverList.add(new TrackerServer(this, new ServerBootstrap(), protocol) {
+            TrackerServer server = new TrackerServer(this, new ServerBootstrap(), protocol) {
                 @Override
                 protected void addSpecificHandlers(ChannelPipeline pipeline) {
                     pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(4 * 1024, 12, 2, 2, 0));
                     pipeline.addLast("objectDecoder", new NavisProtocolDecoder(ServerManager.this));
                 }
-            });
+            };
+            server.setEndianness(ByteOrder.LITTLE_ENDIAN);
+            serverList.add(server);
         }
     }
 
diff --git a/src/org/traccar/protocol/ProgressProtocolDecoder.java b/src/org/traccar/protocol/ProgressProtocolDecoder.java
index 06904711e..2282d690b 100644
--- a/src/org/traccar/protocol/ProgressProtocolDecoder.java
+++ b/src/org/traccar/protocol/ProgressProtocolDecoder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2012 Anton Tananaev (anton.tananaev@gmail.com)
+ * Copyright 2012 - 2013 Anton Tananaev (anton.tananaev@gmail.com)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,8 +15,14 @@
  */
 package org.traccar.protocol;
 
+import java.nio.ByteOrder;
 import java.nio.charset.Charset;
+import java.sql.ResultSet;
+import java.sql.Statement;
 import java.util.Calendar;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
 import java.util.TimeZone;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
@@ -24,7 +30,10 @@ import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.traccar.BaseProtocolDecoder;
 import org.traccar.ServerManager;
+import org.traccar.helper.AdvancedConnection;
 import org.traccar.helper.Log;
+import org.traccar.helper.NamedParameterStatement;
+import org.traccar.model.Device;
 import org.traccar.model.Position;
 
 /**
@@ -32,14 +41,9 @@ import org.traccar.model.Position;
  */
 public class ProgressProtocolDecoder extends BaseProtocolDecoder {
 
-    /**
-     * Device ID
-     */
     private long deviceId;
+    private long lastIndex;
 
-    /**
-     * Initialize
-     */
     public ProgressProtocolDecoder(ServerManager serverManager) {
         super(serverManager);
     }
@@ -60,6 +64,41 @@ public class ProgressProtocolDecoder extends BaseProtocolDecoder {
     private static final String HEX_CHARS = "0123456789ABCDEF";
 
     /**
+     * Hack to load last index
+     */
+    /*private void loadLastIndex() {
+        try {
+            Properties p = getServerManager().getProperties();
+            AdvancedConnection connection = new AdvancedConnection(
+                    p.getProperty("database.url"), p.getProperty("database.user"), p.getProperty("database.password"));
+            NamedParameterStatement queryLastIndex = new NamedParameterStatement(connection, p.getProperty("database.selectLastIndex"));
+            queryLastIndex.prepare();
+            queryLastIndex.setLong("device_id", deviceId);
+            ResultSet result = queryLastIndex.executeQuery();
+            if (result.next()) {
+                lastIndex = result.getLong(1);
+            }
+        } catch(Exception error) {
+        }
+    }*/
+
+    /**
+     * Request archive messages
+     */
+    private void requestArchive(Channel channel, long newIndex) {
+        if (lastIndex == 0) {
+            lastIndex = newIndex;
+        } else if (newIndex > lastIndex) {
+            ChannelBuffer request = ChannelBuffers.directBuffer(ByteOrder.BIG_ENDIAN, 12);
+            request.writeShort(MSG_LOG_SYNC);
+            request.writeShort(4);
+            request.writeInt((int) lastIndex);
+            request.writeInt(0);
+            channel.write(request);
+        }
+    }
+
+    /**
      * Decode message
      */
     @Override
@@ -69,12 +108,12 @@ public class ProgressProtocolDecoder extends BaseProtocolDecoder {
 
         ChannelBuffer buf = (ChannelBuffer) msg;
         int type = buf.readUnsignedShort();
-        int length = buf.readUnsignedShort();
+        buf.readUnsignedShort(); // length
 
         // Authentication
         if (type == MSG_IDENT || type == MSG_IDENT_FULL) {
             long id = buf.readUnsignedInt();
-            length = buf.readUnsignedShort();
+            int length = buf.readUnsignedShort();
             buf.skipBytes(length);
             length = buf.readUnsignedShort();
             buf.skipBytes(length);
@@ -82,113 +121,131 @@ public class ProgressProtocolDecoder extends BaseProtocolDecoder {
             String imei = buf.readBytes(length).toString(Charset.defaultCharset());
             try {
                 deviceId = getDataManager().getDeviceByImei(imei).getId();
+                //loadLastIndex();
             } catch(Exception error) {
                 Log.warning("Unknown device - " + imei + " (id - " + id + ")");
             }
         }
 
         // Position
-        else if (deviceId != 0 && (type == MSG_POINT || type == MSG_ALARM)) {
-            Position position = new Position();
-            StringBuilder extendedInfo = new StringBuilder("<protocol>progress</protocol>");
-            position.setDeviceId(deviceId);
-
-            // Message index
-            position.setId(buf.readUnsignedInt());
-
-            // Time
-            Calendar time = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
-            time.clear();
-            time.setTimeInMillis(buf.readUnsignedInt() * 1000);
-            position.setTime(time.getTime());
-
-            // Latitude
-            position.setLatitude(((double) buf.readInt()) / 0x7FFFFFFF * 180.0);
-
-            // Longitude
-            position.setLongitude(((double) buf.readInt()) / 0x7FFFFFFF * 180.0);
-
-            // Speed
-            position.setSpeed(((double) buf.readUnsignedInt()) / 100);
-
-            // Course
-            position.setCourse(((double) buf.readUnsignedShort()) / 100);
-
-            // Altitude
-            position.setAltitude(((double) buf.readUnsignedShort()) / 100);
-
-            // Satellites
-            int satellitesNumber = buf.readUnsignedByte();
-            extendedInfo.append("<satellites>");
-            extendedInfo.append(satellitesNumber);
-            extendedInfo.append("</satellites>");
-
-            // Validity
-            position.setValid(satellitesNumber >= 3); // TODO: probably wrong
-
-            // Cell signal
-            extendedInfo.append("<gsm>");
-            extendedInfo.append(buf.readUnsignedByte());
-            extendedInfo.append("</gsm>");
-
-            // Milage
-            extendedInfo.append("<milage>");
-            extendedInfo.append(buf.readUnsignedInt());
-            extendedInfo.append("</milage>");
-
-            long extraFlags = buf.readLong();
-
-            // Analog inputs
-            if ((extraFlags & 0x1) == 0x1) {
-                int count = buf.readUnsignedShort();
-                for (int i = 1; i <= count; i++) {
-                    extendedInfo.append("<adc").append(i).append(">");
-                    extendedInfo.append(buf.readUnsignedShort());
-                    extendedInfo.append("</adc").append(i).append(">");
-                }
+        else if (deviceId != 0 && (type == MSG_POINT || type == MSG_ALARM || type == MSG_LOGMSG)) {
+            List<Position> positions = new LinkedList<Position>();
 
+            int recordCount = 1;
+            if (type == MSG_LOGMSG) {
+                recordCount = buf.readUnsignedShort();
             }
 
-            // CAN adapter
-            if ((extraFlags & 0x2) == 0x2) {
-                int size = buf.readUnsignedShort();
-                extendedInfo.append("<can>");
-                extendedInfo.append(buf.toString(buf.readerIndex(), size, Charset.defaultCharset()));
-                extendedInfo.append("</can>");
-                buf.skipBytes(size);
-            }
+            for (int j = 0; j < recordCount; j++) {
+                Position position = new Position();
+                StringBuilder extendedInfo = new StringBuilder("<protocol>progress</protocol>");
+                position.setDeviceId(deviceId);
+
+                // Message index
+                if (type == MSG_LOGMSG) {
+                    lastIndex = buf.readUnsignedInt();
+                } else {
+                    requestArchive(channel, buf.readUnsignedInt());
+                }
+
+                // Time
+                Calendar time = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
+                time.clear();
+                time.setTimeInMillis(buf.readUnsignedInt() * 1000);
+                position.setTime(time.getTime());
 
-            // Passenger sensor
-            if ((extraFlags & 0x4) == 0x4) {
-                int size = buf.readUnsignedShort();
+                // Latitude
+                position.setLatitude(((double) buf.readInt()) / 0x7FFFFFFF * 180.0);
+
+                // Longitude
+                position.setLongitude(((double) buf.readInt()) / 0x7FFFFFFF * 180.0);
+
+                // Speed
+                position.setSpeed(((double) buf.readUnsignedInt()) / 100);
+
+                // Course
+                position.setCourse(((double) buf.readUnsignedShort()) / 100);
+
+                // Altitude
+                position.setAltitude(((double) buf.readUnsignedShort()) / 100);
+
+                // Satellites
+                int satellitesNumber = buf.readUnsignedByte();
+                extendedInfo.append("<satellites>");
+                extendedInfo.append(satellitesNumber);
+                extendedInfo.append("</satellites>");
+
+                // Validity
+                position.setValid(satellitesNumber >= 3); // TODO: probably wrong
+
+                // Cell signal
+                extendedInfo.append("<gsm>");
+                extendedInfo.append(buf.readUnsignedByte());
+                extendedInfo.append("</gsm>");
+
+                // Milage
+                extendedInfo.append("<milage>");
+                extendedInfo.append(buf.readUnsignedInt());
+                extendedInfo.append("</milage>");
+
+                long extraFlags = buf.readLong();
+
+                // Analog inputs
+                if ((extraFlags & 0x1) == 0x1) {
+                    int count = buf.readUnsignedShort();
+                    for (int i = 1; i <= count; i++) {
+                        extendedInfo.append("<adc").append(i).append(">");
+                        extendedInfo.append(buf.readUnsignedShort());
+                        extendedInfo.append("</adc").append(i).append(">");
+                    }
 
-                // Convert binary data to hex
-                StringBuilder hex = new StringBuilder();
-                for (int i = buf.readerIndex(); i < buf.readerIndex() + size; i++) {
-                    byte b = buf.getByte(i);
-                    hex.append(HEX_CHARS.charAt((b & 0xf0) >> 4));
-                    hex.append(HEX_CHARS.charAt((b & 0x0F)));
                 }
 
-                extendedInfo.append("<passenger>");
-                extendedInfo.append(hex);
-                extendedInfo.append("</passenger>");
+                // CAN adapter
+                if ((extraFlags & 0x2) == 0x2) {
+                    int size = buf.readUnsignedShort();
+                    extendedInfo.append("<can>");
+                    extendedInfo.append(buf.toString(buf.readerIndex(), size, Charset.defaultCharset()));
+                    extendedInfo.append("</can>");
+                    buf.skipBytes(size);
+                }
 
-                buf.skipBytes(size);
-            }
+                // Passenger sensor
+                if ((extraFlags & 0x4) == 0x4) {
+                    int size = buf.readUnsignedShort();
 
-            // Send response for alarm message
-            if (type == MSG_ALARM) {
-                byte[] response = {(byte)0xC9,0x00,0x00,0x00,0x00,0x00,0x00,0x00};
-                channel.write(ChannelBuffers.wrappedBuffer(response));
+                    // Convert binary data to hex
+                    StringBuilder hex = new StringBuilder();
+                    for (int i = buf.readerIndex(); i < buf.readerIndex() + size; i++) {
+                        byte b = buf.getByte(i);
+                        hex.append(HEX_CHARS.charAt((b & 0xf0) >> 4));
+                        hex.append(HEX_CHARS.charAt((b & 0x0F)));
+                    }
 
-                extendedInfo.append("<alarm>true</alarm>");
-            }
+                    extendedInfo.append("<passenger>");
+                    extendedInfo.append(hex);
+                    extendedInfo.append("</passenger>");
+
+                    buf.skipBytes(size);
+                }
 
-            // Extended info
-            position.setExtendedInfo(extendedInfo.toString());
+                // Send response for alarm message
+                if (type == MSG_ALARM) {
+                    byte[] response = {(byte)0xC9,0x00,0x00,0x00,0x00,0x00,0x00,0x00};
+                    channel.write(ChannelBuffers.wrappedBuffer(response));
+
+                    extendedInfo.append("<alarm>true</alarm>");
+                } else if (type == MSG_LOGMSG) {
+                    extendedInfo.append("<archive>true</archive>");
+                }
+
+                // Extended info
+                position.setExtendedInfo(extendedInfo.toString());
+
+                positions.add(position);
+            }
 
-            return position;
+            return positions;
         }
 
         return null;