- return mrc
-
-
-def _send(mrc, payload, message_type=0):
- """
- Sends a message up to RETRY_TIMES
- If the message is sent successfully, it returns the transactionid
- Does nothing otherwise
- """
- # TODO: investigate moving this below and allocating the space based on the payload size
- sbuf = rmr.rmr_alloc_msg(mrc, 4096)
- payload = payload if isinstance(payload, bytes) else payload.encode("utf-8")
-
- # retry RETRY_TIMES to send the message
- for _ in range(0, RETRY_TIMES):
- # setup the send message
- rmr.set_payload_and_length(payload, sbuf)
- rmr.generate_and_set_transaction_id(sbuf)
- sbuf.contents.state = 0
- sbuf.contents.mtype = message_type
- pre_send_summary = rmr.message_summary(sbuf)
- logger.debug("Pre message send summary: %s", pre_send_summary)
- transaction_id = pre_send_summary["transaction id"] # save the transactionid because we need it later
-
- # send
- sbuf = rmr.rmr_send_msg(mrc, sbuf)
- post_send_summary = rmr.message_summary(sbuf)
- logger.debug("Post message send summary: %s", rmr.message_summary(sbuf))
-
- # check success or failure
- if post_send_summary["message state"] == 0 and post_send_summary["message status"] == "RMR_OK":
- # we are good
- logger.debug("Message sent successfully!")
- rmr.rmr_free_msg(sbuf)
- return transaction_id
-
- # we failed all RETRY_TIMES
- logger.debug("Send failed all %s times, stopping", RETRY_TIMES)
- rmr.rmr_free_msg(sbuf)
- return None
-
-
-# Public
-
-
-def queue_work(item):
- """
- push an item into the work queue
- currently the only type of work is to send out messages
- """
- _SEND_QUEUE.put(item)
-
-
-class RmrLoop:
- """
- class represents an rmr loop meant to be called as a longstanding separate thread
- """
-
- def __init__(self, _init_func_override=None, rcv_func_override=None):
- self._rmr_is_ready = False
- self._keep_going = True
- self._init_func_override = _init_func_override # useful for unit testing
- self._rcv_func_override = rcv_func_override # useful for unit testing to mock certain recieve scenarios
- self._rcv_func = None
-
- def rmr_is_ready(self):
- """returns whether rmr has been initialized"""
- return self._rmr_is_ready
-
- def stop(self):
- """sets a flag for the loop to end"""
- self._keep_going = False
+ def __init__(self, init_func_override=None, rcv_func_override=None):
+ self.keep_going = True
+ self.rcv_func = None
+ self.last_ran = time.time()
+ self.work_queue = queue.Queue() # thread safe queue https://docs.python.org/3/library/queue.html
+
+ # intialize rmr context
+ if init_func_override:
+ self.mrc = init_func_override()
+ else:
+ logger.debug("Waiting for rmr to initialize..")
+ # rmr.RMRFL_MTCALL puts RMR into a multithreaded mode, where a receiving thread populates an
+ # internal ring of messages, and receive calls read from that
+ # currently the size is 2048 messages, so this is fine for the foreseeable future
+ self.mrc = rmr.rmr_init(b"4562", rmr.RMR_MAX_RCV_BYTES, rmr.RMRFL_MTCALL)
+ while rmr.rmr_ready(self.mrc) == 0:
+ time.sleep(0.5)
+
+ # set the receive function
+ self.rcv_func = rcv_func_override if rcv_func_override else lambda: helpers.rmr_rcvall_msgs(self.mrc, [21024])
+
+ # start the work loop
+ self.thread = Thread(target=self.loop)
+ self.thread.start()