xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-09-21 08:33:23 +00:00
Code Issues Projects Releases Wiki Activity

Fix nodediscover register and subscribe in the async port

Browse Source 
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
This commit is contained in:
Markus Hilger
2026-07-27 01:50:24 +02:00
parent 3f2ad75b6d
commit 4197bd9118
1 changed files with 8 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_client/bin/nodediscover
+8 -8
View File
@@ -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")