Implement buffered commands

3 files changed, 46 insertions, 27 deletions

diff --git a/src/org/traccar/api/resource/CommandResource.java b/src/org/traccar/api/resource/CommandResource.java
index 8da9f8447..996c15daf 100644
--- a/src/org/traccar/api/resource/CommandResource.java
+++ b/src/org/traccar/api/resource/CommandResource.java
@@ -62,15 +62,19 @@ public class CommandResource extends ExtendedObjectResource<Command> {
         Context.getPermissionsManager().checkReadonly(getUserId());
         long deviceId = entity.getDeviceId();
         long id = entity.getId();
+        boolean sent;
         if (deviceId != 0 && id != 0) {
             Context.getPermissionsManager().checkPermission(Command.class, getUserId(), id);
             Context.getPermissionsManager().checkDevice(getUserId(), deviceId);
             Context.getPermissionsManager().checkUserDeviceCommand(getUserId(), deviceId, id);
-            Context.getCommandsManager().sendCommand(id, deviceId);
+            sent = Context.getCommandsManager().sendCommand(id, deviceId);
         } else {
             Context.getPermissionsManager().checkLimitCommands(getUserId());
             Context.getPermissionsManager().checkDevice(getUserId(), deviceId);
-            Context.getCommandsManager().sendCommand(entity);
+            sent = Context.getCommandsManager().sendCommand(entity, deviceId);
+        }
+        if (!sent) {
+            return Response.accepted(entity).build();
         }
         return Response.ok(entity).build();
     }
diff --git a/src/org/traccar/database/CommandsManager.java b/src/org/traccar/database/CommandsManager.java
index 521a2e1d1..624fe56cd 100644
--- a/src/org/traccar/database/CommandsManager.java
+++ b/src/org/traccar/database/CommandsManager.java
@@ -21,6 +21,10 @@ import java.lang.reflect.Modifier;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.traccar.BaseProtocol;
 import org.traccar.Context;
@@ -31,26 +35,22 @@ import org.traccar.model.Position;
 
 public class CommandsManager  extends ExtendedObjectManager<Command> {
 
-    private boolean fallbackToText;
+    private final Map<Long, Queue<Command>> deviceQueues = new ConcurrentHashMap<>();
 
     public CommandsManager(DataManager dataManager) {
         super(dataManager, Command.class);
-        fallbackToText = Context.getConfig().getBoolean("command.fallbackToSms");
     }
 
     public boolean checkDeviceCommand(long deviceId, long commandId) {
         return !getAllDeviceItems(deviceId).contains(commandId);
     }
 
-    public void sendCommand(Command command) throws Exception {
-        sendCommand(command, command.getDeviceId(), fallbackToText);
+    public boolean sendCommand(long commandId, long deviceId) throws Exception {
+        return sendCommand(getById(commandId), deviceId);
     }
 
-    public void sendCommand(long commandId, long deviceId) throws Exception {
-        sendCommand(getById(commandId), deviceId, false);
-    }
-
-    public void sendCommand(Command command, long deviceId, boolean fallbackToText) throws Exception {
+    public boolean sendCommand(Command command, long deviceId) throws Exception {
+        boolean sent = true;
         if (command.getTextChannel()) {
             Position lastPosition = Context.getIdentityManager().getLastPosition(deviceId);
             String phone = Context.getIdentityManager().getById(deviceId).getPhone();
@@ -71,32 +71,26 @@ public class CommandsManager  extends ExtendedObjectManager<Command> {
             if (activeDevice != null) {
                 activeDevice.sendCommand(command);
             } else {
-                if (fallbackToText) {
-                    command.setTextChannel(true);
-                    sendCommand(command, deviceId, false);
-                } else {
-                    throw new RuntimeException("Device is not online");
-                }
+                sent = !getDeviceQueue(deviceId).add(command);
             }
         }
+        return sent;
     }
 
     public Collection<Long> getSupportedCommands(long deviceId) {
         List<Long> result = new ArrayList<>();
         Position lastPosition = Context.getIdentityManager().getLastPosition(deviceId);
-        boolean online = Context.getConnectionManager().getActiveDevice(deviceId) != null;
         for (long commandId : getAllDeviceItems(deviceId)) {
             Command command = getById(commandId);
-            if (command.getTextChannel() || online) {
-                if (lastPosition != null) {
-                    BaseProtocol protocol = Context.getServerManager().getProtocol(lastPosition.getProtocol());
-                    if (protocol.getSupportedTextCommands().contains(command.getType())
-                            || online && protocol.getSupportedDataCommands().contains(command.getType())) {
-                        result.add(commandId);
-                    }
-                } else if (command.getType().equals(Command.TYPE_CUSTOM)) {
+            if (lastPosition != null) {
+                BaseProtocol protocol = Context.getServerManager().getProtocol(lastPosition.getProtocol());
+                if ((command.getTextChannel()
+                        ? protocol.getSupportedTextCommands() : protocol.getSupportedDataCommands())
+                        .contains(command.getType())) {
                     result.add(commandId);
                 }
+            } else if (command.getType().equals(Command.TYPE_CUSTOM)) {
+                result.add(commandId);
             }
         }
         return result;
@@ -133,4 +127,22 @@ public class CommandsManager  extends ExtendedObjectManager<Command> {
         return result;
     }
 
+    private Queue<Command> getDeviceQueue(long deviceId) {
+        if (!deviceQueues.containsKey(deviceId)) {
+            deviceQueues.put(deviceId, new ConcurrentLinkedQueue<Command>());
+        }
+        return deviceQueues.get(deviceId);
+    }
+
+    public void sendQueuedCommands(ActiveDevice activeDevice) {
+        Queue<Command> deviceQueue = deviceQueues.get(activeDevice.getDeviceId());
+        if (deviceQueue != null) {
+            Command command = deviceQueue.poll();
+            while (command != null) {
+                activeDevice.sendCommand(command);
+                command = deviceQueue.poll();
+            }
+        }
+    }
+
 }
diff --git a/src/org/traccar/database/ConnectionManager.java b/src/org/traccar/database/ConnectionManager.java
index de11db21b..e5a7a272f 100644
--- a/src/org/traccar/database/ConnectionManager.java
+++ b/src/org/traccar/database/ConnectionManager.java
@@ -57,7 +57,9 @@ public class ConnectionManager {
     }
 
     public void addActiveDevice(long deviceId, Protocol protocol, Channel channel, SocketAddress remoteAddress) {
-        activeDevices.put(deviceId, new ActiveDevice(deviceId, protocol, channel, remoteAddress));
+        ActiveDevice activeDevice = new ActiveDevice(deviceId, protocol, channel, remoteAddress);
+        activeDevices.put(deviceId, activeDevice);
+        Context.getCommandsManager().sendQueuedCommands(activeDevice);
     }
 
     public void removeActiveDevice(Channel channel) {
@@ -122,6 +124,7 @@ public class ConnectionManager {
                 public void run(Timeout timeout) throws Exception {
                     if (!timeout.isCancelled()) {
                         updateDevice(deviceId, Device.STATUS_UNKNOWN, null);
+                        activeDevices.remove(deviceId);
                     }
                 }
             }, deviceTimeout, TimeUnit.MILLISECONDS));