dnet: reimplement node dis(connect) handling

bin/dnet/dnet

@@ -26,6 +26,8 @@ from src.model import Model
 from src.rpc import JsonRpc
 from src.view import View
 
+# TODO: update to latest urwid version.
+
 class Dnetview:
 
     def __init__(self):
@@ -51,44 +53,67 @@ class Dnetview:
         type = node['type']
         info = {}
 
-        while True:
-            try:
-                await rpc.start(host, port)
-                logging.debug(f'Started {name} RPC on port {port}')
-                break
-            except Exception as e:
-                info[name] = {}
-                await self.queue.put(info)
-                continue
-    
-        if type == 'NORMAL':
-            data = await rpc._make_request('p2p.get_info', [])
-            info[name] = data
+        await self.connect_loop(rpc, node, info)
 
-            await self.queue.put(info)
-            await rpc.dnet_switch(True)
-            await rpc.dnet_subscribe_events()
-            
+        if type == 'NORMAL':
+            await self.get_info(rpc, node, info)
             while True:
                 await asyncio.sleep(0.01)
                 data = await rpc.reader.readline()
                 try:
                     data = json.loads(data)
-                    info[name] = data
+                    info[name] = (type, data)
                     await self.queue.put(info)
                 except:
-                    info[name] = {}
-                    await self.queue.put(info)
+                    logging.debug(f'{name} RPC on port {port} disconnected')
+                    # Attempt reconnection
+                    await self.connect_loop(rpc, node, info)
+                    await self.get_info(rpc, node, info)
 
             await rpc.dnet_switch(False)
     
         if type == 'LILITH':
             data = await rpc._make_request('spawns', [])
-            info[name] = data
+            info[name] = (type, data)
             await self.queue.put(info)
 
         await rpc.stop()
 
+    async def get_info(self, rpc, node, info):
+        name = node['name']
+        host = node['host']
+        port = node['port']
+        type = node['type']
+
+        logging.debug(f"get_info START")
+        data = await rpc._make_request('p2p.get_info', [])
+        logging.debug(f"sent rpc request")
+        info[name] = (type, data)
+
+        await self.queue.put(info)
+        await rpc.dnet_switch(True)
+        await rpc.dnet_subscribe_events()
+            
+    async def connect_loop(self, rpc, node, info):
+        name = node['name']
+        host = node['host']
+        port = node['port']
+        type = node['type']
+
+        while True:
+            try:
+                await rpc.start(host, port)
+                logging.debug(f'Started {host} RPC on port {port}')
+                break
+            except Exception as e:
+                logging.debug(f'Cannot connect to {host} RPC on port {port}')
+                info[name] = (type, {})
+                await self.queue.put(info)
+
+                # Sleep for 2s before trying again
+                await asyncio.sleep(2)
+                continue
+
     async def start_connect_slots(self, nodes):
         tasks = []
         async with asyncio.TaskGroup() as tg:
@@ -100,11 +125,22 @@ class Dnetview:
 
     async def update_info(self):
         while True:
-            info = await self.queue.get()
-            values = list(info.values())[0]
-
-            if not values:
-                self.model.add_offline(info)
+            info = {}
+            data = await self.queue.get()
+
+            #logging.debug(f"update_info(): Raw data {data}")
+            # We parse in this manner so we can differentiate between
+            # normal nodes and lilith nodes when they are offline.
+            type = list(data.values())[0][0]
+            values = list(data.values())[0][1]
+            key = list(data.keys())[0]
+            info[key] = values
+            #logging.debug(f"update_info(): Parsed data {type} {info}")
+
+            if not values and type == 'LILITH':
+                self.model.add_offline(info, True)
+            if not values and type == 'NORMAL':
+                self.model.add_offline(info, False)
 
             if 'result' in values:
                 result = values.get('result')

bin/dnet/src/model.py

@@ -74,10 +74,13 @@ class Model:
             url = channel['url']
             self.nodes[name]['manual'][f'{id}'] = url
     
-    def add_offline(self, node):
+    def add_offline(self, node, is_lilith: bool):
         name = list(node.keys())[0]
         values = list(node.values())[0]
-        self.nodes[name] = values
+        if is_lilith:
+            self.liliths[name] = values
+        else:
+            self.nodes[name] = values
 
     def add_event(self, event):
         name = list(event.keys())[0]
