From 4197bd911881689827962d684638e04119512213 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:50:24 +0200 Subject: [PATCH] Fix nodediscover register and subscribe in the async port register_endpoint and subscribe_discovery were left as plain functions iterating the async client generators, so nodediscover register, subscribe and unsubscribe all failed immediately with TypeError: 'async_generator' object is not iterable --- confluent_client/bin/nodediscover | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index f4e2691a..78a45163 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -51,10 +51,10 @@ columnmapping = { } #TODO: add chassis uuid -def register_endpoint(options, session, addr): +async def register_endpoint(options, session, addr): neednewline = False current = 0 - for rsp in session.update('/discovery/register', {'addresses': addr}): + async for rsp in session.update('/discovery/register', {'addresses': addr}): if 'count' in rsp: total = rsp['count'] elif total > 1: @@ -70,15 +70,15 @@ def register_endpoint(options, session, addr): if neednewline: print('') -def subscribe_discovery(options, session, subscribe, targ): +async def subscribe_discovery(options, session, subscribe, targ): keyn = 'subscribe' if subscribe else 'unsubscribe' payload = {keyn: targ} if subscribe: - for rsp in session.update('/discovery/subscriptions/{0}'.format(targ), payload): + async for rsp in session.update('/discovery/subscriptions/{0}'.format(targ), payload): if 'status' in rsp: print(rsp['status']) else: - for rsp in session.delete('/discovery/subscriptions/{0}'.format(targ)): + async for rsp in session.delete('/discovery/subscriptions/{0}'.format(targ)): if 'status' in rsp: print(rsp['status']) @@ -427,11 +427,11 @@ async def main(): if args[0] == 'reassign': await assign_discovery(options, session, False) if args[0] == 'register': - register_endpoint(options, session, args[1]) + await register_endpoint(options, session, args[1]) if args[0] == 'subscribe': - subscribe_discovery(options, session, True, args[1]) + await subscribe_discovery(options, session, True, args[1]) if args[0] == 'unsubscribe': - subscribe_discovery(options, session, False, args[1]) + await subscribe_discovery(options, session, False, args[1]) if args[0] == 'rescan': await blocking_scan(session) print("Rescan complete")