@@ -4,7 +4,6 @@
import argparse
import json
import pprint
-import time
from lib import YnlFamily, Netlink, NlError
@@ -17,7 +16,6 @@ class YnlEncoder(json.JSONEncoder):
return list(obj)
return json.JSONEncoder.default(self, obj)
-
def main():
description = """
YNL CLI utility - a general purpose netlink utility that uses YAML
@@ -43,7 +41,10 @@ def main():
group.add_argument('--list-ops', action='store_true')
group.add_argument('--list-msgs', action='store_true')
- parser.add_argument('--sleep', dest='sleep', type=int)
+ parser.add_argument('--duration', dest='duration', type=int,
+ help='when subscribed, watch for DURATION seconds')
+ parser.add_argument('--sleep', dest='duration', type=int,
+ help='alias for duration')
parser.add_argument('--subscribe', dest='ntf', type=str)
parser.add_argument('--replace', dest='flags', action='append_const',
const=Netlink.NLM_F_REPLACE)
@@ -80,9 +81,6 @@ def main():
if args.ntf:
ynl.ntf_subscribe(args.ntf)
- if args.sleep:
- time.sleep(args.sleep)
-
if args.list_ops:
for op_name, op in ynl.ops.items():
print(op_name, " [", ", ".join(op.modes), "]")
@@ -106,8 +104,8 @@ def main():
exit(1)
if args.ntf:
- ynl.check_ntf()
- output(ynl.async_msg_queue)
+ for msg in ynl.poll_ntf(duration=args.duration):
+ 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,7 +905,7 @@ 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):
while True:
@@ -925,11 +927,25 @@ class YnlFamily(SpecFamily):
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)
+ print("Unexpected msg id while checking for ntf", decoded)
continue
self.handle_ntf(decoded)
+ def poll_ntf(self, interval=0.1, duration=None):
+ endtime = time.time() + duration if duration else None
+ while True:
+ try:
+ self.check_ntf()
+ yield self.async_msg_queue.get_nowait()
+ except queue.Empty:
+ try:
+ time.sleep(interval)
+ except KeyboardInterrupt:
+ return
+ if endtime and endtime < time.time():
+ return
+
def operation_do_attributes(self, name):
"""
For a given operation name, find and return a supported
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 adds async notification handling so that messages can be processed as they are received. This makes it possible to use ynl as a library that supplies notifications in a timely manner. - Add poll_ntf() to be a generator that yields 1 notification at a time and blocks until a notification is available. - Add a --duration parameter to the CLI, with --sleep as an alias. ./tools/net/ynl/cli.py \ --spec <SPEC> --subscribe <TOPIC> [ --duration <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.poll_ntf(): handle(event) Signed-off-by: Donald Hunter <donald.hunter@gmail.com> --- tools/net/ynl/cli.py | 14 ++++++-------- tools/net/ynl/lib/ynl.py | 22 +++++++++++++++++++--- 2 files changed, 25 insertions(+), 11 deletions(-)