Merge pull request #2326 from Squidly271/patch-8

Feat: Allow selected nchan publishers to exit if no listeners

emhttp/plugins/dynamix.docker.manager/nchan/docker_load

@@ -1,13 +1,30 @@
-#!/bin/bash
+#!/usr/bin/php -q
+<?PHP
+/* Copyright 2005-2025, Lime Technology
+ * Copyright 2012-2025, Bergware International.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License version 2,
+ * as published by the Free Software Foundation.
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ */
+?>
+<?
+$docroot ??= ($_SERVER['DOCUMENT_ROOT'] ?: '/usr/local/emhttp');
 
-while :; do
-  output=$(docker stats --no-stream --format='{{.ID}};{{.CPUPerc}};{{.MemUsage}}' 2>&1)
-  if [[ $? -ne 0 ]]; then
-    logger -t webgui "docker stats error: $output"
-    exit 1
-  fi
-  if [[ -n "$output" ]]; then
-    curl -sfd "$output" --unix-socket /var/run/nginx.socket http://localhost/pub/dockerload?buffer_length=1 >/dev/null 2>&1
-  fi
-  sleep 1
-done
+require_once "$docroot/webGui/include/Helpers.php";
+require_once "$docroot/webGui/include/publish.php";
+
+while (true) {
+  $output = shell_exec("docker stats --no-stream --format='{{.ID}};{{.CPUPerc}};{{.MemUsage}}' 2>&1");  
+
+  publish_md5("dockerload", $output,true);
+  if ( $output === null ) {
+    // slow down publishing if no containers are running
+    sleep(10);
+  } else {    
+    sleep(1);
+  }
+}

emhttp/plugins/dynamix.vm.manager/nchan/vm_usage

@@ -87,11 +87,7 @@ while (true) {
   }
   if ($running < 1) $echo = "<tr><td colspan='7' style='text-align:center;padding-top:12px'>"._('No VMs running')."</td></tr>";
  
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('vm_usage',$echo)!==false ? $md5_new : -1;
-  }
+  publish_md5('vm_usage',$echo,true);
 
   sleep($timer);
 }

emhttp/plugins/dynamix/include/publish.php

@@ -1,5 +1,5 @@
 <?PHP
-/* Copyright 2005-2023, Lime Technology
+/* Copyright 2005-2025, Lime Technology
  * Copyright 2012-2023, Bergware International.
  *
  * This program is free software; you can redistribute it and/or
@@ -11,7 +11,8 @@
  */
 ?>
 <?
