psrt feature

UPDATE.rst

@@ -4,6 +4,8 @@ EVA ICS 3.4.2
 What's new
 ==========
 
+- PSRT support
+- "mqtt" parameter in "default cloud" feature renamed to "host"
 - LM PLC stability and performance improvements
 - Bug fixes and general improvements
 

cli/check-psrt.py

@@ -0,0 +1,77 @@
+__author__ = "Altertech Group, https://www.altertech.com/"
+__copyright__ = "Copyright (C) 2012-2021 Altertech Group"
+__license__ = "Apache License 2.0"
+__version__ = "3.4.2"
+
+import sys
+import argparse
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+
+from pathlib import Path
+sys.path.insert(0, (Path(__file__).absolute().parents[1] / 'lib').as_posix())
+
+import eva.core
+import eva.tools
+import eva.notify
+import eva.api
+
+eva.core.set_product('test', -1)
+
+_me = 'EVA ICS PSRT test version %s' % __version__
+
+ap = argparse.ArgumentParser(description=_me)
+ap.add_argument(help='PSRT user:pass@host:port/space',
+                dest='_psrt',
+                metavar='PSRT')
+ap.add_argument('--cafile', help='CA File', dest='_ca_file', metavar='FILE')
+
+try:
+    import argcomplete
+    argcomplete.autocomplete(ap)
+except:
+    pass
+
+a = ap.parse_args()
+
+if not a._psrt:
+    ap.print_usage()
+    sys.exit(99)
+
+if a._psrt.find('@') != -1:
+    try:
+        mqa, mq = a._psrt.split('@')
+        user, password = mqa.split(':')
+    except:
+        ap.print_usage()
+        sys.exit(99)
+else:
+    mq, user, password = a._psrt, None, None
+
+if mq.find('/') != -1:
+    try:
+        x = mq.split('/')
+        mq = x[0]
+        space = '/'.join(x[1:])
+    except:
+        ap.print_usage()
+        sys.exit(99)
+else:
+    space = None
+
+psrt_host, psrt_port = eva.tools.parse_host_port(mq, 2883)
+
+n = eva.notify.PSRTNotifier(notifier_id='test',
+                            host=psrt_host,
+                            port=psrt_port,
+                            space=space,
+                            username=user if user else None,
+                            password=password if password else None,
+                            ca_certs=a._ca_file)
+
+if n.test():
+    print('OK')
+else:
+    print('FAILED')
+    sys.exit(1)

install/mklinks

@@ -9,7 +9,7 @@ ln -sf ../cli/venvl tests/benchmark-uc-crt
 
 BIN="uc-cmd lm-cmd sfa-cmd test-uc-xc test-ext test-phi uc-tpl prepare-sr gen-intl key-deploy key-import"
 NBIN="uc-notifier lm-notifier sfa-notifier"
-SBIN="check-mqtt get-setup-options eva-update-tables apikey-set pypi-mirror"
+SBIN="check-mqtt check-psrt get-setup-options eva-update-tables apikey-set pypi-mirror"
 INSTALL="import-registry-defaults import-registry-schema convert-legacy-configs"
 
 for p in ${BIN}; do

lib/eva/features/default_cloud.py

@@ -1,40 +1,66 @@
 from eva.features import InvalidParameter, dir_eva
 from eva.features import cli_call, is_enabled, exec_shell, restart_controller
 from eva.features import val_to_boolean
+from eva.features import print_warn
 
 from textwrap import dedent
 
 
-def setup(mqtt=None,
+def setup(host=None,
           id=None,
           ca=None,
           cert=None,
           key=None,
           retain=None,
-          announce=None):
-    check_cmd = dir_eva + '/sbin/check-mqtt'
-    retain = True if retain is None else val_to_boolean(retain)
+          announce=None,
+          proto=None,
+          socket_buf_size=None):
+    if proto is None:
+        proto = 'mqtt'
+    elif proto not in ['mqtt', 'psrt']:
+        raise InvalidParameter(f'Invalid protocol: {proto}')
+    check_cmd = dir_eva + f'/sbin/check-{proto}'
+    if proto == 'psrt':
+        retain = False if retain is None else val_to_boolean(retain)
+    else:
+        retain = True if retain is None else val_to_boolean(retain)
     announce = 30 if announce is None else float(announce)
     if not id:
         id = 'eva_1'
-    if '/' in mqtt:
-        _mqtt, space = mqtt.rsplit('/', 1)
+    if '/' in host:
+        _host, space = host.rsplit('/', 1)
     else:
-        _mqtt = mqtt
+        _host = host
         space = None
-    batch = [f'create {id} mqtt:{_mqtt}{(" -s " + space) if space else ""}']
+    batch = [f'create {id} {proto}:{_host}{(" -s " + space) if space else ""}']
     if ca:
         batch.append(f'set {id} ca_certs {ca}')
         check_cmd += f' --cafile {ca}'
     if cert:
