Merge pull request #9 from ISISComputingGroup/forwarder_launch

add forwarder_launch

Author: KathrynBaker

.gitignore

@@ -81,9 +81,6 @@ target/
 profile_default/
 ipython_config.py
 
-# pyenv
-.python-version
-
 # pipenv
 #   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
 #   However, in case of collaboration, if having platform-specific dependencies or dependencies

.python-version

@@ -0,0 +1 @@
+3.12

forwarder/configuration_store.py

@@ -2,7 +2,7 @@
 from typing import Dict
 from unittest import mock
 
-from confluent_kafka import TopicPartition
+from confluent_kafka import TopicPartition, KafkaException, KafkaError
 from streaming_data_types.fbschemas.forwarder_config_update_fc00.UpdateType import (
     UpdateType,
 )
@@ -49,7 +49,20 @@ def save_configuration(self, update_handlers: Dict):
     def retrieve_configuration(self):
         """Retrieve last valid configuration buffer."""
         topic = TopicPartition(self._topic, partition=0)
-        low_offset, high_offset = self._consumer.get_watermark_offsets(topic)
+        try:
+            low_offset, high_offset = self._consumer.get_watermark_offsets(topic)
+        except KafkaException as e:
+            kafka_error = e.args[0]
+
+            if kafka_error == KafkaError._UNKNOWN_PARTITION:
+                # Topic doesn't exist yet - create it (assuming auto-create is enabled)
+                message = serialise_fc00(UpdateType.REMOVEALL, [])
+                self._producer.produce(self._topic, bytes(message), int(time.time() * 1000))
+                self._producer.flush()
+                low_offset, high_offset = self._consumer.get_watermark_offsets(topic)
+            else:
+                raise
+
         # Set offset to current_offset to start retrieving from last message
         current_offset = high_offset - 1
         topic.offset = current_offset

forwarder/kafka/kafka_producer.py

@@ -33,6 +33,9 @@ def _poll_loop(self):
         except BaseException as e:
             self.logger.exception(e)
 
+    def flush(self):
+        self._producer.flush()
+
     def close(self):
         self._cancelled = True
         self._poll_thread.join()

forwarder/scripts/run.py

@@ -194,7 +194,7 @@ def main():
         f"Forwarder version '{version}' started, service Id: {args.service_id}"
     )
     # EPICS
-    ca_ctx = CaContext()
+    ca_ctx = CaContext(client_name="forwarder")
     pva_ctx = PvaContext("pva", nt=False)
     # Using dictionary with Channel as key to ensure we avoid having multiple
     # handlers active for identical configurations: serialising updates from

forwarder_launch.bat

@@ -0,0 +1,13 @@
+call C:\Instrument\Apps\EPICS\isis\forwarder\master\.venv\scripts\activate
+
+set "EPICS_CA_ADDR_LIST=127.255.255.255 130.246.55.255"
+set "EPICS_CA_AUTO_ADDR_LIST=NO"
+
+if "%INSTRUMENT%" == "HIFI" (
+    set "BROKER=130.246.55.29:9092"
+) else (
+    set "BROKER=livedata.isis.cclrc.ac.uk:31092"
+)
+
+echo "starting forwarder"
+python C:\Instrument\Apps\EPICS\isis\forwarder\master\forwarder_launch.py --status-topic=%BROKER%/%INSTRUMENT%_forwarderStatus --config-topic=%BROKER%/%INSTRUMENT%_forwarderConfig --storage-topic=%BROKER%/%INSTRUMENT%_forwarderStorage  --output-broker=%BROKER%

requirements-dev.txt

@@ -1,49 +1,44 @@
-#
-# This file is autogenerated by pip-compile with Python 3.8
-# by the following command:
-#
-#    pip-compile --output-file=requirements.txt pyproject.toml
-#
-caproto==1.1.1
+# This file was autogenerated by uv via the following command:
+#    uv pip compile pyproject.toml -o requirements-frozen.txt
+caproto==1.2.0
     # via forwarder (pyproject.toml)
-configargparse==1.7
+configargparse==1.7.1
     # via forwarder (pyproject.toml)
-confluent-kafka==2.3.0
+confluent-kafka==2.11.0
     # via forwarder (pyproject.toml)
-epicscorelibs==7.0.7.99.0.2
+epicscorelibs==7.0.7.99.1.2
     # via
     #   p4p
     #   pvxslibs
-ess-streaming-data-types==0.26.1
+ess-streaming-data-types==0.27.0
     # via forwarder (pyproject.toml)
-flatbuffers==23.5.26
+flatbuffers==25.2.10
     # via ess-streaming-data-types
 graphyte==1.7.1
     # via forwarder (pyproject.toml)
-graypy==2.1.0
-    # via forwarder (pyproject.toml)
-nose2==0.14.1
+nose2==0.15.1
     # via p4p
-numpy==1.24.4
+numpy==1.26.4
     # via
+    #   forwarder (pyproject.toml)
     #   epicscorelibs
     #   ess-streaming-data-types
-    #   forwarder (pyproject.toml)
     #   p4p
-p4p==4.1.11
+p4p==4.2.1
     # via forwarder (pyproject.toml)
 ply==3.11
     # via p4p
-prometheus-client==0.20.0
+prometheus-client==0.22.1
     # via forwarder (pyproject.toml)
-pvxslibs==1.2.4
+pvxslibs==1.4.0
     # via p4p
-setuptools-dso==2.10
+setuptools==80.9.0
+    # via
+    #   epicscorelibs
+    #   setuptools-dso
+setuptools-dso==2.12.2
     # via
     #   epicscorelibs
     #   pvxslibs
-tomli==2.0.1
+tomli==2.2.1
     # via forwarder (pyproject.toml)
-
-# The following packages are considered to be unsafe in a requirements file:
-# setuptools