1 # ==================================================================================
2 # Copyright (c) 2020 Nokia
3 # Copyright (c) 2020 AT&T Intellectual Property.
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 # ==================================================================================
19 Contains RMR functionality specific to the xapp.
20 The general rmr API is via "rmr"
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)