@@ -153,7 +156,6 @@ class Model:
 
 
     def add_lilith(self, lilith):
-        #logging.debug(f'adding lilith {lilith}')
         key = list(lilith.keys())[0]
         values = list(lilith.values())[0]
         info = values['result']

bin/dnet/src/rpc.py

@@ -25,7 +25,7 @@ import asyncio
 class JsonRpc:
 
     async def start(self, host, port):
-        logging.info(f"trying to connect to {host}:{port}")
+        #logging.info(f"trying to connect to {host}:{port}")
         reader, writer = await asyncio.open_connection(host, port)
         self.reader = reader
         self.writer = writer

bin/dnet/src/view.py

@@ -19,13 +19,11 @@ import urwid
 import logging
 import asyncio
 import datetime as dt
+from enum import Enum
 
 from src.scroll import ScrollBar, Scrollable
 from src.model import Model
 
-# TODO: re-implement offline nodes comes online, online node goes offline
-# handling.
-# TODO: update to latest urwid version.
 class DnetWidget(urwid.WidgetWrap):
     def __init__(self, name, kind):
         self.name = name
@@ -42,17 +40,23 @@ class DnetWidget(urwid.WidgetWrap):
         self._w = urwid.AttrWrap(self._w, None)
         self._w.focus_attr = 'line'
 
+class NodeState(Enum):
+    ON = 0
+    OFF = 1
 
 class Node(DnetWidget):
-    def set_txt(self, is_empty: bool):
-        if is_empty:
+    def __init__(self, name, kind, state):
+        self.state = state
+        super().__init__(name, kind)
+
+    def set_txt(self):
+        if self.state == NodeState.OFF:
             txt = urwid.Text(f"{self.name} (offline)")
             super().update(txt)
         else:
             txt = urwid.Text(f"{self.name}")
             super().update(txt)
 
-
 class Session(DnetWidget):
     def set_txt(self):
         txt = urwid.Text(f"  {self.kind}")
@@ -93,20 +97,16 @@ class View():
         leftbox = urwid.LineBox(self.list)
         columns = urwid.Columns([leftbox, rightbox], focus_column=0)
         self.ui = urwid.Frame(urwid.AttrWrap( columns, 'body' ))
-        self.active_sessions = set()
-        self.active_nodes = set()
+        self.sessions = set()
+        self.nodes = set()
         self.refresh_needed = False
 
-    def add_node(self, name, info):
-        logging.debug("add_node() [START]")
-        node = Node(name, "node")
-        if not info:
-            node.set_txt(True)
-        else:
-            node.set_txt(False)
+    def add_node(self, name, info, state):
+        logging.debug(f"Adding node: {name} {info} {state}")
+        node = Node(name, "node", state)
+        node.set_txt()
+        self.nodes.add(name)
         self.listwalker.append(node)
-        self.active_nodes.add(name)
-
         self.add_sessions(name, info)
 
     def add_sessions(self, name, info):
@@ -125,28 +125,46 @@ class View():
             match session:
                 case "outbound":
                     if addr[1] > 0:
-                        self.active_sessions.add(addr[1])
+                        self.sessions.add(addr[1])
                 case "inbound" | "manual " | "seed":
-                    self.active_sessions.add(i)
+                    self.sessions.add(i)
 
-    def add_lilith(self, name, info):
-        node = Node(name, "lilith-node")
+    def add_lilith(self, name, info, state):
+        logging.debug(f"Adding lilith: {name} {info} {state}")
+        node = Node(name, "lilith-node", state)
         node.set_txt()
+        self.nodes.add(name)
         self.listwalker.append(node)
-        self.active_nodes.add(name)
-        for (i, key) in enumerate(info['spawns'].keys()):
-            slot = Slot(name, "spawn-slot")
-            slot.set_txt(i, key)
-            self.listwalker.append(slot)
+        if state == NodeState.OFF:
+            return
+        else:
+            for (i, key) in enumerate(info['spawns'].keys()):
+                slot = Slot(name, "spawn-slot")
+                slot.set_txt(i, key)
+                self.listwalker.append(slot)
+
+    def update_lilith(self, name, info):
+        for index, widget in enumerate(self.listwalker):
+            if isinstance(widget, Node) and widget.name == name:
+                # Offline node has come online
+                if widget.state == NodeState.OFF and info:
+                        self.refresh_needed = True
+                # Online node has gone offline
+                elif widget.state == NodeState.ON and not info:
+                        self.refresh_needed = True
 
     def update_node(self, name, info):
         for index, widget in enumerate(self.listwalker):
             if isinstance(widget, Node) and widget.name == name:
