update for sr3 configuration and callback

rmv "lp-" from loader for testing

debian/postinst

@@ -33,8 +33,7 @@ export MSC_PYGEOAPI_ES_URL=http://localhost:9200
 export MSC_PYGEOAPI_CACHEDIR=/tmp
 export MSC_PYGEOAPI_OGC_API_URL=https://api.wxod-dev.cmc.ec.gc.ca
 export MSC_PYGEOAPI_OGC_API_URL_BASEPATH=/
-export MSC_PYGEOAPI_METPX_EVENT_FILE_PY=/opt/msc-pygeoapi/event/file_.py
-export MSC_PYGEOAPI_METPX_EVENT_MESSAGE_PY=/opt/msc-pygeoapi/event/message.py
+export MSC_PYGEOAPI_METPX_FLOW_CALLBACK=msc_pygeoapi.event.event_plugin.Event
 
 export XDG_CACHE_HOME=/tmp/geoadm-sarra-logs
 

deploy/default/sarracenia/aqhi-realtime.conf

@@ -1,11 +1,13 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+instances 2
+
 subtopic *.WXO-DD.air_quality.aqhi.*.*.realtime.json.#
+
 mirror True
-discard on
-report_back False
+discard True
+report False
 directory ${MSC_PYGEOAPI_CACHEDIR}
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-accept .*
-instances 2
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}

deploy/default/sarracenia/bulletins-realtime.conf

@@ -1,14 +1,12 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
-directory /dev/null
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
 instances 4
-subtopic *.WXO-DD.bulletins.alphanumeric.#
-mirror True
-notify_only True
-accept .*
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_MESSAGE_PY}
 
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+subtopic *.WXO-DD.bulletins.alphanumeric.#
 
-report_back False
+mirror True
+directory /tmp/bulletins
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+report False

deploy/default/sarracenia/cap-alerts.conf

@@ -1,11 +1,12 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
-directory /data/geomet/feeds/hpfx
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
 instances 2
+
 subtopic *.WXO-DD.alerts.cap.#
+
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+directory /data/geomet/feeds/hpfx
 mirror True
 discard True
 skip 3
-accept .*
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}

deploy/default/sarracenia/citypageweather.conf

@@ -1,11 +1,12 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
-directory ${MSC_PYGEOAPI_CACHEDIR}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
 instances 2
+
 subtopic *.WXO-DD.citypage_weather.xml.#
+
+directory ${MSC_PYGEOAPI_CACHEDIR}
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
 mirror True
 discard True
-slip 3
-accept .*
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
+strip 3

deploy/default/sarracenia/coastal-flood-risk-index.conf

@@ -1,20 +1,14 @@
 # TODO: Update for HPFX once the data is on HPFX
 broker amqps://anonymous:[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+instances 2
 
 subtopic coastal-flooding.risk-index.#
 
-mirror True
-
-discard on
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-
 directory ${MSC_PYGEOAPI_CACHEDIR}
-
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-
-report_back False
-
-instances 2
-accept .*
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+mirror True
+discard True
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+report False

deploy/default/sarracenia/hurricanes.conf

@@ -1,11 +1,12 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
-directory ${MSC_PYGEOAPI_CACHEDIR}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
 instances 2
+
 subtopic *.WXO-DD.hurricanes.#
+
+directory ${MSC_PYGEOAPI_CACHEDIR}
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
 mirror True
 discard True
 slip 3
-accept .*
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}

deploy/default/sarracenia/hydrometric-realtime.conf

@@ -1,22 +1,19 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
-subtopic *.WXO-DD.hydrometric.#
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
 instances 4
-reject .*_.*_hourly_.*
-
-discard on
 
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
+subtopic *.WXO-DD.hydrometric.#
 
 directory ${MSC_PYGEOAPI_CACHEDIR}
-
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-
-report_back False
-
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+discard True
+report False
 skip 3
 
+reject .*_.*_hourly_.*
 accept .*hydrometric_StationList.csv
-
 accept .*_.*_daily_.*
 accept .*_hourly_.*
+acceptUnmatched False

deploy/default/sarracenia/marine_weather.conf

@@ -1,12 +1,15 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
-directory /data/geomet/feeds/hpfx
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
 instances 2
+
 subtopic *.WXO-DD.marine_weather.xml.#
+
+directory /data/geomet/feeds/hpfx
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
 mirror True
 discard True
-reject .*regionList.xml
 skip 3
-accept .*
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-chmod_log 0644
+permLog 0644
+
+reject .*regionList.xml

deploy/default/sarracenia/metnotes.conf

@@ -1,11 +1,13 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+instances 2
+
 subtopic *.WXO-DD.metnotes.#
-mirror True
-discard on
-report_back False
+
 directory ${MSC_PYGEOAPI_CACHEDIR}
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-accept .*
-instances 2
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+mirror True
+discard True
+report False
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}

deploy/default/sarracenia/swob-realtime.conf

@@ -1,20 +1,14 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+exchange xpublic
+instances 4
 
 subtopic *.WXO-DD.observations.swob-ml.#
 
-mirror True
-
-discard on
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-
 directory ${MSC_PYGEOAPI_CACHEDIR}
