ricxappframe/xapp_rmr.py

   1 """
   2 Contains rmr functionality specific to the xapp
   3 The general rmr API is via "rmr"
   4 """
   5 # ==================================================================================
   6 #       Copyright (c) 2020 Nokia
   7 #       Copyright (c) 2020 AT&T Intellectual Property.
   8 #
   9 #   Licensed under the Apache License, Version 2.0 (the "License");
  10 #   you may not use this file except in compliance with the License.
  11 #   You may obtain a copy of the License at
  12 #
  13 #          http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 #   Unless required by applicable law or agreed to in writing, software
  16 #   distributed under the License is distributed on an "AS IS" BASIS,
  17 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 #   See the License for the specific language governing permissions and
  19 #   limitations under the License.
  20 # ==================================================================================
  21
  22
  23 import time
  24 import queue
  25 from threading import Thread
  26 from mdclogpy import Logger
  27 from rmr import rmr, helpers
  28
  29
  30 mdc_logger = Logger(name=__name__)
  31
  32
  33 class RmrLoop:
  34     """
  35     Class represents an rmr loop that constantly reads from rmr
  36
  37     Note, we use a queue here, and a thread, rather than the xapp frame just looping and calling consume, so that a possibly slow running consume function does not block the reading of new messages
  38     """
  39
  40     def __init__(self, port, wait_for_ready=True):
  41         """
  42         sets up rmr, then launches a thread that reads and injects messages into a queue
  43
  44         Parameters
  45         ----------
  46         port: int
  47             port to listen on
  48
  49         wait_for_ready: bool (optional)
  50             if this is True, then this function hangs until rmr is ready to send, which includes having a valid routing file.
  51             this can be set to False if the client only wants to *receive only*
  52         """
  53
  54         # Public
  55         # thread safe queue https://docs.python.org/3/library/queue.html
  56         # We use a thread and a queue so that a long running consume callback function can never block reads.
  57         # IE a consume implementation could take a long time and the ring size for rmr blows up here and messages are lost
  58         self.rcv_queue = queue.Queue()
  59
  60         # rmr context; RMRFL_MTCALL puts RMR into a multithreaded mode, where a thread populates a ring of messages that receive calls read from
  61         self.mrc = rmr.rmr_init(str(port).encode(), rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
  62
  63         if wait_for_ready:
  64             mdc_logger.debug("Waiting for rmr to init on port {}..".format(port))
  65             while rmr.rmr_ready(self.mrc) == 0:
  66                 time.sleep(0.1)
  67
  68         # Private
  69         self._keep_going = True
  70         self._last_ran = time.time()
  71
  72         # start the work loop
  73         mdc_logger.debug("Starting loop thread")
  74
  75         def loop():
  76             mdc_logger.debug("Work loop starting")
  77             while self._keep_going:
  78
  79                 # read our mailbox
  80                 # TODO: take a flag as to whether RAW is needed or not
  81                 # RAW allows for RTS however the caller must free, and the caller may not need RTS.
  82                 # Currently after consuming, callers should do  rmr.rmr_free_msg(sbuf)
  83
  84                 for (msg, sbuf) in helpers.rmr_rcvall_msgs_raw(self.mrc):
  85                     self.rcv_queue.put((msg, sbuf))
  86
  87                 self._last_ran = time.time()
  88
  89         self._thread = Thread(target=loop)
  90         self._thread.start()
  91
  92     def stop(self):
  93         """
  94         sets a flag that will cleanly stop the thread
  95         note, this does not yet have a use yet for xapps to call, however this is very handy during unit testing.
  96         """
  97         self._keep_going = False
  98
  99     def healthcheck(self, seconds=30):
 100         """
 101         returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
 102         1. is it running?,
 103         2. is it stuck in a long (> seconds) loop?
 104
 105         Parameters
 106         ----------
 107         seconds: int (optional)
 108             the rmr loop is determined healthy if it has completed in the last (seconds)
 109         """
 110         return self._thread.is_alive() and ((time.time() - self._last_ran) < seconds)