2 Contains rmr functionality specific to the xapp
3 The general rmr API is via "rmr"
5 # ==================================================================================
6 # Copyright (c) 2020 Nokia
7 # Copyright (c) 2020 AT&T Intellectual Property.
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
13 # http://www.apache.org/licenses/LICENSE-2.0
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 # ==================================================================================
25 from threading import Thread
26 from mdclogpy import Logger
27 from rmr import rmr, helpers
30 mdc_logger = Logger(name=__name__)
35 Class represents an rmr loop that constantly reads from rmr
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
40 def __init__(self, port, wait_for_ready=True):
42 sets up rmr, then launches a thread that reads and injects messages into a queue
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*
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()
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)
64 mdc_logger.debug("Waiting for rmr to init on port {}..".format(port))
65 while rmr.rmr_ready(self.mrc) == 0:
69 self._keep_going = True
70 self._last_ran = time.time()
73 mdc_logger.debug("Starting loop thread")
76 mdc_logger.debug("Work loop starting")
77 while self._keep_going:
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)
84 for (msg, sbuf) in helpers.rmr_rcvall_msgs_raw(self.mrc):
85 self.rcv_queue.put((msg, sbuf))
87 self._last_ran = time.time()
89 self._thread = Thread(target=loop)
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.
97 self._keep_going = False
99 def healthcheck(self, seconds=30):
101 returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
103 2. is it stuck in a long (> seconds) loop?
107 seconds: int (optional)
108 the rmr loop is determined healthy if it has completed in the last (seconds)
110 return self._thread.is_alive() and ((time.time() - self._last_ran) < seconds)