diff options
Diffstat (limited to 'src/main/java/org/traccar/broadcast')
5 files changed, 369 insertions, 105 deletions
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..01b212c60 --- /dev/null +++ b/src/main/java/org/traccar/broadcast/BaseBroadcastService.java @@ -0,0 +1,128 @@ +/* + * 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.Set; + +import org.traccar.model.BaseModel; +import org.traccar.model.Device; +import org.traccar.model.Event; +import org.traccar.model.ObjectOperation; +import org.traccar.model.Permission; +import org.traccar.model.Position; + +public abstract class BaseBroadcastService implements BroadcastService { + + private final Set<BroadcastInterface> 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 <T extends BaseModel> void invalidateObject( + boolean local, Class<T> clazz, long id, ObjectOperation operation) { + BroadcastMessage message = new BroadcastMessage(); + var invalidateObject = new BroadcastMessage.InvalidateObject(); + invalidateObject.setClazz(Permission.getKey(clazz)); + invalidateObject.setId(id); + invalidateObject.setOperation(operation); + message.setInvalidateObject(invalidateObject); + sendMessage(message); + } + + @Override + public synchronized <T1 extends BaseModel, T2 extends BaseModel> void invalidatePermission( + boolean local, Class<T1> clazz1, long id1, Class<T2> clazz2, long id2, boolean link) { + BroadcastMessage message = new BroadcastMessage(); + var invalidatePermission = new BroadcastMessage.InvalidatePermission(); + invalidatePermission.setClazz1(Permission.getKey(clazz1)); + invalidatePermission.setId1(id1); + invalidatePermission.setClazz2(Permission.getKey(clazz2)); + invalidatePermission.setId2(id2); + invalidatePermission.setLink(link); + message.setInvalidatePermission(invalidatePermission); + sendMessage(message); + } + + protected abstract void sendMessage(BroadcastMessage message); + + protected void handleMessage(BroadcastMessage message) throws Exception { + 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.getInvalidateObject() != null) { + var invalidateObject = message.getInvalidateObject(); + for (BroadcastInterface listener : listeners) { + listener.invalidateObject( + false, + Permission.getKeyClass(invalidateObject.getClazz()), invalidateObject.getId(), + invalidateObject.getOperation()); + } + } else if (message.getInvalidatePermission() != null) { + var invalidatePermission = message.getInvalidatePermission(); + for (BroadcastInterface listener : listeners) { + listener.invalidatePermission( + false, + Permission.getKeyClass(invalidatePermission.getClazz1()), invalidatePermission.getId1(), + Permission.getKeyClass(invalidatePermission.getClazz2()), invalidatePermission.getId2(), + invalidatePermission.getLink()); + } + } + } + +} diff --git a/src/main/java/org/traccar/broadcast/BroadcastInterface.java b/src/main/java/org/traccar/broadcast/BroadcastInterface.java index 673ebd8b8..d0a491cd2 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastInterface.java +++ b/src/main/java/org/traccar/broadcast/BroadcastInterface.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2022 - 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. @@ -18,6 +18,7 @@ package org.traccar.broadcast; import org.traccar.model.BaseModel; import org.traccar.model.Device; import org.traccar.model.Event; +import org.traccar.model.ObjectOperation; import org.traccar.model.Position; public interface BroadcastInterface { @@ -34,12 +35,12 @@ public interface BroadcastInterface { default void updateCommand(boolean local, long deviceId) { } - default void invalidateObject(boolean local, Class<? extends BaseModel> clazz, long id) { + default <T extends BaseModel> void invalidateObject( + boolean local, Class<T> clazz, long id, ObjectOperation operation) throws Exception { } - default void invalidatePermission( - boolean local, - Class<? extends BaseModel> clazz1, long id1, - Class<? extends BaseModel> clazz2, long id2) { + default <T1 extends BaseModel, T2 extends BaseModel> void invalidatePermission( + boolean local, Class<T1> clazz1, long id1, Class<T2> clazz2, long id2, boolean link) throws Exception { } + } diff --git a/src/main/java/org/traccar/broadcast/BroadcastMessage.java b/src/main/java/org/traccar/broadcast/BroadcastMessage.java index 985848d04..0d15d7495 100644 --- a/src/main/java/org/traccar/broadcast/BroadcastMessage.java +++ b/src/main/java/org/traccar/broadcast/BroadcastMessage.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 Anton Tananaev (anton@traccar.org) + * Copyright 2022 - 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. @@ -17,10 +17,9 @@ package org.traccar.broadcast; import org.traccar.model.Device; import org.traccar.model.Event; +import org.traccar.model.ObjectOperation; import org.traccar.model.Position; -import java.util.Map; - public class BroadcastMessage { private Device device; @@ -73,13 +72,112 @@ public class BroadcastMessage { this.commandDeviceId = commandDeviceId; } - private Map<String, Long> changes; + public static class InvalidateObject { + + private String clazz; + + public String getClazz() { + return clazz; + } + + public void setClazz(String clazz) { + this.clazz = clazz; + } + + private long id; + + public long getId() { + return id; + } + + public void setId(long id) { + this.id = id; + } + + private ObjectOperation operation; + + public ObjectOperation getOperation() { + return operation; + } + + public void setOperation(ObjectOperation operation) { + this.operation = operation; + } - public Map<String, Long> getChanges() { - return changes; } - public void setChanges(Map<String, Long> changes) { - this.changes = changes; + private InvalidateObject invalidateObject; + + public InvalidateObject getInvalidateObject() { + return invalidateObject; + } + + public void setInvalidateObject(InvalidateObject invalidateObject) { + this.invalidateObject = invalidateObject; } + + public static class InvalidatePermission { + + private String clazz1; + + public String getClazz1() { + return clazz1; + } + + public void setClazz1(String clazz1) { + this.clazz1 = clazz1; + } + + private long id1; + + public long getId1() { + return id1; + } + + public void setId1(long id1) { + this.id1 = id1; + } + + private String clazz2; + + public String getClazz2() { + return clazz2; + } + + public void setClazz2(String clazz2) { + this.clazz2 = clazz2; + } + + private long id2; + + public long getId2() { + return id2; + } + + public void setId2(long id2) { + this.id2 = id2; + } + + private boolean link; + + public boolean getLink() { + return link; + } + + public void setLink(boolean link) { + this.link = link; + } + + } + + private InvalidatePermission invalidatePermission; + + public InvalidatePermission getInvalidatePermission() { + return invalidatePermission; + } + + public void setInvalidatePermission(InvalidatePermission invalidatePermission) { + this.invalidatePermission = invalidatePermission; + } + } diff --git a/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java b/src/main/java/org/traccar/broadcast/MulticastBroadcastService.java index b1b66f1e3..793c6df36 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<BroadcastInterface> 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<? extends BaseModel> clazz, long id) { - BroadcastMessage message = new BroadcastMessage(); - message.setChanges(Map.of(Permission.getKey(clazz), id)); - sendMessage(message); - } - - @Override - public void invalidatePermission( - boolean local, - Class<? extends BaseModel> clazz1, long id1, - Class<? extends BaseModel> 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); @@ -191,7 +103,7 @@ public class MulticastBroadcastService implements BroadcastService { } publisherSocket = null; socket.leaveGroup(group, networkInterface); - } catch (IOException e) { + } catch (Exception e) { throw new RuntimeException(e); } } diff --git a/src/main/java/org/traccar/broadcast/RedisBroadcastService.java b/src/main/java/org/traccar/broadcast/RedisBroadcastService.java new file mode 100644 index 000000000..697c45a4a --- /dev/null +++ b/src/main/java/org/traccar/broadcast/RedisBroadcastService.java @@ -0,0 +1,125 @@ +/* + * 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 com.fasterxml.jackson.databind.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.config.Config; +import org.traccar.config.Keys; + +import java.io.IOException; +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 extends BaseBroadcastService { + + private static final Logger LOGGER = LoggerFactory.getLogger(RedisBroadcastService.class); + + private final ObjectMapper objectMapper; + + private final ExecutorService service = Executors.newSingleThreadExecutor(); + + private final String channel = "traccar"; + + private Jedis subscriber; + private Jedis publisher; + + private final String id = UUID.randomUUID().toString(); + + public RedisBroadcastService(Config config, ObjectMapper objectMapper) throws IOException { + this.objectMapper = objectMapper; + String url = config.getString(Keys.BROADCAST_ADDRESS); + + try { + subscriber = new Jedis(url); + publisher = new Jedis(url); + subscriber.connect(); + } catch (JedisConnectionException e) { + throw new IOException(e); + } + } + + @Override + public boolean singleInstance() { + return false; + } + + @Override + protected void sendMessage(BroadcastMessage message) { + try { + String payload = id + ":" + objectMapper.writeValueAsString(message); + publisher.publish(channel, payload); + } catch (IOException | JedisConnectionException e) { + LOGGER.warn("Broadcast failed", e); + } + } + + @Override + public void start() throws IOException { + service.submit(receiver); + } + + @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() { + try { + subscriber.subscribe(new JedisPubSub() { + @Override + public void onMessage(String messageChannel, String message) { + try { + String[] parts = message.split(":", 2); + if (messageChannel.equals(channel) && parts.length == 2 && !id.equals(parts[0])) { + handleMessage(objectMapper.readValue(parts[1], BroadcastMessage.class)); + } + } catch (Exception e) { + LOGGER.warn("Broadcast handleMessage failed", e); + } + } + }, channel); + } catch (JedisException e) { + throw new RuntimeException(e); + } + } + }; + +} |