Merge pull request #3571 from ControlSystemStudio/CSSTUDIO-3410

CSSTUDIO-3410 Add check for and annunciation of disconnection from the Alarm Server to the Annunciator

Author: shroffk

app/alarm/model/src/main/java/org/phoebus/applications/alarm/talk/TalkClient.java

@@ -7,13 +7,19 @@
  *******************************************************************************/
 package org.phoebus.applications.alarm.talk;
 
+import static java.lang.Thread.sleep;
 import static org.phoebus.applications.alarm.AlarmSystem.logger;
+import static org.phoebus.applications.alarm.AlarmSystem.nag_period_ms;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -41,7 +47,10 @@ public class TalkClient
     private final CopyOnWriteArrayList<TalkClientListener> listeners = new CopyOnWriteArrayList<>();
     private final AtomicBoolean running = new AtomicBoolean(true);
     private final Consumer<String, String> consumer;
+    private final Consumer<String, String> heartbeatConsumer;
     private final Thread thread;
+    private final Thread updateHeartbeatTimestampThread;
+    private final Thread annunciateDisconnectionThread;
 
     /** @param server - Kafka Server host:port
      *  @param config_name - Name of kafka config topic that the talk topic accompanies.
@@ -56,6 +65,16 @@ public TalkClient(final String server, final String config_name)
 
         thread = new Thread(this::run, "TalkClient");
         thread.setDaemon(true);
+
+        {
+            heartbeatConsumer = KafkaHelper.connectConsumer(server, List.of(config_name), Collections.emptyList(), AlarmSystem.kafka_properties);
+            updateHeartbeatTimestampThread = new Thread(() -> updateHeartbeatTimestampLoop(), "UpdateHeartbeatTimestampThread");
+            updateHeartbeatTimestampThread.setDaemon(false);
+            updateHeartbeatTimestampThread.start();
+            annunciateDisconnectionThread = new Thread(() -> annunciateDisconnectionLoop(), "AnnunciateDisconnectionThread");
+            annunciateDisconnectionThread.setDaemon(false);
+            annunciateDisconnectionThread.start();
+        }
     }
 
     /** @param listener - Listener to add */
@@ -144,11 +163,62 @@ private void checkUpdates()
         }
     }
 
+    private Optional<Instant> nextDisconnectedAnnunciation = Optional.empty(); // When alarm server is disconnected: point in time for next annunciation of disconnection.
+    private final Duration disconnectionAnnunciationPeriod = Duration.ofMillis(AlarmSystem.nag_period_ms);
+    private final Duration idleTimeoutDuration = Duration.ofMillis(AlarmSystem.idle_timeout_ms).multipliedBy(3);
+    private AtomicReference<Instant> lastReceivedUpdateFromAlarmServer = new AtomicReference<>(Instant.now());
+
+    private void updateHeartbeatTimestampLoop() {
+        while (running.get()) {
+            final ConsumerRecords<String, String> records = heartbeatConsumer.poll(100);
+            if (!records.isEmpty()) {
+                lastReceivedUpdateFromAlarmServer.set(Instant.now());
+            }
+            try {
+                sleep(1000);
+            } catch (InterruptedException e) {
+                logger.log(Level.WARNING, "updateHeartbeatTimestampLoop() was interrupted when sleeping.");
+            }
+        }
+    }
+
+    /** Background thread loop that detects and annunciates disconnections. */
+    private void annunciateDisconnectionLoop() {
+        while (running.get()) {
+            Instant now = Instant.now();
+            if (Duration.between(lastReceivedUpdateFromAlarmServer.get(), now).compareTo(idleTimeoutDuration) > 0) {
+                if (nextDisconnectedAnnunciation.isEmpty() || nextDisconnectedAnnunciation.get().isBefore(now)) {
+                    try {
+                        for (final TalkClientListener listener : listeners) {
+                            listener.messageReceived(SeverityLevel.UNDEFINED, true, "Alarm Server Disconnected");
+                        }
+                    } catch (final Exception ex) {
+                        logger.log(Level.WARNING, "Talk error for " + SeverityLevel.UNDEFINED + ", " + "Alarm Server Disconnected", ex);
+                    }
+                    if (nag_period_ms > 0) {
+                        nextDisconnectedAnnunciation = Optional.of(now.plus(disconnectionAnnunciationPeriod)); // Annunciate disconnect again after nag period_ms
+                    }
+                    else {
+                        nextDisconnectedAnnunciation = Optional.of(Instant.MAX); // When nag_period_ms == 0, don't annunciate the disconnection again.
+                    }
+                }
+            } else {
+                nextDisconnectedAnnunciation = Optional.empty(); // Connection to the Alarm Server exists.
+            }
+            try {
+                sleep(1000);
+            } catch (InterruptedException e) {
+                logger.log(Level.WARNING, "annunciateDisconnectionLoop() was interrupted when sleeping.");
+            }
+        }
+    }
+
     /** Stop client */
     public void shutdown()
     {
         running.set(false);
         consumer.wakeup();
+        heartbeatConsumer.wakeup();
         try
         {
             thread.join(2000);
@@ -157,6 +227,22 @@ public void shutdown()
         {
             logger.log(Level.WARNING, "Talk client thread doesn't shut down", ex);
         }
+        try
+        {
+            annunciateDisconnectionThread.join(2000);
+        }
+        catch (final InterruptedException ex)
+        {
+            logger.log(Level.WARNING, "Annunciate Disconnection from Alarm Server thread doesn't shut down", ex);
+        }
+        try
+        {
+            updateHeartbeatTimestampThread.join(2000);
+        }
+        catch (final InterruptedException ex)
+        {
+            logger.log(Level.WARNING, "Update Alarm Server Heartbeat thread doesn't shut down", ex);
+        }
         logger.info(thread.getName() + " shut down");
     }
 }