Merge pull request #31 from neutrons/stomp.py

Move from stompest to stomp.py and to python 3.9

.github/workflows/actions.yml

@@ -86,23 +86,24 @@ jobs:
       env:
         CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
 
-  rpm:
-    runs-on: ubuntu-latest
-
-    steps:
-    - name: Checkout Repository
-      uses: actions/checkout@v2
-
-    - name: Build RPM inside Docker
-      run: |
-        docker build --tag postprocess --target=package -f Dockerfile .
-        fname=`docker run -v $(pwd):/store postprocess ls /root/rpmbuild/RPMS/noarch`
-        docker run -v `pwd`:/store postprocess mv /root/rpmbuild/RPMS/noarch/$fname /store
-        one=${fname#*postprocessing-}
-        two=${one%.noarch*}
-        echo "::set-output name=version::$two"
-        echo "::set-output name=fname::$fname"
-      continue-on-error: false
+# disable until spec file is updated to python 3
+#  rpm:
+#    runs-on: ubuntu-latest
+#
+#    steps:
+#    - name: Checkout Repository
+#      uses: actions/checkout@v2
+#
+#    - name: Build RPM inside Docker
+#      run: |
+#        docker build --tag postprocess --target=package -f Dockerfile .
+#        fname=`docker run -v $(pwd):/store postprocess ls /root/rpmbuild/RPMS/noarch`
+#        docker run -v `pwd`:/store postprocess mv /root/rpmbuild/RPMS/noarch/$fname /store
+#        one=${fname#*postprocessing-}
+#        two=${one%.noarch*}
+#        echo "::set-output name=version::$two"
+#        echo "::set-output name=fname::$fname"
+#      continue-on-error: false
 
 # TODO: uncomment this once we have switched to python3 and can conda install the module build
 #  wheel:

configuration/post_process_consumer.conf.development

@@ -1,5 +1,6 @@
 {
     "failover_uri": "failover:(tcp://localhost:61613)?randomize=false,startupMaxReconnectAttempts=100,initialReconnectDelay=1000,maxReconnectDelay=5000,maxReconnectAttempts=-1",
+    "brokers": [["localhost", 61613]],
     "amq_queues": ["/queue/REDUCTION.DATA_READY"],
     "amq_user": "icat",
     "amq_pwd": "icat",

configuration/post_process_consumer.conf.local

@@ -1,5 +1,6 @@
 {
     "failover_uri": "failover:(tcp://localhost:61613)?randomize=false,startupMaxReconnectAttempts=100,initialReconnectDelay=1000,maxReconnectDelay=5000,maxReconnectAttempts=-1",
+    "brokers": [("localhost", 61613)],
     "amq_queues": ["/queue/REDUCTION.CREATE_SCRIPT", "/queue/REDUCTION.DATA_READY", "/queue/CATALOG.DATA_READY", "/queue/REDUCTION_CATALOG.DATA_READY"],
     "amq_user": "",
     "amq_pwd": "",

environment.yml

@@ -3,17 +3,13 @@ channels:
   - conda-forge
   - defaults
 dependencies:
-  - python=3.6
-  - pip
+  - python=3.9
   - plotly
   - pre-commit
   - pytest
   - pytest-cov
   - pytest-mock
-  - requests
-  - twisted
-  - pip: # will need to be installed separately if pip is broken
-    - stompest
-    - stompest-async
+  - requests=2.25
+  - stomp.py=7
 # needed for wheel - add when switching to python3
 #  - build

postprocessing/Configuration.py

@@ -13,18 +13,18 @@
 import importlib
 
 
-class Configuration(object):
+class Configuration:
     """
     Read and process configuration file and provide an easy way to create a configured Client object
     """
 
     def __init__(self, config_file):
         if os.access(config_file, os.R_OK) is False:
             raise RuntimeError(
-                "Configuration file doesn't exist or is not readable: %s" % config_file
+                f"Configuration file doesn't exist or is not readable: {config_file}"
             )
-        cfg = open(config_file, "r")
-        json_encoded = cfg.read()
+        with open(config_file, "r") as cfg:
+            json_encoded = cfg.read()
         config = json.loads(json_encoded)
 
         # Keep a record of which config file we are using
@@ -34,6 +34,7 @@ def __init__(self, config_file):
         self.amq_pwd = config["amq_pwd"]
         # ActiveMQ broker information
         self.failover_uri = config["failover_uri"]
+        self.brokers = [(host, port) for host, port in config["brokers"]]
         self.queues = config["amq_queues"]
         self.sw_dir = config["sw_dir"] if "sw_dir" in config else "/opt/postprocessing"
         self.postprocess_error = config["postprocess_error"]
@@ -93,7 +94,7 @@ def __init__(self, config_file):
             config["dev_output_dir"].strip() if "dev_output_dir" in config else ""
         )
         self.python_executable = (
-            config["python_exec"] if "python_exec" in config else "python"
+            config["python_exec"] if "python_exec" in config else "python3"
         )
 
         self.max_procs = config["max_procs"] if "max_procs" in config else 5
@@ -139,10 +140,10 @@ def __init__(self, config_file):
                 if len(toks) == 2:
                     # for instance, emulate `from oncat_processor import ONCatProcessor`
                     processor_module = importlib.import_module(  # noqa: F841
-                        "postprocessing.processors.%s" % toks[0]
+                        f"postprocessing.processors.{toks[0]}"
                     )
                     try:
-                        processor_class = eval("processor_module.%s" % toks[1])
+                        processor_class = eval(f"processor_module.{toks[1]}")
                         self.queues.append(processor_class.get_input_queue_name())
                     except:  # noqa: E722
                         logging.error(
@@ -172,12 +173,7 @@ def log_configuration(self, logger=logging):
         logger.info("  - Error exceptions: %s", str(self.exceptions))
 
 
-# Set the log level for the Stomp client
-stomp_logger = logging.getLogger("stompest.sync.client")
-stomp_logger.setLevel(logging.ERROR)
-
-
-class StreamToLogger(object):
+class StreamToLogger:
     r"""File-like stream object that redirects writes to a Logger instance."""
 
     def __init__(self, logger, log_level=logging.INFO):
@@ -256,8 +252,7 @@ def read_configuration(
                 break
         else:
             raise RuntimeError(
-                "Default configuration file(s) do not exist, or unreadable: %s"
-                % str(defaults)
+                f"Default configuration file(s) do not exist, or unreadable: {defaults}"
             )
 
     configuration = Configuration(config_file)