-
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-
-report_back False
-
-instances 4
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+mirror True
+discard True
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+report False
 skip 3
-accept .*

deploy/default/sarracenia/thunderstorm-outlook.conf

@@ -1,20 +1,14 @@
 # TODO: Update for HPFX once the data is on HPFX
 broker amqps://anonymous:[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+instances 2
 
 subtopic thunderstorm-outlooks.#
 
-mirror True
-
-discard on
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-
 directory ${MSC_PYGEOAPI_CACHEDIR}
-
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-
-report_back False
-
-instances 2
-accept .*
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+mirror True
+discard True
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+report False

deploy/default/sarracenia/umos-realtime.conf

@@ -1,20 +1,14 @@
 broker amqps://[email protected]
-queue_name q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+exchange xpublic
+queueName q_${BROKER_USER}.${PROGRAM}.${CONFIG}.${HOSTNAME}
+instances 4
 
 subtopic *.WXO-DD.model_gem_global.stat-post-processing.#
 subtopic *.WXO-DD.model_gem_regional.stat-post-processing.#
 
-mirror True
-
-discard on
-
-plugin ${MSC_PYGEOAPI_METPX_EVENT_FILE_PY}
-
 directory ${MSC_PYGEOAPI_CACHEDIR}
-
-loglevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
-
-report_back False
-
-instances 4
-accept .*
+callback ${MSC_PYGEOAPI_METPX_FLOW_CALLBACK}
+mirror True
+discard True
+logLevel ${MSC_PYGEOAPI_LOGGING_LOGLEVEL}
+report False

msc-pygeoapi.env

@@ -8,8 +8,7 @@ export MSC_PYGEOAPI_ES_URL=http://${MSC_PYGEOAPI_ES_USERNAME}:${MSC_PYGEOAPI_ES_
 export MSC_PYGEOAPI_CACHEDIR=/tmp
 export MSC_PYGEOAPI_OGC_API_URL=https://api.wxod-dev.cmc.ec.gc.ca
 export MSC_PYGEOAPI_OGC_API_URL_BASEPATH=/
-export MSC_PYGEOAPI_METPX_EVENT_FILE_PY=/opt/msc-pygeoapi/event/file_.py
-export MSC_PYGEOAPI_METPX_EVENT_MESSAGE_PY=/opt/msc-pygeoapi/event/message.py
+export MSC_PYGEOAPI_METPX_FLOW_CALLBACK=msc_pygeoapi.event.event_plugin.Event
 export MSC_PYGEOAPI_TEMPLATES=theme/templates
 export MSC_PYGEOAPI_STATIC=theme/static
 export MSC_PYGEOAPI_LOCALE=locale

msc_pygeoapi/event/file_.py → msc_pygeoapi/event/event_plugin.py

@@ -1,8 +1,13 @@
 # =================================================================
 #
 # Author: Tom Kralidis <[email protected]>
+#         Louis-Philippe Rousseau-Lambert
+#             <[email protected]>
+#         Etienne Pelletier <[email protected]>
 #
 # Copyright (c) 2023 Tom Kralidis
+# Copyright (c) 2024 Louis-Philippe Rousseau-Lambert
+# Copyright (c) 2024 Etienne Pelletier
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation
@@ -26,39 +31,44 @@
 # OTHER DEALINGS IN THE SOFTWARE.
 #
 # =================================================================
+import logging
+import os
 
+from sarracenia.flowcb import FlowCB
 
-class FileEvent:
-    """core event"""
+LOGGER = logging.getLogger(__name__)
 
-    def __init__(self, parent):
-        """initialize"""
-        pass
 
-    def on_file(self, parent):
+class Event(FlowCB):
+    """sr3 plugin callback"""
+
+    # replace the previous sr2 on_file event
+    def after_work(self, worklist) -> None:
         """
         sarracenia dispatcher
 
-        :param parent: `sarra.sr_subscribe.sr_subscribe`
+        :param worklist: `sarracenia.flowcb`
 
-        :returns: `bool` of dispatch result
+        :returns: None
         """
 
-        try:
-            from msc_pygeoapi.handler.core import CoreHandler
+        for msg in worklist.incoming:
 
-            filepath = parent.msg.local_file
-            parent.logger.debug(f'Filepath: {filepath}')
-            handler = CoreHandler(filepath)
-            result = handler.handle()
-            parent.logger.debug(f'Result: {result}')
-            return True
-        except Exception as err:
-            parent.logger.warning(err)
-            return False
+            try:
+                from msc_pygeoapi.handler.core import CoreHandler
 
-    def __repr__(self):
-        return '<Event>'
+                filepath = os.path.join(msg['new_dir'], msg['new_file'])
+                LOGGER.debug(f'Filepath: {filepath}')
+                handler = CoreHandler(filepath)
+                result = handler.handle()
+                LOGGER.debug(f'Result: {result}')
+            except Exception as err:
+                LOGGER.error(f'Error handling message: {err}', exc_info=True)
+                worklist.failed.append(msg)
+                return False
 
+    # replace the previous sr2 on_message event
+    after_accept = after_work
 
-self.plugin = 'FileEvent'  # noqa
+    def __repr__(self):
+        return '<Event>'