-        batch.append(f'set {id} certfile {ca}')
-        check_cmd += f' --cert {cert}'
-    if key:
-        batch.append(f'set {id} keyfile {ca}')
-        check_cmd += f' --key {key}'
-    check_cmd += f' {mqtt}'
+        if proto == 'psrt':
+            print_warn('cert/key auth no supported by psrt')
+        else:
+            batch.append(f'set {id} certfile {ca}')
+            check_cmd += f' --cert {cert}'
+    if key and proto != 'psrt':
+        if proto == 'psrt':
+            print_warn('cert/key auth no supported by psrt')
+        else:
+            batch.append(f'set {id} keyfile {ca}')
+            check_cmd += f' --key {key}'
+    check_cmd += f' {host}'
     exec_shell(check_cmd, passthru=True)
-    batch.append(f'set {id} retain_enabled {retain}')
+    if retain and proto == 'psrt':
+        print_warn('retain not supported by psrt')
+    if proto == 'mqtt':
+        batch.append(f'set {id} retain_enabled {retain}')
+    if socket_buf_size:
+        if proto == 'psrt':
+            batch.append(f'set {id} socket_buf_size {socket_buf_size}')
+        else:
+            print_warn('socket_buf_size supported by psrt only')
+    if proto == 'mqtt':
+        batch.append(f'set {id} retain_enabled {retain}')
     batch.append(f'test {id}')
     batch.append(f'subscribe state {id} -p "#" -g "#"')
     batch.append(f'subscribe log {id}')
@@ -60,5 +86,4 @@ def remove(id=None):
                 cli_call(f'ns {c}', f'destroy {id}', return_result=True)
                 restart_controller(c)
             except:
-                from eva.features import print_warn
                 print_warn(f'unable to destroy {id} notifier for {c}')

lib/eva/features/default_cloud.yml

@@ -6,18 +6,20 @@ help: |
   Sets default announce interval to 30 seconds. Default connection id is eva_1
 
   Subscribes MQTT to all state, log (INFO) and server events
-example: {{ setup_cmd }}mqtt=user:password@192.168.1.12
+example: {{ setup_cmd }}host=user:password@192.168.1.12
 doc-url: https://eva-ics.readthedocs.io/en/{{ EVA_VERSION }}/notifiers.html#mqtt-mqtt
 setup:
   mandatory-args:
-    mqtt: "as [user:password]@host[:port][/space]"
+    host: "as [user:password]@host[:port][/space]"
   optional-args:
-    id: "MQTT connection ID (default: eva_1)"
+    id: "MQTT/PSRT connection ID (default: eva_1)"
     ca: CA file (absolute path)
-    cert: Certificate file (absolute path)
-    key: Key file (absolute path)
+    cert: "Certificate file (absolute path, for MQTT only)"
+    key: "Key file (absolute path, for MQTT only"
     retain: "use retain topics (default: True)"
     announce: custom annonce interval
+    proto: "mqtt or psrt"
+    socket_buf_size: "psrt socket buffer size"
 remove:
   optional-args:
-    id: "MQTT connection ID (default: eva_1)"
+    id: "MQTT/PSRT connection ID (default: eva_1)"

lib/eva/notify.py

@@ -2454,7 +2454,7 @@ def start_announcer(self):
     def on_connect(self, client, userdata, flags, rc):
         if eva.core.is_shutdown_requested():
             return
-        logging.debug('.%s mqtt reconnect' % self.notifier_id)
+        logging.debug(f'.{self.notifier_id} {self.proto} reconnect')
         self.mq_connected.set()
         if self.announce_interval and not self.test_only_mode:
             eva.core.spawn_daemon(self.start_announcer)
@@ -2682,8 +2682,8 @@ def on_message(self, client, userdata, msg):
             d = msg.payload if msg.payload.startswith(
                 b'\x00') else msg.payload.decode()
         except:
-            logging.warning('.Invalid message from MQTT server: {}'.format(
-                msg.payload))
+            logging.warning(
+                f'.Invalid message from {self.proto} server, topic: {t}')
             eva.core.log_traceback(notifier=True)
             return
         try:
@@ -3031,8 +3031,8 @@ def test(self):
                     f'{self.pfx}controller/{eva.core.product.code}'
                     f'/{eva.core.config.system_name}/test-{uuid.uuid4()}')
                 self.test_topic = test_topic
-                logging.debug('.Testing mqtt notifier %s (%s:%u)' % \
-                        (self.notifier_id,self.host, self.port))
+                logging.debug('.Testing %s notifier %s (%s:%u)' % \
+                        (self.proto, self.notifier_id,self.host, self.port))
                 if not self.check_connection():
                     return False
                 t_start = time.perf_counter()