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