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 ricxappframe.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
38 and calling consume, so that a possibly slow running consume function does not
39 block the reading of new messages
42 def __init__(self, port, wait_for_ready=True):
44 sets up RMR, then launches a thread that reads and injects messages into a queue.
51 wait_for_ready: bool (optional)
52 If True, then this function hangs until RMR is ready to send, which includes
53 having a valid routing file. This can be set to False if the client only wants
58 # thread safe queue https://docs.python.org/3/library/queue.html
59 # We use a thread and a queue so that a long running consume callback function can
60 # never block reads. IE a consume implementation could take a long time and the ring
61 # size for rmr blows up here and messages are lost.
62 self.rcv_queue = queue.Queue()
64 # RMR context; RMRFL_MTCALL puts RMR into a multithreaded mode, where a thread
65 # populates a ring of messages that receive calls read from
66 self.mrc = rmr.rmr_init(str(port).encode(), rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
69 mdc_logger.debug("Waiting for rmr to init on port {}..".format(port))
70 while rmr.rmr_ready(self.mrc) == 0:
74 self._keep_going = True # used to tell this thread to stop
75 self._last_ran = time.time() # used for healthcheck
76 self._loop_is_running = False # used in stop to know when it's safe to kill the mrc
79 mdc_logger.debug("Work loop starts")
80 self._loop_is_running = True
81 while self._keep_going:
84 # TODO: take a flag as to whether RAW is needed or not
85 # RAW allows for RTS however the caller must free, and the caller may not need RTS.
86 # Currently after consuming, callers should do rmr.rmr_free_msg(sbuf)
88 for (msg, sbuf) in helpers.rmr_rcvall_msgs_raw(self.mrc, timeout=1000):
89 self.rcv_queue.put((msg, sbuf))
91 self._last_ran = time.time()
93 self._loop_is_running = False
94 mdc_logger.debug("Work loop ends")
97 mdc_logger.debug("Starting loop thread")
98 self._thread = Thread(target=loop)
103 sets a flag that will cleanly stop the thread
105 mdc_logger.debug("Stopping RMR thread. Waiting for last iteration to finish..")
106 self._keep_going = False
107 # wait until the current batch of messages is done, then kill the rmr connection
108 # note; I debated putting this in "loop" however if the while loop was still going
109 # setting mrc to close here would blow up any processing still currently happening
110 # probably more polite to at least finish the current batch and then close. So here
111 # we wait until the current batch is done, then we kill the mrc
112 while self._loop_is_running:
114 mdc_logger.debug("Closing rmr connection")
115 rmr.rmr_close(self.mrc)
117 def healthcheck(self, seconds=30):
119 returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
121 2. is it stuck in a long (> seconds) loop?
125 seconds: int (optional)
126 the rmr loop is determined healthy if it has completed in the last (seconds)
128 return self._thread.is_alive() and ((time.time() - self._last_ran) < seconds)