How to handle event messages asynchronously in Python?

[grondo]

In #4569 @vsoch asked:

I'd like to be able to add arbitrary events that are rpc (not necessarily just the job journal) to be triggered with handle.reactor_run() and then allow me to trigger a callback. For example, for a custom heartbeat, I can subscribe to it and receive the message, but I have to do that synchronously and do some loop that blocks (and prevents me from running my one call to the reactor). My solution now is not to use flux and create a separate thread, but I'd ideally like everything (events wise) coming from one source, and from flux.

[grondo]

To handle event messages in Flux, you first have to subscribe to the event topic of interest with handle.event_subscribe. Then create a message watcher for the EVENT type message with handle.msg_watcher_create(). Finally start the message handler, then enter the reactor.

Here's a simple example that subscribes to heartbeat.pulse and calls a callback for each event:

import flux
  
#  Message handler callback:
#  Args:
#   handle: Flux handle on which the msg was receieved
#   msg_handler: this message handler
#   msg: the message
#   arg, arbitrary argument passed to args parameter of msg_watcher_create()
#
def hb_cb(handle, msg_handler, msg, arg):
    print(f"got {msg.topic}")

handle = flux.Flux()

#  Subscribe to heartbeat.pulse event
handle.event_subscribe("heartbeat.pulse")

#  Create a msg handler for the heartbeat.pulse event messages
msg_handler = handle.msg_watcher_create(
            hb_cb, flux.constants.FLUX_MSGTYPE_EVENT, "heartbeat.pulse"
)
#  The message handler is not enabled until .start() is called.
#  (The msg handler can be stopped with .stop()
msg_handler.start()

#  Run the reactor. The reactor will not stop until forcibly stopped with
#  handle.reactor_stop() or if the message handler is stopped with
#  msg_handler.stop() (if there are no other watchers enabled on this
#  handle)
handle.reactor_run()

[vsoch]

Thank you - this worked perfectly! My mistake was not knowing to start the watcher. The other bit of wisdom you provided in slack is how to customize the period. In this example, the self.cfg (config) heartbeat is a custom number of seconds. Note that can be another temporal unit (e.g., m for minute) to define the period.

# Customize the heartbeat duration by reloading the module
self.handle.rpc("module.remove", {"name": "heartbeat"}).get()
self.handle.rpc(
     "module.load", {"path": "heartbeat", "args": [f"period={self.cfg.heartbeat}s"]}
).get()

I did that before the event subscribe.