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 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 # used to tell this thread to stop it's work
70 self._last_ran = time.time() # used for healthcheck
71 self._loop_is_running = False # used in stop to know when it's safe to kill the mrc
74 mdc_logger.debug("Starting loop thread")
77 mdc_logger.debug("Work loop starting")
78 self._loop_is_running = True
79 while self._keep_going:
82 # TODO: take a flag as to whether RAW is needed or not
83 # RAW allows for RTS however the caller must free, and the caller may not need RTS.
84 # Currently after consuming, callers should do rmr.rmr_free_msg(sbuf)
86 for (msg, sbuf) in helpers.rmr_rcvall_msgs_raw(self.mrc):
87 self.rcv_queue.put((msg, sbuf))
89 self._last_ran = time.time()
91 self._loop_is_running = False
93 self._thread = Thread(target=loop)
98 sets a flag that will cleanly stop the thread
100 mdc_logger.debug("Stopping rmr thread. Waiting for last iteration to finish..")
101 self._keep_going = False
102 # wait until the current batch of messages is done, then kill the rmr connection
103 # note; I debated putting this in "loop" however if the while loop was still going setting mrc to close here would blow up any processing still currently happening
104 # probably more polite to at least finish the current batch and then close. So here we wait until the current batch is done, then we kill the mrc
105 while self._loop_is_running:
107 mdc_logger.debug("Closing rmr connection")
108 rmr.rmr_close(self.mrc)
110 def healthcheck(self, seconds=30):
112 returns a boolean representing whether the rmr loop is healthy, by checking two attributes:
114 2. is it stuck in a long (> seconds) loop?
118 seconds: int (optional)
119 the rmr loop is determined healthy if it has completed in the last (seconds)
121 return self._thread.is_alive() and ((time.time() - self._last_ran) < seconds)