Merge pull request #11 from ISISComputingGroup/add_status_reporting_interval

Add status reporting interval

Author: rerpha

forwarder/parse_commandline_args.py

@@ -183,6 +183,14 @@ def parse_args():
         type=int,
         default=1000,
     )
+    parser.add_argument(
+        "--status-interval",
+        help="Period for status updates (units=milliseconds)",
+        required=False,
+        env_var="STATUS_INTERVAL",
+        type=int,
+        default=4000,
+    )
     log_choice_to_enum = {
         "Trace": logging.DEBUG,
         "Debug": logging.DEBUG,

forwarder/scripts/run.py

@@ -81,6 +81,7 @@ def create_status_reporter(
     service_id,
     version,
     logger,
+    status_interval,
 ):
     (
         broker,
@@ -107,6 +108,7 @@ def create_status_reporter(
         service_id,
         version,
         logger,
+        status_interval,
     )
     return status_reporter
 
@@ -249,6 +251,7 @@ def main():
             args.service_id,
             version,
             get_logger(),
+            args.status_interval,
         )
         exit_stack.callback(status_reporter.stop)
         status_reporter.start()

forwarder/status_reporter.py

@@ -22,7 +22,7 @@ def __init__(
         service_id: str,
         version: str,
         logger: Logger,
-        interval_ms: int = 4000,
+        interval_ms: int,
     ):
         self._repeating_timer = RepeatTimer(
             milliseconds_to_seconds(interval_ms), self.report_status

forwarder_launch.bat

@@ -13,5 +13,6 @@ set "EPICS_CA_ADDR_LIST=127.255.255.255 130.246.55.255"
 set "EPICS_CA_AUTO_ADDR_LIST=NO"
 
 echo "starting forwarder"
-python %~dp0forwarder_launch.py --status-topic=%KAFKA_BROKER%/%INSTRUMENT%_forwarderStatus --config-topic=%KAFKA_BROKER%/%INSTRUMENT%_forwarderConfig --storage-topic=%KAFKA_BROKER%/%INSTRUMENT%_forwarderStorage  --output-broker=%KAFKA_BROKER%
+python %~dp0forwarder_launch.py --status-topic=%KAFKA_BROKER%/%INSTRUMENT%_forwarderStatus --config-topic=%KAFKA_BROKER%/%INSTRUMENT%_forwarderConfig --storage-topic=%KAFKA_BROKER%/%INSTRUMENT%_forwarderStorage  --output-broker=%KAFKA_BROKER%  --status-interval=300000
+
 

tests/status_reporter_test.py

@@ -24,7 +24,7 @@ def test_when_update_handlers_exist_their_channel_names_are_reported_in_status()
     }
 
     fake_producer = FakeProducer()
-    status_reporter = StatusReporter(update_handlers, fake_producer, "status_topic", "", "version", logger)  # type: ignore
+    status_reporter = StatusReporter(update_handlers, fake_producer, "status_topic", "", "version", logger, 4000)  # type: ignore
     status_reporter.report_status()
 
     if fake_producer.published_payloads:
@@ -43,7 +43,7 @@ def test_when_no_update_handlers_exist_no_streams_are_present_in_reported_status
     update_handlers: Dict = {}
 
     fake_producer = FakeProducer()
-    status_reporter = StatusReporter(update_handlers, fake_producer, "status_topic", "", "version", logger)  # type: ignore
+    status_reporter = StatusReporter(update_handlers, fake_producer, "status_topic", "", "version", logger, 4000)  # type: ignore
     status_reporter.report_status()
 
     if fake_producer.published_payloads:
@@ -59,7 +59,7 @@ def test_status_message_contains_service_id():
     update_handlers: Dict = {}
 
     fake_producer = FakeProducer()
-    status_reporter = StatusReporter(update_handlers, fake_producer, "status_topic", service_id, "version", logger)  # type: ignore
+    status_reporter = StatusReporter(update_handlers, fake_producer, "status_topic", service_id, "version", logger, 4000)  # type: ignore
     status_reporter.report_status()
 
     if fake_producer.published_payloads: