From c53d98c668af9c79767e22964f05c7bf7dc866f2 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sat, 25 Jun 2022 13:33:35 -0700 Subject: Integrate broadcast service --- .../broadcast/MulticastBroadcastService.java | 170 +++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 src/main/java/org/traccar/broadcast/MulticastBroadcastService.java (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java new file mode 100644 index 000000000..0525fa742 --- /dev/null +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -0,0 +1,170 @@ +/* + * Copyright 2022 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.broadcast; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.config.Config; +import org.traccar.config.Keys; +import org.traccar.model.BaseModel; +import org.traccar.model.Device; +import org.traccar.model.Event; +import org.traccar.model.Permission; +import org.traccar.model.Position; + +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.net.MulticastSocket; +import java.nio.charset.StandardCharsets; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +public class MulticastBroadcastService implements BroadcastService { + + private static final Logger LOGGER = LoggerFactory.getLogger(MulticastBroadcastService.class); + + private final ObjectMapper objectMapper; + + private final InetAddress address; + private final int port; + + private DatagramSocket publisherSocket; + + private final ExecutorService service = Executors.newSingleThreadExecutor(); + private final byte[] receiverBuffer = new byte[4096]; + + private final Set listeners = new HashSet<>(); + + public MulticastBroadcastService(Config config, ObjectMapper objectMapper) throws IOException { + this.objectMapper = objectMapper; + address = InetAddress.getByName(config.getString(Keys.BROADCAST_ADDRESS)); + port = config.getInteger(Keys.BROADCAST_PORT); + } + + @Override + public void registerListener(BroadcastInterface listener) { + listeners.add(listener); + } + + @Override + public void updateDevice(Device device) { + BroadcastMessage message = new BroadcastMessage(); + message.setDevice(device); + sendMessage(message); + } + + @Override + public void updatePosition(Position position) { + BroadcastMessage message = new BroadcastMessage(); + message.setPosition(position); + sendMessage(message); + } + + @Override + public void updateEvent(long userId, Event event) { + BroadcastMessage message = new BroadcastMessage(); + message.setUserId(userId); + message.setEvent(event); + sendMessage(message); + } + + @Override + public void invalidateObject(Class clazz, long id) { + BroadcastMessage message = new BroadcastMessage(); + message.setChanges(Map.of(Permission.getKey(clazz), id)); + sendMessage(message); + } + + @Override + public void invalidatePermission( + Class clazz1, long id1, + Class clazz2, long id2) { + BroadcastMessage message = new BroadcastMessage(); + message.setChanges(Map.of(Permission.getKey(clazz1), id1, Permission.getKey(clazz2), id2)); + sendMessage(message); + } + + private void sendMessage(BroadcastMessage message) { + try { + byte[] buffer = objectMapper.writeValueAsString(message).getBytes(StandardCharsets.UTF_8); + DatagramPacket packet = new DatagramPacket(buffer, buffer.length, address, port); + publisherSocket.send(packet); + } catch (IOException e) { + LOGGER.warn("Broadcast failed", e); + } + } + + private void handleMessage(BroadcastMessage message) { + if (message.getDevice() != null) { + listeners.forEach(listener -> listener.updateDevice(message.getDevice())); + } else if (message.getPosition() != null) { + listeners.forEach(listener -> listener.updatePosition(message.getPosition())); + } else if (message.getUserId() != null && message.getEvent() != null) { + listeners.forEach(listener -> listener.updateEvent(message.getUserId(), message.getEvent())); + } else if (message.getChanges() != null) { + var iterator = message.getChanges().entrySet().iterator(); + if (iterator.hasNext()) { + var first = iterator.next(); + if (iterator.hasNext()) { + var second = iterator.next(); + listeners.forEach(listener -> listener.invalidatePermission( + Permission.getKeyClass(first.getKey()), first.getValue(), + Permission.getKeyClass(second.getKey()), second.getValue())); + } else { + listeners.forEach(listener -> listener.invalidateObject( + Permission.getKeyClass(first.getKey()), first.getValue())); + } + } + } + } + + @Override + public void start() throws IOException { + service.submit(receiver); + publisherSocket = new DatagramSocket(); + } + + @Override + public void stop() { + publisherSocket.close(); + service.shutdown(); + } + + private final Runnable receiver = new Runnable() { + @SuppressWarnings("deprecation") + @Override + public void run() { + try (MulticastSocket socket = new MulticastSocket(port)) { + socket.joinGroup(address); + while (!service.isShutdown()) { + DatagramPacket packet = new DatagramPacket(receiverBuffer, receiverBuffer.length); + socket.receive(packet); + String data = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); + handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); + } + socket.leaveGroup(address); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + }; +} -- cgit v1.2.3 From 2fcffc5b55f59310d289a21d1ebc2ee6bf15bcd5 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sat, 25 Jun 2022 14:17:00 -0700 Subject: Fix connection users issue --- src/main/java/org/traccar/MainEventHandler.java | 4 +- src/main/java/org/traccar/api/AsyncSocket.java | 7 +-- .../org/traccar/api/resource/DeviceResource.java | 4 +- .../org/traccar/broadcast/BroadcastInterface.java | 6 +- .../broadcast/MulticastBroadcastService.java | 12 ++-- .../handler/events/GeofenceEventHandler.java | 2 +- .../org/traccar/notificators/NotificatorWeb.java | 4 +- .../org/traccar/session/ConnectionManager.java | 66 +++++++++++++++------- 8 files changed, 65 insertions(+), 40 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/MainEventHandler.java b/src/main/java/org/traccar/MainEventHandler.java index 0a8c69b54..e2cad15c6 100644 --- a/src/main/java/org/traccar/MainEventHandler.java +++ b/src/main/java/org/traccar/MainEventHandler.java @@ -92,8 +92,8 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter { new Condition.Equals("id", "id"))); cacheManager.updatePosition(position); - connectionManager.updatePosition(position); - broadcastService.updatePosition(position); + connectionManager.updatePosition(true, position); + broadcastService.updatePosition(true, position); } } catch (StorageException error) { LOGGER.warn("Failed to update device", error); diff --git a/src/main/java/org/traccar/api/AsyncSocket.java b/src/main/java/org/traccar/api/AsyncSocket.java index 40aa68e88..5fc4b4412 100644 --- a/src/main/java/org/traccar/api/AsyncSocket.java +++ b/src/main/java/org/traccar/api/AsyncSocket.java @@ -58,15 +58,14 @@ public class AsyncSocket extends WebSocketAdapter implements ConnectionManager.U public void onWebSocketConnect(Session session) { super.onWebSocketConnect(session); - Map> data = new HashMap<>(); try { + Map> data = new HashMap<>(); data.put(KEY_POSITIONS, PositionUtil.getLatestPositions(storage, userId)); + sendData(data); + connectionManager.addListener(userId, this); } catch (StorageException e) { throw new RuntimeException(e); } - sendData(data); - - connectionManager.addListener(userId, this); } @Override diff --git a/src/main/java/org/traccar/api/resource/DeviceResource.java b/src/main/java/org/traccar/api/resource/DeviceResource.java index 2b673a108..e205f2d28 100644 --- a/src/main/java/org/traccar/api/resource/DeviceResource.java +++ b/src/main/java/org/traccar/api/resource/DeviceResource.java @@ -137,8 +137,8 @@ public class DeviceResource extends BaseObjectResource { try { cacheManager.addDevice(position.getDeviceId()); cacheManager.updatePosition(position); - connectionManager.updatePosition(position); - broadcastService.updatePosition(position); + connectionManager.updatePosition(true, position); + broadcastService.updatePosition(true, position); } finally { cacheManager.removeDevice(position.getDeviceId()); } diff --git a/src/main/java/org/traccar/broadcast/BroadcastInterface.java b/src/main/java/org/traccar/broadcast/BroadcastInterface.java index d5b49f213..69e610dc6 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastInterface.java +++ b/src/main/java/org/traccar/broadcast/BroadcastInterface.java @@ -22,13 +22,13 @@ import org.traccar.model.Position; public interface BroadcastInterface { - default void updateDevice(Device device) { + default void updateDevice(boolean local, Device device) { } - default void updatePosition(Position position) { + default void updatePosition(boolean local, Position position) { } - default void updateEvent(long userId, Event event) { + default void updateEvent(boolean local, long userId, Event event) { } default void invalidateObject(Class clazz, long id) { diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index 0525fa742..ac0fcbd86 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -66,21 +66,21 @@ public class MulticastBroadcastService implements BroadcastService { } @Override - public void updateDevice(Device device) { + public void updateDevice(boolean local, Device device) { BroadcastMessage message = new BroadcastMessage(); message.setDevice(device); sendMessage(message); } @Override - public void updatePosition(Position position) { + public void updatePosition(boolean local, Position position) { BroadcastMessage message = new BroadcastMessage(); message.setPosition(position); sendMessage(message); } @Override - public void updateEvent(long userId, Event event) { + public void updateEvent(boolean local, long userId, Event event) { BroadcastMessage message = new BroadcastMessage(); message.setUserId(userId); message.setEvent(event); @@ -115,11 +115,11 @@ public class MulticastBroadcastService implements BroadcastService { private void handleMessage(BroadcastMessage message) { if (message.getDevice() != null) { - listeners.forEach(listener -> listener.updateDevice(message.getDevice())); + listeners.forEach(listener -> listener.updateDevice(false, message.getDevice())); } else if (message.getPosition() != null) { - listeners.forEach(listener -> listener.updatePosition(message.getPosition())); + listeners.forEach(listener -> listener.updatePosition(false, message.getPosition())); } else if (message.getUserId() != null && message.getEvent() != null) { - listeners.forEach(listener -> listener.updateEvent(message.getUserId(), message.getEvent())); + listeners.forEach(listener -> listener.updateEvent(false, message.getUserId(), message.getEvent())); } else if (message.getChanges() != null) { var iterator = message.getChanges().entrySet().iterator(); if (iterator.hasNext()) { diff --git a/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java b/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java index 0a924cfc3..ca3fc3f89 100644 --- a/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java +++ b/src/main/java/org/traccar/handler/events/GeofenceEventHandler.java @@ -68,7 +68,7 @@ public class GeofenceEventHandler extends BaseEventHandler { device.setGeofenceIds(currentGeofences); if (!oldGeofences.isEmpty() || !newGeofences.isEmpty()) { - connectionManager.updateDevice(device); + connectionManager.updateDevice(true, device); } Map events = new HashMap<>(); diff --git a/src/main/java/org/traccar/notificators/NotificatorWeb.java b/src/main/java/org/traccar/notificators/NotificatorWeb.java index efbbf24cc..061018ba9 100644 --- a/src/main/java/org/traccar/notificators/NotificatorWeb.java +++ b/src/main/java/org/traccar/notificators/NotificatorWeb.java @@ -56,8 +56,8 @@ public final class NotificatorWeb implements Notificator { var message = notificationFormatter.formatMessage(user, event, position, "short"); copy.set("message", message.getBody()); - connectionManager.updateEvent(user.getId(), copy); - broadcastService.updateEvent(user.getId(), copy); + connectionManager.updateEvent(true, user.getId(), copy); + broadcastService.updateEvent(true, user.getId(), copy); } } diff --git a/src/main/java/org/traccar/session/ConnectionManager.java b/src/main/java/org/traccar/session/ConnectionManager.java index 74427c08b..0e0ea1eb8 100644 --- a/src/main/java/org/traccar/session/ConnectionManager.java +++ b/src/main/java/org/traccar/session/ConnectionManager.java @@ -30,6 +30,7 @@ import org.traccar.database.NotificationManager; import org.traccar.handler.events.MotionEventHandler; import org.traccar.handler.events.OverspeedEventHandler; import org.traccar.helper.model.AttributeUtil; +import org.traccar.model.BaseModel; import org.traccar.model.Device; import org.traccar.model.Event; import org.traccar.model.Position; @@ -45,6 +46,7 @@ import javax.inject.Inject; import javax.inject.Singleton; import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.HashSet; @@ -52,6 +54,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; @Singleton public class ConnectionManager implements BroadcastInterface { @@ -74,7 +77,10 @@ public class ConnectionManager implements BroadcastInterface { private final Timer timer; private final BroadcastService broadcastService; - private final Map> listeners = new ConcurrentHashMap<>(); + private final Map> listeners = new HashMap<>(); + private final Map> userDevices = new HashMap<>(); + private final Map> deviceUsers = new HashMap<>(); + private final Map timeouts = new ConcurrentHashMap<>(); @Inject @@ -197,6 +203,10 @@ public class ConnectionManager implements BroadcastInterface { public void deviceUnknown(long deviceId) { updateDevice(deviceId, Device.STATUS_UNKNOWN, null); + removeDeviceSession(deviceId); + } + + private void removeDeviceSession(long deviceId) { DeviceSession deviceSession = sessionsByDeviceId.remove(deviceId); if (deviceSession != null) { cacheManager.removeDevice(deviceId); @@ -274,8 +284,8 @@ public class ConnectionManager implements BroadcastInterface { LOGGER.warn("Update device status error", e); } - updateDevice(device); - broadcastService.updateDevice(device); + updateDevice(true, device); + broadcastService.updateDevice(true, device); } public DeviceState getDeviceState(long deviceId) { @@ -313,10 +323,14 @@ public class ConnectionManager implements BroadcastInterface { } @Override - public synchronized void updateDevice(Device device) { - for (User user : cacheManager.getDeviceObjects(device.getId(), User.class)) { - if (listeners.containsKey(user.getId())) { - for (UpdateListener listener : listeners.get(user.getId())) { + public synchronized void updateDevice(boolean local, Device device) { + if (!local && Device.STATUS_ONLINE.equals(device.getStatus())) { + timeouts.remove(device.getId()); + removeDeviceSession(device.getId()); + } + for (long userId : deviceUsers.getOrDefault(device.getId(), Collections.emptySet())) { + if (listeners.containsKey(userId)) { + for (UpdateListener listener : listeners.get(userId)) { listener.onUpdateDevice(device); } } @@ -324,11 +338,10 @@ public class ConnectionManager implements BroadcastInterface { } @Override - public synchronized void updatePosition(Position position) { - long deviceId = position.getDeviceId(); - for (User user : cacheManager.getDeviceObjects(deviceId, User.class)) { - if (listeners.containsKey(user.getId())) { - for (UpdateListener listener : listeners.get(user.getId())) { + public synchronized void updatePosition(boolean local, Position position) { + for (long userId : deviceUsers.getOrDefault(position.getDeviceId(), Collections.emptySet())) { + if (listeners.containsKey(userId)) { + for (UpdateListener listener : listeners.get(userId)) { listener.onUpdatePosition(position); } } @@ -336,7 +349,7 @@ public class ConnectionManager implements BroadcastInterface { } @Override - public synchronized void updateEvent(long userId, Event event) { + public synchronized void updateEvent(boolean local, long userId, Event event) { if (listeners.containsKey(userId)) { for (UpdateListener listener : listeners.get(userId)) { listener.onUpdateEvent(event); @@ -351,18 +364,31 @@ public class ConnectionManager implements BroadcastInterface { void onUpdateEvent(Event event); } - public synchronized void addListener(long userId, UpdateListener listener) { - if (!listeners.containsKey(userId)) { - listeners.put(userId, new HashSet<>()); + public synchronized void addListener(long userId, UpdateListener listener) throws StorageException { + var set = listeners.get(userId); + if (set == null) { + set = new HashSet<>(); + listeners.put(userId, set); + + var devices = storage.getObjects(Device.class, new Request( + new Columns.Include("id"), new Condition.Permission(User.class, userId, Device.class))); + userDevices.put(userId, devices.stream().map(BaseModel::getId).collect(Collectors.toUnmodifiableSet())); + devices.forEach(device -> deviceUsers.computeIfAbsent(device.getId(), id -> new HashSet<>()).add(userId)); } - listeners.get(userId).add(listener); + set.add(listener); } public synchronized void removeListener(long userId, UpdateListener listener) { - if (!listeners.containsKey(userId)) { - listeners.put(userId, new HashSet<>()); + var set = listeners.get(userId); + set.remove(listener); + if (set.isEmpty()) { + listeners.remove(userId); + + userDevices.remove(userId).forEach(deviceId -> deviceUsers.computeIfPresent(deviceId, (x, userIds) -> { + userIds.remove(userId); + return userIds.isEmpty() ? null : userIds; + })); } - listeners.get(userId).remove(listener); } } -- cgit v1.2.3 From 95ba3446d936213683d604eb545710124b9bf689 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sat, 2 Jul 2022 09:06:49 -0700 Subject: Add broadcast logging --- src/main/java/org/traccar/broadcast/MulticastBroadcastService.java | 1 + 1 file changed, 1 insertion(+) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index ac0fcbd86..81b6cd5ec 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -159,6 +159,7 @@ public class MulticastBroadcastService implements BroadcastService { DatagramPacket packet = new DatagramPacket(receiverBuffer, receiverBuffer.length); socket.receive(packet); String data = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); + LOGGER.info("Broadcast received: {}", data); handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); } socket.leaveGroup(address); -- cgit v1.2.3 From 65cb239b78d3e75db565309b0196378e29a439d4 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sun, 10 Jul 2022 10:25:25 -0700 Subject: Specify multicast interface --- .../broadcast/MulticastBroadcastService.java | 22 ++++++++++++++++------ src/main/java/org/traccar/config/Keys.java | 7 +++++++ 2 files changed, 23 insertions(+), 6 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index 81b6cd5ec..877008eda 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -30,7 +30,9 @@ import java.io.IOException; import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.MulticastSocket; +import java.net.NetworkInterface; import java.nio.charset.StandardCharsets; import java.util.HashSet; import java.util.Map; @@ -44,8 +46,9 @@ public class MulticastBroadcastService implements BroadcastService { private final ObjectMapper objectMapper; - private final InetAddress address; + private final NetworkInterface networkInterface; private final int port; + private final InetSocketAddress group; private DatagramSocket publisherSocket; @@ -56,8 +59,15 @@ public class MulticastBroadcastService implements BroadcastService { public MulticastBroadcastService(Config config, ObjectMapper objectMapper) throws IOException { this.objectMapper = objectMapper; - address = InetAddress.getByName(config.getString(Keys.BROADCAST_ADDRESS)); port = config.getInteger(Keys.BROADCAST_PORT); + String interfaceName = config.getString(Keys.BROADCAST_INTERFACE); + if (interfaceName.indexOf('.') >= 0 || interfaceName.indexOf(':') >= 0) { + networkInterface = NetworkInterface.getByInetAddress(InetAddress.getByName(interfaceName)); + } else { + networkInterface = NetworkInterface.getByName(interfaceName); + } + InetAddress address = InetAddress.getByName(config.getString(Keys.BROADCAST_ADDRESS)); + group = new InetSocketAddress(address, port); } @Override @@ -106,7 +116,7 @@ public class MulticastBroadcastService implements BroadcastService { private void sendMessage(BroadcastMessage message) { try { byte[] buffer = objectMapper.writeValueAsString(message).getBytes(StandardCharsets.UTF_8); - DatagramPacket packet = new DatagramPacket(buffer, buffer.length, address, port); + DatagramPacket packet = new DatagramPacket(buffer, buffer.length, group); publisherSocket.send(packet); } catch (IOException e) { LOGGER.warn("Broadcast failed", e); @@ -150,11 +160,10 @@ public class MulticastBroadcastService implements BroadcastService { } private final Runnable receiver = new Runnable() { - @SuppressWarnings("deprecation") @Override public void run() { try (MulticastSocket socket = new MulticastSocket(port)) { - socket.joinGroup(address); + socket.joinGroup(group, networkInterface); while (!service.isShutdown()) { DatagramPacket packet = new DatagramPacket(receiverBuffer, receiverBuffer.length); socket.receive(packet); @@ -162,10 +171,11 @@ public class MulticastBroadcastService implements BroadcastService { LOGGER.info("Broadcast received: {}", data); handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); } - socket.leaveGroup(address); + socket.leaveGroup(group, networkInterface); } catch (IOException e) { throw new RuntimeException(e); } } }; + } diff --git a/src/main/java/org/traccar/config/Keys.java b/src/main/java/org/traccar/config/Keys.java index 8e11b4013..6f19ee863 100644 --- a/src/main/java/org/traccar/config/Keys.java +++ b/src/main/java/org/traccar/config/Keys.java @@ -1444,6 +1444,13 @@ public final class Keys { List.of(KeyType.CONFIG), "time,position,speed,course,accuracy,result"); + /** + * Multicast interface. It can be either an IP address or an interface name. + */ + public static final ConfigKey BROADCAST_INTERFACE = new StringConfigKey( + "broadcast.interface", + List.of(KeyType.CONFIG)); + /** * Multicast address for broadcasting synchronization events. */ -- cgit v1.2.3 From e5fed94589cff4614b248cb3408397febc0b1be7 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Mon, 11 Jul 2022 07:20:34 -0700 Subject: Use same socket for sending --- src/main/java/org/traccar/broadcast/MulticastBroadcastService.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index 877008eda..134bba797 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -150,12 +150,10 @@ public class MulticastBroadcastService implements BroadcastService { @Override public void start() throws IOException { service.submit(receiver); - publisherSocket = new DatagramSocket(); } @Override public void stop() { - publisherSocket.close(); service.shutdown(); } @@ -163,6 +161,7 @@ public class MulticastBroadcastService implements BroadcastService { @Override public void run() { try (MulticastSocket socket = new MulticastSocket(port)) { + publisherSocket = socket; socket.joinGroup(group, networkInterface); while (!service.isShutdown()) { DatagramPacket packet = new DatagramPacket(receiverBuffer, receiverBuffer.length); @@ -172,6 +171,7 @@ public class MulticastBroadcastService implements BroadcastService { handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); } socket.leaveGroup(group, networkInterface); + publisherSocket = null; } catch (IOException e) { throw new RuntimeException(e); } -- cgit v1.2.3 From 767a449904b3d154ded4c5facefc73d7e33650f1 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Mon, 11 Jul 2022 17:58:35 -0700 Subject: Interface for sending multicast --- src/main/java/org/traccar/broadcast/MulticastBroadcastService.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index 134bba797..c4ba0e059 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -161,17 +161,17 @@ public class MulticastBroadcastService implements BroadcastService { @Override public void run() { try (MulticastSocket socket = new MulticastSocket(port)) { - publisherSocket = socket; + socket.setNetworkInterface(networkInterface); socket.joinGroup(group, networkInterface); + publisherSocket = socket; while (!service.isShutdown()) { DatagramPacket packet = new DatagramPacket(receiverBuffer, receiverBuffer.length); socket.receive(packet); String data = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); - LOGGER.info("Broadcast received: {}", data); handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); } - socket.leaveGroup(group, networkInterface); publisherSocket = null; + socket.leaveGroup(group, networkInterface); } catch (IOException e) { throw new RuntimeException(e); } -- cgit v1.2.3 From 859b9d4aad68f4dafdee60ff14d3c9c0b32693a0 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Wed, 13 Jul 2022 07:18:16 -0700 Subject: Ignore own multicast packets --- debug.xml | 5 +++-- src/main/java/org/traccar/broadcast/MulticastBroadcastService.java | 6 ++++-- 2 files changed, 7 insertions(+), 4 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/debug.xml b/debug.xml index debbd06b7..a597d83c5 100644 --- a/debug.xml +++ b/debug.xml @@ -24,7 +24,8 @@ true 6037 - + diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index c4ba0e059..be24989dd 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -167,8 +167,10 @@ public class MulticastBroadcastService implements BroadcastService { while (!service.isShutdown()) { DatagramPacket packet = new DatagramPacket(receiverBuffer, receiverBuffer.length); socket.receive(packet); - String data = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); - handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); + if (networkInterface.inetAddresses().noneMatch(a -> a.equals(packet.getAddress()))) { + String data = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); + handleMessage(objectMapper.readValue(data, BroadcastMessage.class)); + } } publisherSocket = null; socket.leaveGroup(group, networkInterface); -- cgit v1.2.3 From a8e38b74e5fc6789676bf35a9b92594a230e3ad8 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Wed, 13 Jul 2022 17:31:52 -0700 Subject: Avoid broadcast loops (fix #4894) --- src/main/java/org/traccar/api/BaseObjectResource.java | 6 +++--- .../org/traccar/api/resource/PermissionsResource.java | 2 ++ .../java/org/traccar/api/resource/ServerResource.java | 2 +- .../java/org/traccar/broadcast/BroadcastInterface.java | 3 ++- .../org/traccar/broadcast/MulticastBroadcastService.java | 5 ++++- src/main/java/org/traccar/session/ConnectionManager.java | 1 + src/main/java/org/traccar/session/cache/CacheManager.java | 15 ++++++++++----- 7 files changed, 23 insertions(+), 11 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/api/BaseObjectResource.java b/src/main/java/org/traccar/api/BaseObjectResource.java index d10843917..2a3bbe239 100644 --- a/src/main/java/org/traccar/api/BaseObjectResource.java +++ b/src/main/java/org/traccar/api/BaseObjectResource.java @@ -71,8 +71,8 @@ public abstract class BaseObjectResource extends BaseResour entity.setId(storage.addObject(entity, new Request(new Columns.Exclude("id")))); LogAction.create(getUserId(), entity); storage.addPermission(new Permission(User.class, getUserId(), baseClass, entity.getId())); - cacheManager.invalidatePermission(User.class, getUserId(), baseClass, entity.getId()); - connectionManager.invalidatePermission(User.class, getUserId(), baseClass, entity.getId()); + cacheManager.invalidatePermission(true, User.class, getUserId(), baseClass, entity.getId()); + connectionManager.invalidatePermission(true, User.class, getUserId(), baseClass, entity.getId()); LogAction.link(getUserId(), User.class, getUserId(), baseClass, entity.getId()); return Response.ok(entity).build(); @@ -98,7 +98,7 @@ public abstract class BaseObjectResource extends BaseResour storage.updateObject(entity, new Request( new Columns.Exclude("id"), new Condition.Equals("id", "id"))); - cacheManager.updateOrInvalidate(entity); + cacheManager.updateOrInvalidate(true, entity); LogAction.edit(getUserId(), entity); return Response.ok(entity).build(); diff --git a/src/main/java/org/traccar/api/resource/PermissionsResource.java b/src/main/java/org/traccar/api/resource/PermissionsResource.java index 44fc897ac..d35cb98bb 100644 --- a/src/main/java/org/traccar/api/resource/PermissionsResource.java +++ b/src/main/java/org/traccar/api/resource/PermissionsResource.java @@ -72,6 +72,7 @@ public class PermissionsResource extends BaseResource { checkPermission(permission); storage.addPermission(permission); cacheManager.invalidatePermission( + true, permission.getOwnerClass(), permission.getOwnerId(), permission.getPropertyClass(), permission.getPropertyId()); LogAction.link(getUserId(), @@ -96,6 +97,7 @@ public class PermissionsResource extends BaseResource { checkPermission(permission); storage.removePermission(permission); cacheManager.invalidatePermission( + true, permission.getOwnerClass(), permission.getOwnerId(), permission.getPropertyClass(), permission.getPropertyId()); LogAction.unlink(getUserId(), diff --git a/src/main/java/org/traccar/api/resource/ServerResource.java b/src/main/java/org/traccar/api/resource/ServerResource.java index 3e6792f5b..4fc76a0d7 100644 --- a/src/main/java/org/traccar/api/resource/ServerResource.java +++ b/src/main/java/org/traccar/api/resource/ServerResource.java @@ -77,7 +77,7 @@ public class ServerResource extends BaseResource { storage.updateObject(entity, new Request( new Columns.Exclude("id"), new Condition.Equals("id", "id"))); - cacheManager.updateOrInvalidate(entity); + cacheManager.updateOrInvalidate(true, entity); LogAction.edit(getUserId(), entity); return Response.ok(entity).build(); } diff --git a/src/main/java/org/traccar/broadcast/BroadcastInterface.java b/src/main/java/org/traccar/broadcast/BroadcastInterface.java index 69e610dc6..dddba68b6 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastInterface.java +++ b/src/main/java/org/traccar/broadcast/BroadcastInterface.java @@ -31,10 +31,11 @@ public interface BroadcastInterface { default void updateEvent(boolean local, long userId, Event event) { } - default void invalidateObject(Class clazz, long id) { + default void invalidateObject(boolean local, Class clazz, long id) { } default void invalidatePermission( + boolean local, Class clazz1, long id1, Class clazz2, long id2) { } diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index be24989dd..3eafe07d3 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -98,7 +98,7 @@ public class MulticastBroadcastService implements BroadcastService { } @Override - public void invalidateObject(Class clazz, long id) { + public void invalidateObject(boolean local, Class clazz, long id) { BroadcastMessage message = new BroadcastMessage(); message.setChanges(Map.of(Permission.getKey(clazz), id)); sendMessage(message); @@ -106,6 +106,7 @@ public class MulticastBroadcastService implements BroadcastService { @Override public void invalidatePermission( + boolean local, Class clazz1, long id1, Class clazz2, long id2) { BroadcastMessage message = new BroadcastMessage(); @@ -137,10 +138,12 @@ public class MulticastBroadcastService implements BroadcastService { if (iterator.hasNext()) { var second = iterator.next(); listeners.forEach(listener -> listener.invalidatePermission( + false, Permission.getKeyClass(first.getKey()), first.getValue(), Permission.getKeyClass(second.getKey()), second.getValue())); } else { listeners.forEach(listener -> listener.invalidateObject( + false, Permission.getKeyClass(first.getKey()), first.getValue())); } } diff --git a/src/main/java/org/traccar/session/ConnectionManager.java b/src/main/java/org/traccar/session/ConnectionManager.java index 62fdd833b..9888cca2b 100644 --- a/src/main/java/org/traccar/session/ConnectionManager.java +++ b/src/main/java/org/traccar/session/ConnectionManager.java @@ -364,6 +364,7 @@ public class ConnectionManager implements BroadcastInterface { @Override public synchronized void invalidatePermission( + boolean local, Class clazz1, long id1, Class clazz2, long id2) { if (clazz1.equals(User.class) && clazz2.equals(Device.class)) { diff --git a/src/main/java/org/traccar/session/cache/CacheManager.java b/src/main/java/org/traccar/session/cache/CacheManager.java index 4e99161dd..58eb95327 100644 --- a/src/main/java/org/traccar/session/cache/CacheManager.java +++ b/src/main/java/org/traccar/session/cache/CacheManager.java @@ -188,12 +188,12 @@ public class CacheManager implements BroadcastInterface { } @Override - public void invalidateObject(Class clazz, long id) { + public void invalidateObject(boolean local, Class clazz, long id) { try { var object = storage.getObject(clazz, new Request( new Columns.All(), new Condition.Equals("id", "id", id))); if (object != null) { - updateOrInvalidate(object); + updateOrInvalidate(local, object); } else { invalidate(clazz, id); } @@ -202,8 +202,10 @@ public class CacheManager implements BroadcastInterface { } } - public void updateOrInvalidate(T object) throws StorageException { - broadcastService.invalidateObject(object.getClass(), object.getId()); + public void updateOrInvalidate(boolean local, T object) throws StorageException { + if (local) { + broadcastService.invalidateObject(true, object.getClass(), object.getId()); + } boolean invalidate = false; var before = getObject(object.getClass(), object.getId()); @@ -232,9 +234,12 @@ public class CacheManager implements BroadcastInterface { @Override public void invalidatePermission( + boolean local, Class clazz1, long id1, Class clazz2, long id2) { - broadcastService.invalidatePermission(clazz1, id1, clazz2, id2); + if (local) { + broadcastService.invalidatePermission(true, clazz1, id1, clazz2, id2); + } try { invalidate(new CacheKey(clazz1, id1), new CacheKey(clazz2, id2)); -- cgit v1.2.3 From c4509720c5dab9d7ba17dd6e89f0d846052b0c39 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Mon, 18 Jul 2022 20:47:11 -0700 Subject: Reset start on restart --- src/main/java/org/traccar/Main.java | 6 ++++ .../org/traccar/broadcast/BroadcastService.java | 1 + .../broadcast/MulticastBroadcastService.java | 5 ++++ .../traccar/broadcast/NullBroadcastService.java | 5 ++++ .../java/org/traccar/helper/model/DeviceUtil.java | 33 ++++++++++++++++++++++ 5 files changed, 50 insertions(+) create mode 100644 src/main/java/org/traccar/helper/model/DeviceUtil.java (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/Main.java b/src/main/java/org/traccar/Main.java index 6a968ac7e..e34fbb72a 100644 --- a/src/main/java/org/traccar/Main.java +++ b/src/main/java/org/traccar/Main.java @@ -20,8 +20,10 @@ import com.google.inject.Injector; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.broadcast.BroadcastService; +import org.traccar.helper.model.DeviceUtil; import org.traccar.schedule.ScheduleManager; import org.traccar.storage.DatabaseModule; +import org.traccar.storage.Storage; import org.traccar.web.WebModule; import org.traccar.web.WebServer; @@ -120,6 +122,10 @@ public final class Main { LOGGER.info("Version: " + Main.class.getPackage().getImplementationVersion()); LOGGER.info("Starting server..."); + if (injector.getInstance(BroadcastService.class).singleInstance()) { + DeviceUtil.resetStatus(injector.getInstance(Storage.class)); + } + var services = Stream.of( ServerManager.class, WebServer.class, ScheduleManager.class, BroadcastService.class) .map(injector::getInstance) diff --git a/src/main/java/org/traccar/broadcast/BroadcastService.java b/src/main/java/org/traccar/broadcast/BroadcastService.java index 8a2e4bafc..a86c43b5b 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastService.java +++ b/src/main/java/org/traccar/broadcast/BroadcastService.java @@ -18,5 +18,6 @@ package org.traccar.broadcast; import org.traccar.LifecycleObject; public interface BroadcastService extends LifecycleObject, BroadcastInterface { + boolean singleInstance(); void registerListener(BroadcastInterface listener); } diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index 3eafe07d3..be65b7826 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -70,6 +70,11 @@ public class MulticastBroadcastService implements BroadcastService { group = new InetSocketAddress(address, port); } + @Override + public boolean singleInstance() { + return false; + } + @Override public void registerListener(BroadcastInterface listener) { listeners.add(listener); diff --git a/src/main/java/org/traccar/broadcast/NullBroadcastService.java b/src/main/java/org/traccar/broadcast/NullBroadcastService.java index 3f41299db..f95037990 100644 --- a/src/main/java/org/traccar/broadcast/NullBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/NullBroadcastService.java @@ -17,6 +17,11 @@ package org.traccar.broadcast; public class NullBroadcastService implements BroadcastService { + @Override + public boolean singleInstance() { + return true; + } + @Override public void registerListener(BroadcastInterface listener) { } diff --git a/src/main/java/org/traccar/helper/model/DeviceUtil.java b/src/main/java/org/traccar/helper/model/DeviceUtil.java new file mode 100644 index 000000000..597078caf --- /dev/null +++ b/src/main/java/org/traccar/helper/model/DeviceUtil.java @@ -0,0 +1,33 @@ +/* + * Copyright 2022 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.helper.model; + +import org.traccar.model.Device; +import org.traccar.storage.Storage; +import org.traccar.storage.StorageException; +import org.traccar.storage.query.Columns; +import org.traccar.storage.query.Request; + +public final class DeviceUtil { + + private DeviceUtil() { + } + + public static void resetStatus(Storage storage) throws StorageException { + storage.updateObject(new Device(), new Request(new Columns.Include("status"))); + } + +} -- cgit v1.2.3 From 0a853f2aa3556554acd0b43a5008c43c345fa300 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Thu, 29 Sep 2022 17:57:22 -0700 Subject: Synchronize queued commands --- .../org/traccar/broadcast/BroadcastInterface.java | 3 +++ .../org/traccar/broadcast/BroadcastMessage.java | 10 ++++++++++ .../broadcast/MulticastBroadcastService.java | 9 +++++++++ .../java/org/traccar/database/CommandsManager.java | 22 ++++++++++++++++++++-- 4 files changed, 42 insertions(+), 2 deletions(-) (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/broadcast/BroadcastInterface.java b/src/main/java/org/traccar/broadcast/BroadcastInterface.java index dddba68b6..673ebd8b8 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastInterface.java +++ b/src/main/java/org/traccar/broadcast/BroadcastInterface.java @@ -31,6 +31,9 @@ public interface BroadcastInterface { default void updateEvent(boolean local, long userId, Event event) { } + default void updateCommand(boolean local, long deviceId) { + } + default void invalidateObject(boolean local, Class clazz, long id) { } diff --git a/src/main/java/org/traccar/broadcast/BroadcastMessage.java b/src/main/java/org/traccar/broadcast/BroadcastMessage.java index 3e22be7e0..985848d04 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastMessage.java +++ b/src/main/java/org/traccar/broadcast/BroadcastMessage.java @@ -63,6 +63,16 @@ public class BroadcastMessage { this.event = event; } + private Long commandDeviceId; + + public Long getCommandDeviceId() { + return commandDeviceId; + } + + public void setCommandDeviceId(Long commandDeviceId) { + this.commandDeviceId = commandDeviceId; + } + private Map changes; public Map getChanges() { diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index be65b7826..b1b66f1e3 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -102,6 +102,13 @@ public class MulticastBroadcastService implements BroadcastService { sendMessage(message); } + @Override + public void updateCommand(boolean local, long deviceId) { + BroadcastMessage message = new BroadcastMessage(); + message.setCommandDeviceId(deviceId); + sendMessage(message); + } + @Override public void invalidateObject(boolean local, Class clazz, long id) { BroadcastMessage message = new BroadcastMessage(); @@ -136,6 +143,8 @@ public class MulticastBroadcastService implements BroadcastService { listeners.forEach(listener -> listener.updatePosition(false, message.getPosition())); } else if (message.getUserId() != null && message.getEvent() != null) { listeners.forEach(listener -> listener.updateEvent(false, message.getUserId(), message.getEvent())); + } else if (message.getCommandDeviceId() != null) { + listeners.forEach(listener -> listener.updateCommand(false, message.getCommandDeviceId())); } else if (message.getChanges() != null) { var iterator = message.getChanges().entrySet().iterator(); if (iterator.hasNext()) { diff --git a/src/main/java/org/traccar/database/CommandsManager.java b/src/main/java/org/traccar/database/CommandsManager.java index d56b4d472..53040ad53 100644 --- a/src/main/java/org/traccar/database/CommandsManager.java +++ b/src/main/java/org/traccar/database/CommandsManager.java @@ -18,6 +18,8 @@ package org.traccar.database; import org.traccar.BaseProtocol; import org.traccar.ServerManager; +import org.traccar.broadcast.BroadcastInterface; +import org.traccar.broadcast.BroadcastService; import org.traccar.model.Command; import org.traccar.model.Device; import org.traccar.model.Position; @@ -40,21 +42,24 @@ import java.util.Collection; import java.util.stream.Collectors; @Singleton -public class CommandsManager { +public class CommandsManager implements BroadcastInterface { private final Storage storage; private final ServerManager serverManager; private final SmsManager smsManager; private final ConnectionManager connectionManager; + private final BroadcastService broadcastService; @Inject public CommandsManager( Storage storage, ServerManager serverManager, @Nullable SmsManager smsManager, - ConnectionManager connectionManager) { + ConnectionManager connectionManager, BroadcastService broadcastService) { this.storage = storage; this.serverManager = serverManager; this.smsManager = smsManager; this.connectionManager = connectionManager; + this.broadcastService = broadcastService; + broadcastService.registerListener(this); } public boolean sendCommand(Command command) throws Exception { @@ -81,6 +86,7 @@ public class CommandsManager { deviceSession.sendCommand(command); } else { storage.addObject(QueuedCommand.fromCommand(command), new Request(new Columns.Exclude("id"))); + broadcastService.updateCommand(true, deviceId); return false; } } @@ -108,4 +114,16 @@ public class CommandsManager { } } + @Override + public void updateCommand(boolean local, long deviceId) { + if (!local) { + DeviceSession deviceSession = connectionManager.getDeviceSession(deviceId); + if (deviceSession != null && deviceSession.supportsLiveCommands()) { + for (Command command : readQueuedCommands(deviceId)) { + deviceSession.sendCommand(command); + } + } + } + } + } -- cgit v1.2.3 From 3397d4f07ef7fae909c0d6258dbc1263c4acc68b Mon Sep 17 00:00:00 2001 From: "Rafael E. Ajuria" Date: Thu, 8 Jun 2023 01:52:09 +0000 Subject: Add BaseBroadcastService pr comment adjustments --- src/main/java/org/traccar/MainModule.java | 2 +- .../traccar/broadcast/BaseBroadcastService.java | 118 ++++++++++++++ .../broadcast/MulticastBroadcastService.java | 92 +---------- .../traccar/broadcast/RedisBroadcastService.java | 176 ++++++--------------- src/main/java/org/traccar/config/Keys.java | 2 +- 5 files changed, 172 insertions(+), 218 deletions(-) create mode 100644 src/main/java/org/traccar/broadcast/BaseBroadcastService.java (limited to 'src/main/java/org/traccar/broadcast/MulticastBroadcastService.java') diff --git a/src/main/java/org/traccar/MainModule.java b/src/main/java/org/traccar/MainModule.java index b1a8006ee..29d846154 100644 --- a/src/main/java/org/traccar/MainModule.java +++ b/src/main/java/org/traccar/MainModule.java @@ -349,7 +349,7 @@ public class MainModule extends AbstractModule { case "redis": return new RedisBroadcastService(config, objectMapper); default: - return new NullBroadcastService(); + break; } } return new NullBroadcastService(); diff --git a/src/main/java/org/traccar/broadcast/BaseBroadcastService.java b/src/main/java/org/traccar/broadcast/BaseBroadcastService.java new file mode 100644 index 000000000..1ed639dfd --- /dev/null +++ b/src/main/java/org/traccar/broadcast/BaseBroadcastService.java @@ -0,0 +1,118 @@ +/* + * 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.broadcast; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import org.traccar.model.BaseModel; +import org.traccar.model.Device; +import org.traccar.model.Event; +import org.traccar.model.Permission; +import org.traccar.model.Position; + +public abstract class BaseBroadcastService implements BroadcastService { + + private final Set listeners = new HashSet<>(); + + @Override + public boolean singleInstance() { + return true; + } + + @Override + public void registerListener(BroadcastInterface listener) { + listeners.add(listener); + } + + @Override + public void updateDevice(boolean local, Device device) { + BroadcastMessage message = new BroadcastMessage(); + message.setDevice(device); + sendMessage(message); + } + + @Override + public void updatePosition(boolean local, Position position) { + BroadcastMessage message = new BroadcastMessage(); + message.setPosition(position); + sendMessage(message); + } + + @Override + public void updateEvent(boolean local, long userId, Event event) { + BroadcastMessage message = new BroadcastMessage(); + message.setUserId(userId); + message.setEvent(event); + sendMessage(message); + } + + @Override + public void updateCommand(boolean local, long deviceId) { + BroadcastMessage message = new BroadcastMessage(); + message.setCommandDeviceId(deviceId); + sendMessage(message); + } + + @Override + public void invalidateObject(boolean local, Class clazz, long id) { + BroadcastMessage message = new BroadcastMessage(); + message.setChanges(Map.of(Permission.getKey(clazz), id)); + sendMessage(message); + } + + @Override + public void invalidatePermission( + boolean local, + Class clazz1, long id1, + Class clazz2, long id2) { + BroadcastMessage message = new BroadcastMessage(); + message.setChanges(Map.of(Permission.getKey(clazz1), id1, Permission.getKey(clazz2), id2)); + sendMessage(message); + } + + protected abstract void sendMessage(BroadcastMessage message); + + protected void handleMessage(BroadcastMessage message) { + if (message.getDevice() != null) { + listeners.forEach(listener -> listener.updateDevice(false, message.getDevice())); + } else if (message.getPosition() != null) { + listeners.forEach(listener -> listener.updatePosition(false, message.getPosition())); + } else if (message.getUserId() != null && message.getEvent() != null) { + listeners.forEach(listener -> listener.updateEvent(false, message.getUserId(), message.getEvent())); + } else if (message.getCommandDeviceId() != null) { + listeners.forEach(listener -> listener.updateCommand(false, message.getCommandDeviceId())); + } else if (message.getChanges() != null) { + var iterator = message.getChanges().entrySet().iterator(); + if (iterator.hasNext()) { + var first = iterator.next(); + if (iterator.hasNext()) { + var second = iterator.next(); + listeners.forEach(listener -> listener.invalidatePermission( + false, + Permission.getKeyClass(first.getKey()), first.getValue(), + Permission.getKeyClass(second.getKey()), second.getValue())); + } else { + listeners.forEach(listener -> listener.invalidateObject( + false, + Permission.getKeyClass(first.getKey()), first.getValue())); + } + } + } + } + +} \ No newline at end of file diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index b1b66f1e3..1c02b319b 100644 --- a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java @@ -20,11 +20,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Config; import org.traccar.config.Keys; -import org.traccar.model.BaseModel; -import org.traccar.model.Device; -import org.traccar.model.Event; -import org.traccar.model.Permission; -import org.traccar.model.Position; import java.io.IOException; import java.net.DatagramPacket; @@ -34,13 +29,10 @@ import java.net.InetSocketAddress; import java.net.MulticastSocket; import java.net.NetworkInterface; import java.nio.charset.StandardCharsets; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -public class MulticastBroadcastService implements BroadcastService { +public class MulticastBroadcastService extends BaseBroadcastService { private static final Logger LOGGER = LoggerFactory.getLogger(MulticastBroadcastService.class); @@ -55,8 +47,6 @@ public class MulticastBroadcastService implements BroadcastService { private final ExecutorService service = Executors.newSingleThreadExecutor(); private final byte[] receiverBuffer = new byte[4096]; - private final Set listeners = new HashSet<>(); - public MulticastBroadcastService(Config config, ObjectMapper objectMapper) throws IOException { this.objectMapper = objectMapper; port = config.getInteger(Keys.BROADCAST_PORT); @@ -76,57 +66,7 @@ public class MulticastBroadcastService implements BroadcastService { } @Override - public void registerListener(BroadcastInterface listener) { - listeners.add(listener); - } - - @Override - public void updateDevice(boolean local, Device device) { - BroadcastMessage message = new BroadcastMessage(); - message.setDevice(device); - sendMessage(message); - } - - @Override - public void updatePosition(boolean local, Position position) { - BroadcastMessage message = new BroadcastMessage(); - message.setPosition(position); - sendMessage(message); - } - - @Override - public void updateEvent(boolean local, long userId, Event event) { - BroadcastMessage message = new BroadcastMessage(); - message.setUserId(userId); - message.setEvent(event); - sendMessage(message); - } - - @Override - public void updateCommand(boolean local, long deviceId) { - BroadcastMessage message = new BroadcastMessage(); - message.setCommandDeviceId(deviceId); - sendMessage(message); - } - - @Override - public void invalidateObject(boolean local, Class clazz, long id) { - BroadcastMessage message = new BroadcastMessage(); - message.setChanges(Map.of(Permission.getKey(clazz), id)); - sendMessage(message); - } - - @Override - public void invalidatePermission( - boolean local, - Class clazz1, long id1, - Class clazz2, long id2) { - BroadcastMessage message = new BroadcastMessage(); - message.setChanges(Map.of(Permission.getKey(clazz1), id1, Permission.getKey(clazz2), id2)); - sendMessage(message); - } - - private void sendMessage(BroadcastMessage message) { + protected void sendMessage(BroadcastMessage message) { try { byte[] buffer = objectMapper.writeValueAsString(message).getBytes(StandardCharsets.UTF_8); DatagramPacket packet = new DatagramPacket(buffer, buffer.length, group); @@ -136,34 +76,6 @@ public class MulticastBroadcastService implements BroadcastService { } } - private void handleMessage(BroadcastMessage message) { - if (message.getDevice() != null) { - listeners.forEach(listener -> listener.updateDevice(false, message.getDevice())); - } else if (message.getPosition() != null) { - listeners.forEach(listener -> listener.updatePosition(false, message.getPosition())); - } else if (message.getUserId() != null && message.getEvent() != null) { - listeners.forEach(listener -> listener.updateEvent(false, message.getUserId(), message.getEvent())); - } else if (message.getCommandDeviceId() != null) { - listeners.forEach(listener -> listener.updateCommand(false, message.getCommandDeviceId())); - } else if (message.getChanges() != null) { - var iterator = message.getChanges().entrySet().iterator(); - if (iterator.hasNext()) { - var first = iterator.next(); - if (iterator.hasNext()) { - var second = iterator.next(); - listeners.forEach(listener -> listener.invalidatePermission( - false, - Permission.getKeyClass(first.getKey()), first.getValue(), - Permission.getKeyClass(second.getKey()), second.getValue())); - } else { - listeners.forEach(listener -> listener.invalidateObject( - false, - Permission.getKeyClass(first.getKey()), first.getValue())); - } - } - } - } - @Override public void start() throws IOException { service.submit(receiver); diff --git a/src/main/java/org/traccar/broadcast/RedisBroadcastService.java b/src/main/java/org/traccar/broadcast/RedisBroadcastService.java index a6968c894..e619fef60 100644 --- a/src/main/java/org/traccar/broadcast/RedisBroadcastService.java +++ b/src/main/java/org/traccar/broadcast/RedisBroadcastService.java @@ -20,23 +20,17 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.traccar.config.Config; import org.traccar.config.Keys; -import org.traccar.model.BaseModel; -import org.traccar.model.Device; -import org.traccar.model.Event; -import org.traccar.model.Permission; -import org.traccar.model.Position; import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; +import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPubSub; +import redis.clients.jedis.exceptions.JedisConnectionException; +import redis.clients.jedis.exceptions.JedisException; -public class RedisBroadcastService implements BroadcastService { +public class RedisBroadcastService extends BaseBroadcastService { private static final Logger LOGGER = LoggerFactory.getLogger(RedisBroadcastService.class); @@ -44,25 +38,25 @@ public class RedisBroadcastService implements BroadcastService { private final ExecutorService service = Executors.newSingleThreadExecutor(); - private final Set listeners = new HashSet<>(); - private final String url; - private final String pubsubChannel = "traccar:cast"; + private final String channel = "traccar"; - private final Jedis subscriberJedis; - private final Jedis publisherJedis; + private Jedis subscriber; + private Jedis publisher; - private final String id; + private final String id = UUID.randomUUID().toString(); public RedisBroadcastService(Config config, ObjectMapper objectMapper) throws IOException { this.objectMapper = objectMapper; url = config.getString(Keys.BROADCAST_ADDRESS); - subscriberJedis = new Jedis(url); - publisherJedis = new Jedis(url); - - // id that will be used to identify this instance of the server - id = String.valueOf(System.currentTimeMillis()); + try { + subscriber = new Jedis(url); + publisher = new Jedis(url); + subscriber.connect(); + } catch (JedisConnectionException e) { + throw new IOException(e); + } } @Override @@ -71,90 +65,14 @@ public class RedisBroadcastService implements BroadcastService { } @Override - public void registerListener(BroadcastInterface listener) { - listeners.add(listener); - } - - @Override - public void updateDevice(boolean local, Device device) { - BroadcastMessage message = new BroadcastMessage(); - message.setDevice(device); - sendMessage(message); - } - - @Override - public void updatePosition(boolean local, Position position) { - BroadcastMessage message = new BroadcastMessage(); - message.setPosition(position); - sendMessage(message); - } - - @Override - public void updateEvent(boolean local, long userId, Event event) { - BroadcastMessage message = new BroadcastMessage(); - message.setUserId(userId); - message.setEvent(event); - sendMessage(message); - } - - @Override - public void updateCommand(boolean local, long deviceId) { - BroadcastMessage message = new BroadcastMessage(); - message.setCommandDeviceId(deviceId); - sendMessage(message); - } - - @Override - public void invalidateObject(boolean local, Class clazz, long id) { - BroadcastMessage message = new BroadcastMessage(); - message.setChanges(Map.of(Permission.getKey(clazz), id)); - sendMessage(message); - } - - @Override - public void invalidatePermission( - boolean local, - Class clazz1, long id1, - Class clazz2, long id2) { - BroadcastMessage message = new BroadcastMessage(); - message.setChanges(Map.of(Permission.getKey(clazz1), id1, Permission.getKey(clazz2), id2)); - sendMessage(message); - } - - private void sendMessage(BroadcastMessage message) { + protected void sendMessage(BroadcastMessage message) { try { String payload = id + ":" + objectMapper.writeValueAsString(message); - publisherJedis.publish(pubsubChannel, payload); + publisher.publish(channel, payload); } catch (IOException e) { LOGGER.warn("Broadcast failed", e); - } - } - - private void handleMessage(BroadcastMessage message) { - if (message.getDevice() != null) { - listeners.forEach(listener -> listener.updateDevice(false, message.getDevice())); - } else if (message.getPosition() != null) { - listeners.forEach(listener -> listener.updatePosition(false, message.getPosition())); - } else if (message.getUserId() != null && message.getEvent() != null) { - listeners.forEach(listener -> listener.updateEvent(false, message.getUserId(), message.getEvent())); - } else if (message.getCommandDeviceId() != null) { - listeners.forEach(listener -> listener.updateCommand(false, message.getCommandDeviceId())); - } else if (message.getChanges() != null) { - var iterator = message.getChanges().entrySet().iterator(); - if (iterator.hasNext()) { - var first = iterator.next(); - if (iterator.hasNext()) { - var second = iterator.next(); - listeners.forEach(listener -> listener.invalidatePermission( - false, - Permission.getKeyClass(first.getKey()), first.getValue(), - Permission.getKeyClass(second.getKey()), second.getValue())); - } else { - listeners.forEach(listener -> listener.invalidateObject( - false, - Permission.getKeyClass(first.getKey()), first.getValue())); - } - } + } catch (JedisConnectionException e) { + LOGGER.warn("Broadcast failed", e); } } @@ -165,39 +83,45 @@ public class RedisBroadcastService implements BroadcastService { @Override public void stop() { + try { + if (subscriber != null) { + subscriber.close(); + subscriber = null; + } + } catch (JedisException e) { + LOGGER.warn("Subscriber close failed", e); + } + try { + if (publisher != null) { + publisher.close(); + publisher = null; + } + } catch (JedisException e) { + LOGGER.warn("Publisher close failed", e); + } service.shutdown(); } private final Runnable receiver = new Runnable() { @Override public void run() { - subscriberJedis.subscribe(new JedisPubSub() { - @Override - public void onMessage(String channel, String message) { - try { - String[] parts = message.split(":", 2); - if (channel == pubsubChannel && parts.length == 2 && !id.equals(parts[0])) { - handleMessage(objectMapper.readValue(parts[1], BroadcastMessage.class)); + try { + subscriber.subscribe(new JedisPubSub() { + @Override + public void onMessage(String messageChannel, String message) { + try { + String[] parts = message.split(":", 2); + if (messageChannel == channel && parts.length == 2 && !id.equals(parts[0])) { + handleMessage(objectMapper.readValue(parts[1], BroadcastMessage.class)); + } + } catch (IOException e) { + LOGGER.warn("Broadcast handleMessage failed", e); } - } catch (IOException e) { - LOGGER.warn("Broadcast handleMessage failed", e); } - } - }, pubsubChannel); - - while (!service.isShutdown()) { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - break; - } - } - - try { - subscriberJedis.close(); - publisherJedis.close(); - } catch (Exception e) { - LOGGER.warn("Failed to close pubsub", e); + }, channel); + } catch (JedisConnectionException e) { + throw new RuntimeException(e); + } catch (JedisException e) { throw new RuntimeException(e); } } diff --git a/src/main/java/org/traccar/config/Keys.java b/src/main/java/org/traccar/config/Keys.java index 6f42e8937..381e3a108 100644 --- a/src/main/java/org/traccar/config/Keys.java +++ b/src/main/java/org/traccar/config/Keys.java @@ -1771,7 +1771,7 @@ public final class Keys { List.of(KeyType.CONFIG)); /** - * Multicast address for broadcasting synchronization events. + * Multicast address or Redis URL for broadcasting synchronization events. */ public static final ConfigKey BROADCAST_ADDRESS = new StringConfigKey( "broadcast.address", -- cgit v1.2.3