From a2f215fe63bd313bfac2a894c170673ccbc252c4 Mon Sep 17 00:00:00 2001 From: Abyss777 Date: Thu, 19 Oct 2017 10:03:02 +0500 Subject: Implement buffered commands --- src/org/traccar/database/CommandsManager.java | 60 +++++++++++++++---------- src/org/traccar/database/ConnectionManager.java | 5 ++- 2 files changed, 40 insertions(+), 25 deletions(-) (limited to 'src/org/traccar/database') 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 { - private boolean fallbackToText; + private final Map> 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 { 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 getSupportedCommands(long deviceId) { List 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 { return result; } + private Queue getDeviceQueue(long deviceId) { + if (!deviceQueues.containsKey(deviceId)) { + deviceQueues.put(deviceId, new ConcurrentLinkedQueue()); + } + return deviceQueues.get(deviceId); + } + + public void sendQueuedCommands(ActiveDevice activeDevice) { + Queue 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)); -- cgit v1.2.3