-                if not info:
-                    widget.set_txt(True)
+                # Offline node has come online
+                if widget.state == NodeState.OFF and info:
+                        self.refresh_needed = True
+                # Online node has gone offline
+                elif widget.state == NodeState.ON and not info:
+                        self.refresh_needed = True
                 else:
-                    widget.set_txt(False)
-                return index + 1
+                    widget.set_txt()
+                    return index + 1
         return None
 
     def update_slots(self, name, info):
@@ -163,24 +181,13 @@ class View():
                     widget.name == name and \
                     widget.kind == f"{session}-slot" and \
                     widget.i == i:
-                widget.set_txt(i, addr)
-                break
-
-    #-----------------------------------------------------------------
-    # Render dnet.subscribe_events() RPC call 
-    # Left hand panel only
-    #-----------------------------------------------------------------
-    def update_left_box(self):
-        for index, item in enumerate(self.listwalker):
-            # Update outbound slot info
-            if item.kind == "outbound-slot":
-                key = (f"{item.name}", f"{item.i}")
-                if key in self.model.nodes[item.name]['event']:
-                    info = self.model.nodes[item.name]['event'].get(key)
-                    slot = Slot(item.name, item.kind)
-                    slot.set_txt(item.i, info)
-                    self.listwalker[index] = slot
-
+                key = (f"{widget.name}", f"{widget.i}")
+                if key in self.model.nodes[widget.name]['event']:
+                    info = self.model.nodes[widget.name]['event'].get(key)
+                    widget.set_txt(i, info)
+                    self.listwalker[index] = widget
+                    break
+    
     #-----------------------------------------------------------------
     # Render dnet.subscribe_events() RPC call
     # Right hand menu only
@@ -217,65 +224,78 @@ class View():
                                 f"{time}: {event}: {msg}"),
                                 self.pile.options()))
             case "spawn-slot":
-                if session == "spawn-slot":
-                    name = focus_w[0].name
-                    spawn_name = focus_w[0].id
-                    lilith = self.model.liliths.get(name)
-                    spawns = lilith.get('spawns')
-                    info = spawns.get(spawn_name)
-
-                    if info['urls']:
-                        urls = info['urls']
+                name = focus_w[0].name
+                spawn_name = focus_w[0].id
+                lilith = self.model.liliths.get(name)
+                spawns = lilith.get('spawns')
+                info = spawns.get(spawn_name)
+
+                if info['urls']:
+                    urls = info['urls']
+                    self.pile.contents.append((urwid.Text(
+                        f"Accept addrs:"),
+                        self.pile.options()))
+                    for url in urls:
                         self.pile.contents.append((urwid.Text(
-                            f"Accept addrs:"),
+                            f"  {url}"),
                             self.pile.options()))
-                        for url in urls:
-                            self.pile.contents.append((urwid.Text(
-                                f"  {url}"),
-                                self.pile.options()))
 
-                    if info['whitelist']:
-                        whitelist = info['whitelist']
+                if info['whitelist']:
+                    whitelist = info['whitelist']
+                    self.pile.contents.append((urwid.Text(
+                        f"Whitelist:"),
+                        self.pile.options()))
+                    for host in whitelist:
                         self.pile.contents.append((urwid.Text(
-                            f"Whitelist:"),
+                            f"  {host}"),
                             self.pile.options()))
-                        for host in whitelist:
-                            self.pile.contents.append((urwid.Text(
-                                f"  {host}"),
-                                self.pile.options()))
 
-                    if info['greylist']:
-                        greylist = info['greylist']
+                if info['greylist']:
+                    greylist = info['greylist']
+                    self.pile.contents.append((urwid.Text(
+                        f"Greylist:"),
+                        self.pile.options()))
+                    for host in greylist:
                         self.pile.contents.append((urwid.Text(
-                            f"Greylist:"),
+                            f"  {host}"),
                             self.pile.options()))
