@@ -5,6 +5,7 @@ import argparse
import json
import pprint
import time
+import signal
from lib import YnlFamily, Netlink, NlError
@@ -17,6 +18,8 @@ class YnlEncoder(json.JSONEncoder):
return list(obj)
return json.JSONEncoder.default(self, obj)
+def handle_timeout(sig, frame):
+ exit(0)
def main():
description = """
@@ -81,7 +84,8 @@ def main():
ynl.ntf_subscribe(args.ntf)
if args.sleep:
- time.sleep(args.sleep)
+ signal.signal(signal.SIGALRM, handle_timeout)
+ signal.alarm(args.sleep)
if args.list_ops:
for op_name, op in ynl.ops.items():
@@ -106,8 +110,8 @@ def main():
exit(1)
if args.ntf:
- ynl.check_ntf()
- output(ynl.async_msg_queue)
+ for msg in ynl.check_ntf():
+ output(msg)
if __name__ == "__main__":
@@ -12,6 +12,8 @@ import sys
import yaml
import ipaddress
import uuid
+import queue
+import time
from .nlspec import SpecFamily
@@ -489,7 +491,7 @@ class YnlFamily(SpecFamily):
self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_GET_STRICT_CHK, 1)
self.async_msg_ids = set()
- self.async_msg_queue = []
+ self.async_msg_queue = queue.Queue()
for msg in self.msgs.values():
if msg.is_async:
@@ -903,32 +905,39 @@ class YnlFamily(SpecFamily):
msg['name'] = op['name']
msg['msg'] = attrs
- self.async_msg_queue.append(msg)
+ self.async_msg_queue.put(msg)
- def check_ntf(self):
+ def check_ntf(self, interval=0.1):
while True:
try:
reply = self.sock.recv(self._recv_size, socket.MSG_DONTWAIT)
- except BlockingIOError:
- return
+ nms = NlMsgs(reply)
+ self._recv_dbg_print(reply, nms)
+ for nl_msg in nms:
+ if nl_msg.error:
+ print("Netlink error in ntf!?", os.strerror(-nl_msg.error))
+ print(nl_msg)
+ continue
+ if nl_msg.done:
+ print("Netlink done while checking for ntf!?")
+ continue
- nms = NlMsgs(reply)
- self._recv_dbg_print(reply, nms)
- for nl_msg in nms:
- if nl_msg.error:
- print("Netlink error in ntf!?", os.strerror(-nl_msg.error))
- print(nl_msg)
- continue
- if nl_msg.done:
- print("Netlink done while checking for ntf!?")
- continue
+ decoded = self.nlproto.decode(self, nl_msg, None)
+ if decoded.cmd() not in self.async_msg_ids:
+ print("Unexpected msg id while checking for ntf", decoded)
+ continue
- decoded = self.nlproto.decode(self, nl_msg, None)
- if decoded.cmd() not in self.async_msg_ids:
- print("Unexpected msg id done while checking for ntf", decoded)
- continue
+ self.handle_ntf(decoded)
+ except BlockingIOError:
+ pass
- self.handle_ntf(decoded)
+ try:
+ yield self.async_msg_queue.get_nowait()
+ except queue.Empty:
+ try:
+ time.sleep(interval)
+ except KeyboardInterrupt:
+ return
def operation_do_attributes(self, name):
"""
The notification handling in ynl is currently very simple, using sleep() to wait a period of time and then handling all the buffered messages in a single batch. This patch changes the notification handling so that messages are processed as they are received. This makes it possible to use ynl as a library that supplies notifications in a timely manner. - Change check_ntf() to be a generator that yields 1 notification at a time and blocks until a notification is available. - Use the --sleep parameter to set an alarm and exit when it fires. This means that the CLI has the same interface, but notifications get printed as they are received: ./tools/net/ynl/cli.py --spec <SPEC> --subscribe <TOPIC> [ --sleep <SECS> ] Here is an example python snippet that shows how to use ynl as a library for receiving notifications: ynl = YnlFamily(f"{dir}/rt_route.yaml") ynl.ntf_subscribe('rtnlgrp-ipv4-route') for event in ynl.check_ntf(): handle(event) Signed-off-by: Donald Hunter <donald.hunter@gmail.com> --- tools/net/ynl/cli.py | 10 +++++--- tools/net/ynl/lib/ynl.py | 49 ++++++++++++++++++++++++---------------- 2 files changed, 36 insertions(+), 23 deletions(-)