-$docroot ??= ($_SERVER['DOCUMENT_ROOT'] ?: '/usr/local/emhttp');
+$docroot ??= ($_SERVER['DOCUMENT_ROOT'] ?? '/usr/local/emhttp');
+
 require_once "$docroot/webGui/include/Wrappers.php";
 
 function curl_socket($socket, $url, $message='') {
@@ -23,17 +24,40 @@ function curl_socket($socket, $url, $message='') {
   if ($reply===false) my_logger("curl to $url failed", 'curl_socket');
   return $reply;
 }
-function publish($endpoint, $message, $len=1) {
-  static $com  = [];
-  static $lens = [];
+
+// $message: if an array, it will be converted to a JSON string
+// $abort: if true, the script will exit if the endpoint is without subscribers on the next publish attempt after $abortTime seconds
+// If a script publishes to multiple endpoints, timing out on one endpoint will terminate the entire script even if other enpoints succeed.
+// If this is a problem, don't use $abort and instead handle this in the script or utilize a single sript per endpoint.
+//
+// $opt1: if numeric it's the length of the buffer.  If its true|false it signifies whether to utilise the abort if no listeners.
+// $opt2: if $opt1 is numeric, it's a boolean for $abort. If $opt1 is boolean, its the timeout defaulting to $opt3
+// $opt3: if $opt1 is numeric, it's a value for $abortTime.  If $opt1 is boolean, this parameter shouldn't be used
+function publish($endpoint, $message, $opt1=1, $opt2=false, $opt3=120) {
+  static $abortStart = [], $com = [], $lens = [];
 
   if ( is_file("/tmp/publishPaused") )
     return false;
 
+  if ( is_array($message) ) {
+    $message = json_encode($message);
+  }
+
+  // handle the $opt1/$opt2/$opt3 parameters while remaining backwards compatible
+  if ( is_numeric($opt1) ) {
+    $len = $opt1;
+    $abort = $opt2;
+    $abortTime = $opt3;
+  } else {
+    $len = 1;
+    $abort = $opt1;
+    $abortTime = $opt2 ?: $opt3;
+  }
+
   // Check for the unlikely case of a buffer length change
   if ( (($lens[$endpoint] ?? 1) !== $len) && isset($com[$endpoint]) ) {
     curl_close($com[$endpoint]);
-    unset($com[$endpoint]);
+    unset($com[$endpoint],$abortStart[$endpoint]);
   }
   if ( !isset($com[$endpoint]) ) {
     $lens[$endpoint] = $len;
@@ -50,6 +74,7 @@ function publish($endpoint, $message, $len=1) {
   $reply = curl_exec($com[$endpoint]);
   $err   = curl_error($com[$endpoint]);
   if ($err) {
+    my_logger("curl error: $err endpoint: $endpoint");
     curl_close($com[$endpoint]);
     unset($com[$endpoint]);
 
@@ -65,7 +90,76 @@ function publish($endpoint, $message, $len=1) {
       @unlink("/tmp/publishPaused");
     }
   }
-  if ($reply===false) my_logger("curl to $endpoint failed", 'publish');
+  if ($reply===false) 
+    my_logger("curl to $endpoint failed", 'publish');
+
+  if ($abort) {
+    $json = @json_decode($reply,true);
+    if ($reply===false || ( is_array($json) && ($json['subscribers']??false) === 0) ) {
+      if ( ! ($abortStart[$endpoint]??false) ) 
+        $abortStart[$endpoint] = time();
+      if ( (time() - $abortStart[$endpoint]) > $abortTime) {
+        my_logger("$endpoint timed out after $abortTime seconds.  Exiting.", 'publish');
+        
+        removeNChanScript();
+        exit();
+      }
+    } else {
+      // a subscriber is present.  Reset the abort timer if it's set
+      $abortStart[$endpoint] = null;
+    }
+  }
   return $reply;
 }
+
+// Function to not continually republish the same message if it hasn't changed since the last publish
+function publish_md5($endpoint, $message, $opt1=1, $opt2=false, $opt3=120) {
+  static $md5_old = [];
+  static $md5_time = [];
+ 
+  if ( is_numeric($opt1) ) {
+    $timeout = $opt3;
+    $abort = $opt2;
+  } else {
+    $abort = $opt1;
+    $timeout = $opt2 ?: $opt3;
+  }
+
+  if ( is_array($message) ) {
+    $message = json_encode($message);
+  }
+
+  // if abort is set, republish the message even if it hasn't changed after $timeout seconds to check for subscribers and exit accordingly
+  if ( $abort ) {
+    if ( (time() - ($md5_time[$endpoint]??0)) > $timeout ) {
+      $md5_old[$endpoint] = null;
+    }
+  }
+
+  // Always hash the payload to avoid collapsing distinct "falsey" values
+  $md5_new = md5((string)$message, true);
+  if ($md5_new !== ($md5_old[$endpoint]??null)) {
+    $md5_old[$endpoint] = $md5_new;
+    $md5_time[$endpoint] = time();
+
+    return publish($endpoint, $message, $opt1, $opt2, $opt3);
+  }
+}
+
+
+// Removes the script calling this function from nchan.pid
+function removeNChanScript() {
+  global $docroot, $argv;
+
+  $script = trim(str_replace("$docroot/","",(php_sapi_name() === 'cli') ? $argv[0] : $_SERVER['argv'][0]));
+  $nchan = @file("/var/run/nchan.pid",FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES) ?: [];
+  $nchan = array_filter($nchan,function($x) use ($script) {
+     return $script !== trim(explode(":",$x)[0]);
+  });
+  if (count($nchan) > 0) {
+    file_put_contents_atomic("/var/run/nchan.pid",implode("\n",$nchan)."\n");
+  } else {
+    @unlink("/var/run/nchan.pid");
+  }
+}
 ?>

emhttp/plugins/dynamix/nchan/update_1

@@ -122,11 +122,9 @@ while (true) {
   $share = array_keys($shares);
   $count = array_count_values($lsof ?? array());
   foreach ($share as $name) $echo['stream'][] = $count[$name]??0;
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('update1',$echo)!==false ? $md5_new : -1;
-  }
+
+  publish_md5('update1',$echo,true);
+  
   sleep(5);
 }
 ?>

emhttp/plugins/dynamix/nchan/update_2

@@ -499,11 +499,7 @@ while (true) {
     $echo[$a] = [implode($echo[$a]), $extra];
   }
 
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('update2',$echo)!==false ? $md5_new : -1;
-  }
+  publish_md5('update2',$echo,true);
   sleep(2);
 }
 ?>

emhttp/plugins/dynamix/nchan/update_3

@@ -159,12 +159,8 @@ while (true) {
   $date  = my_date($xdate ? 'D j M Y, T' : $display['date'].', T',$now);
   $echo['time'] = [$clock,_($date,0)];
 
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('update3',$echo)!==false ? $md5_new : -1;
-    $time0 = $time1;
-  }
+  publish_md5('update3',$echo,true);
+
   sleep(1);
   $time1 = microtime(true);
 }