-                        for host in greylist:
-                            self.pile.contents.append((urwid.Text(
-                                f"  {host}"),
-                                self.pile.options()))
 
-                    if info['goldlist']:
-                        goldlist = info['goldlist']
+                if info['goldlist']:
+                    goldlist = info['goldlist']
+                    self.pile.contents.append((urwid.Text(
+                        f"Goldlist:"),
+                        self.pile.options()))
+                    for host in goldlist:
                         self.pile.contents.append((urwid.Text(
-                            f"Goldlist:"),
+                            f"  {host}"),
                             self.pile.options()))
-                        for host in goldlist:
-                            self.pile.contents.append((urwid.Text(
-                                f"  {host}"),
-                                self.pile.options()))
+
+    def update_node_state(self, info):
+        if info:
+            logging.debug(f"update_node_state(): Returning {NodeState.ON}")
+            return NodeState.ON
+        else:
+            logging.debug(f"update_node_state(): Returning {NodeState.OFF}")
+            return NodeState.OFF
 
     def refresh(self):
         logging.debug("Refresh initiated.")
         self.listwalker.clear()
-        self.active_nodes.clear()
-        self.active_sessions.clear()
+        self.sessions.clear()
+        self.nodes.clear()
 
         # Repopulate
         for name, info in self.model.nodes.items():
-            logging.debug(f"refresh(): {name}, {info}")
-            self.add_node(name, info)
+            logging.debug(f"refresh nodes(): (nodes) updating node state")
+            state = self.update_node_state(info)
+            logging.debug(f"refresh nodes(): {name}, {info}, {state}")
+            self.add_node(name, info, state)
+
         for name, info in self.model.liliths.items():
-            self.add_lilith(name, info)
+            logging.debug(f"refresh nodes(): (lilith) updating node state")
+            state = self.update_node_state(info)
+            logging.debug(f"refresh liliths(): {name}, {info}, {state}")
+            self.add_lilith(name, info, state)
 
         logging.debug("Refresh complete.")
 
@@ -289,44 +309,47 @@ class View():
                 self.refresh_needed = False
             else:
                 for name, info in self.model.nodes.items():
-                    if info is None:
-                        continue
-
+                    #logging.debug(f"update_view(): found {name}, {info}")
                     # Check for new nodes or update existing slots.
-                    if name not in self.active_nodes:
-                        self.add_node(name, info)
+                    if name not in self.nodes:
+                        logging.debug(f"update_view(): (node) updating node state")
+                        state = self.update_node_state(info)
+                        self.add_node(name, info, state)
                     else:
                         start_index = self.update_node(name, info)
                         if start_index is not None:
                             self.update_slots(name, info)
-
-                    for name, info in self.model.liliths.items():
-                        if name not in self.active_nodes:
-                            self.add_lilith(name, info)
-
                     # Check for outbound or inbound connections coming
                     # online or going offline, which requires a redraw.
                     if 'outbound' in info:
                         for i, (addr, id) in info['outbound'].items():
-                            if id > 0 and id not in self.active_sessions:
+                            if id > 0 and id not in self.sessions:
                                 logging.debug(f"Outbound {id}, {addr} came online.")
                                 self.refresh_needed = True
                                 break
 
                     if 'inbound' in info:
                         for key, addr in info['inbound'].items():
-                            if key in self.active_sessions and not addr:
+                            if key in self.sessions and not addr:
                                 logging.debug(f"Inbound {key} went offline.")
                                 # Delete this key from the model.
                                 del(info['inbound'][f'{key}'])
                                 self.refresh_needed = True     
                                 break
-                            if key not in self.active_sessions:
+                            if key not in self.sessions:
                                 logging.debug(f"Inbound {key}, {addr} came online.")
                                 self.refresh_needed = True
                                 break
 
-            self.update_left_box()
-            self.update_right_box()
+                # Check for new lilith nodes. 
+                for name, info in self.model.liliths.items():
+                    if name not in self.nodes:
+                        logging.debug(f"update_view(): (lilith) updating node state")
+                        state = self.update_node_state(info)
+                        self.add_lilith(name, info, state)
+                    else:
+                        self.update_lilith(name, info)
 
+
+            self.update_right_box()
             evloop.call_soon(loop.draw_screen)