From 2a6e75c5d9527cb7b5b54e257fd756e717d5347a Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sat, 30 Apr 2022 18:29:37 -0700 Subject: Implement cluster broadcasting --- src/main/java/org/traccar/Context.java | 11 +++ src/main/java/org/traccar/Main.java | 15 ++-- src/main/java/org/traccar/MainEventHandler.java | 10 +++ .../org/traccar/broadcast/BroadcastMessage.java | 42 +++++++++ .../org/traccar/broadcast/BroadcastService.java | 100 +++++++++++++++++++++ .../java/org/traccar/broadcast/DeviceStatus.java | 52 +++++++++++ src/main/java/org/traccar/config/Keys.java | 14 +++ 7 files changed, 238 insertions(+), 6 deletions(-) create mode 100644 src/main/java/org/traccar/broadcast/BroadcastMessage.java create mode 100644 src/main/java/org/traccar/broadcast/BroadcastService.java create mode 100644 src/main/java/org/traccar/broadcast/DeviceStatus.java (limited to 'src/main/java') diff --git a/src/main/java/org/traccar/Context.java b/src/main/java/org/traccar/Context.java index aeba9c4c9..c44d432b2 100644 --- a/src/main/java/org/traccar/Context.java +++ b/src/main/java/org/traccar/Context.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.datatype.jsr353.JSR353Module; import org.apache.velocity.app.VelocityEngine; import org.eclipse.jetty.util.URIUtil; +import org.traccar.broadcast.BroadcastService; import org.traccar.config.Config; import org.traccar.config.Keys; import org.traccar.database.AttributesManager; @@ -171,6 +172,12 @@ public final class Context { return scheduleManager; } + private static BroadcastService broadcastService; + + public static BroadcastService getBroadcastService() { + return broadcastService; + } + private static GeofenceManager geofenceManager; public static GeofenceManager getGeofenceManager() { @@ -335,6 +342,10 @@ public final class Context { serverManager = new ServerManager(); scheduleManager = new ScheduleManager(); + if (config.hasKey(Keys.BROADCAST_ADDRESS)) { + broadcastService = new BroadcastService(config, objectMapper); + } + if (config.hasKey(Keys.EVENT_FORWARD_URL)) { eventForwarder = new EventForwarder(); } diff --git a/src/main/java/org/traccar/Main.java b/src/main/java/org/traccar/Main.java index 6daf72bb4..2eaf394af 100644 --- a/src/main/java/org/traccar/Main.java +++ b/src/main/java/org/traccar/Main.java @@ -1,5 +1,5 @@ /* - * Copyright 2012 - 2020 Anton Tananaev (anton@traccar.org) + * Copyright 2012 - 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. @@ -118,13 +118,14 @@ public final class Main { List services = new LinkedList<>(); services.add(Context.getServerManager()); - if (Context.getWebServer() != null) { - services.add(Context.getWebServer()); - } + services.add(Context.getWebServer()); services.add(Context.getScheduleManager()); + services.add(Context.getBroadcastService()); for (LifecycleObject service : services) { - service.start(); + if (service != null) { + service.start(); + } } Thread.setDefaultUncaughtExceptionHandler((t, e) -> LOGGER.error("Thread exception", e)); @@ -134,7 +135,9 @@ public final class Main { Collections.reverse(services); for (LifecycleObject service : services) { - service.stop(); + if (service != null) { + service.stop(); + } } })); } catch (Exception e) { diff --git a/src/main/java/org/traccar/MainEventHandler.java b/src/main/java/org/traccar/MainEventHandler.java index a3f6f4105..0e60ac06d 100644 --- a/src/main/java/org/traccar/MainEventHandler.java +++ b/src/main/java/org/traccar/MainEventHandler.java @@ -23,12 +23,14 @@ import io.netty.handler.codec.http.HttpRequestDecoder; import io.netty.handler.timeout.IdleStateEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.traccar.broadcast.BroadcastMessage; import org.traccar.config.Keys; import org.traccar.database.StatisticsManager; import org.traccar.helper.DateUtil; import org.traccar.model.Position; import org.traccar.storage.StorageException; +import java.io.IOException; import java.util.Arrays; import java.util.HashSet; import java.util.LinkedHashSet; @@ -61,6 +63,14 @@ public class MainEventHandler extends ChannelInboundHandlerAdapter { LOGGER.warn("Failed to update device", error); } + BroadcastMessage message = new BroadcastMessage(); + message.setPosition(position); + try { + Context.getBroadcastService().sendMessage(message); + } catch (IOException e) { + e.printStackTrace(); + } + String uniqueId = Context.getIdentityManager().getById(position.getDeviceId()).getUniqueId(); StringBuilder builder = new StringBuilder(); diff --git a/src/main/java/org/traccar/broadcast/BroadcastMessage.java b/src/main/java/org/traccar/broadcast/BroadcastMessage.java new file mode 100644 index 000000000..6b103f373 --- /dev/null +++ b/src/main/java/org/traccar/broadcast/BroadcastMessage.java @@ -0,0 +1,42 @@ +/* + * 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 org.traccar.model.Position; + +public class BroadcastMessage { + + private DeviceStatus deviceStatus; + + public DeviceStatus getDeviceStatus() { + return deviceStatus; + } + + public void setDeviceStatus(DeviceStatus deviceStatus) { + this.deviceStatus = deviceStatus; + } + + private Position position; + + public Position getPosition() { + return position; + } + + public void setPosition(Position position) { + this.position = position; + } + +} diff --git a/src/main/java/org/traccar/broadcast/BroadcastService.java b/src/main/java/org/traccar/broadcast/BroadcastService.java new file mode 100644 index 000000000..b17857fcd --- /dev/null +++ b/src/main/java/org/traccar/broadcast/BroadcastService.java @@ -0,0 +1,100 @@ +/* + * 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.LifecycleObject; +import org.traccar.config.Config; +import org.traccar.config.Keys; + +import javax.inject.Inject; +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.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +public class BroadcastService implements LifecycleObject { + + private static final Logger LOGGER = LoggerFactory.getLogger(BroadcastService.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]; + + @Inject + public BroadcastService(Config config, ObjectMapper objectMapper) throws IOException { + this.objectMapper = objectMapper; + address = InetAddress.getByName(config.getString(Keys.BROADCAST_ADDRESS)); + port = config.getInteger(Keys.BROADCAST_PORT); + } + + public void sendMessage(BroadcastMessage message) throws IOException { + byte[] buffer = objectMapper.writeValueAsString(message).getBytes(StandardCharsets.UTF_8); + DatagramPacket packet = new DatagramPacket(buffer, buffer.length, address, port); + publisherSocket.send(packet); + } + + private void handleMessage(BroadcastMessage message) { + if (message.getDeviceStatus() != null) { + LOGGER.info("Broadcast received device {}", message.getDeviceStatus().getDeviceId()); + } else if (message.getPosition() != null) { + LOGGER.info("Broadcast received position {}", message.getPosition().getDeviceId()); + } + } + + @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() { + @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); + } + } + }; + +} diff --git a/src/main/java/org/traccar/broadcast/DeviceStatus.java b/src/main/java/org/traccar/broadcast/DeviceStatus.java new file mode 100644 index 000000000..4f0143319 --- /dev/null +++ b/src/main/java/org/traccar/broadcast/DeviceStatus.java @@ -0,0 +1,52 @@ +/* + * 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 java.util.Date; + +public class DeviceStatus { + + private long deviceId; + + public long getDeviceId() { + return deviceId; + } + + public void setDeviceId(long deviceId) { + this.deviceId = deviceId; + } + + private String status; + + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + + private Date lastUpdate; + + public Date getLastUpdate() { + return this.lastUpdate; + } + + public void setLastUpdate(Date lastUpdate) { + this.lastUpdate = lastUpdate; + } + +} diff --git a/src/main/java/org/traccar/config/Keys.java b/src/main/java/org/traccar/config/Keys.java index ccfe4bee7..dc6bcbec9 100644 --- a/src/main/java/org/traccar/config/Keys.java +++ b/src/main/java/org/traccar/config/Keys.java @@ -1309,4 +1309,18 @@ public final class Keys { Collections.singletonList(KeyType.GLOBAL), "time,position,speed,course,accuracy,result"); + /** + * Multicast address for broadcasting synchronization events. + */ + public static final ConfigKey BROADCAST_ADDRESS = new ConfigKey<>( + "broadcast.address", + Collections.singletonList(KeyType.GLOBAL)); + + /** + * Multicast port for broadcasting synchronization events. + */ + public static final ConfigKey BROADCAST_PORT = new ConfigKey<>( + "broadcast.port", + Collections.singletonList(KeyType.GLOBAL)); + } -- cgit v1.2.3