emhttp/plugins/dynamix/nchan/ups_status

@@ -131,11 +131,9 @@ while (true) {
     elseif (isset($load)) $echo[5] = ($load<90 ? "<span>" : "<span $red>").$echo[5]."</span>";
     $echo[6] = isset($output) ? ((empty($volt) || ($minv<$output && $output<$maxv) ? "<span $green>" : "<span $red>").$echo[6].(isset($freq) ? " ~ $freq Hz" : "")."</span>") : $echo[6];
   }
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('apcups',$echo)!==false ? $md5_new : -1;
-  }
+
+  publish_md5('apcups',$echo,true);
+
   sleep(3);
 }
 ?>

emhttp/plugins/dynamix/nchan/vm_dashusage

@@ -40,14 +40,14 @@ if ($var['fsState'] == "Started" || $var['fsState'] == "Starting") {
     } 
     sleep(10);
   } elseif ($domain_cfg['SERVICE'] != "enable") { 
-    #Add remove_nchan_pid_entry("webGui/nchan/vm_dashusage");
-    exec("sed -i '/webGui\/nchan\/vm_dashusage/d' /var/run/nchan.pid");
-    return;
+    sleep(10);
+    removeNChanScript();
+    exit;
   }
 } else {
-  #Add remove_nchan_pid_entry("webGui/nchan/vm_dashusage");
-  exec("sed -i '/webGui\/nchan\/vm_dashusage/d' /var/run/nchan.pid");
-  return;
+  sleep(10);
+  removeNChanScript();
+  exit;
 }
 
 require_once "$docroot/plugins/dynamix.vm.manager/include/libvirt_helpers.php";
@@ -109,11 +109,8 @@ while (true) {
     }
   }
 
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('vm_dashusage',$echo)!==false ? $md5_new : -1;
-  }
+  publish_md5('vm_dashusage',$echo,true);
+
   sleep($timer);
 }
 ?>

emhttp/plugins/dynamix/nchan/wg_poller

@@ -37,11 +37,7 @@ while (true) {
     $row = preg_split('/\s+/',$row);
     if (count($row)>5) $echo[] = [$row[0], $row[5]?$now-$row[5]:0, my_scale($row[6],$unit)." $unit", my_scale($row[7],$unit)." $unit"];
   }
-  $echo = json_encode($echo);
-  $md5_new = md5($echo,true);
-  if ($md5_new !== $md5_old) {
-    $md5_old = publish('wireguard',$echo)!==false ? $md5_new : -1;
-  }
+  publish_md5('wireguard',$echo,true);
   sleep(1);
 }
 ?>

emhttp/plugins/dynamix/nchan/wlan0

@@ -65,7 +65,8 @@ while (true) {
       $echo['title'] = _('WiFi status unavailable');
     }
   }
-  publish('wlan0',json_encode($echo));
+  // Short of closing all web pages, the timeout should only happen if wifi is enabled and then disabled
+  publish_md5('wlan0',$echo,true);
   sleep(3);
 }
 ?>