Refactoring netcode so that background crawling can be implemented

src/cutecoin/core/community.py

@@ -9,7 +9,8 @@ from ucoinpy import PROTOCOL_VERSION
 from ucoinpy.documents.peer import Peer, Endpoint, BMAEndpoint
 from ucoinpy.documents.block import Block
 from ..tools.exceptions import NoPeerAvailable
-from .network.node import Node
+from .net.node import Node
+from .net.network import Network
 import logging
 import inspect
 import hashlib
@@ -75,13 +76,12 @@ class Community(object):
     classdocs
     '''
 
-    def __init__(self, currency, peers):
+    def __init__(self, currency, network):
         '''
         A community is a group of nodes using the same currency.
         '''
         self.currency = currency
-        self.peers = [p for p in peers if p.currency == currency]
-        self._nodes = [Node.from_peer(p) for p in peers]
+        self._network = network
         self._cache = Cache(self)
 
         self._cache.refresh()
@@ -97,47 +97,11 @@ class Community(object):
 
     @classmethod
     def load(cls, json_data):
-        peers = []
-
         currency = json_data['currency']
 
-        for data in json_data['peers']:
-            for e in data['endpoints']:
-                if Endpoint.from_inline(e) is not None:
-                    endpoint = Endpoint.from_inline(e)
-                    try:
-                        peering = bma.network.Peering(endpoint.conn_handler())
-                        peer_data = peering.get()
-                        peer = Peer.from_signed_raw("{0}{1}\n".format(peer_data['raw'],
-                                                          peer_data['signature']))
-                        peers.append(peer)
-                        break
-                    except:
-                        pass
-
-        community = cls(currency, peers)
-        logging.debug("Creating community")
-        community.peers = community.peering()
-        logging.debug("{0} peers found".format(len(community.peers)))
-        logging.debug([peer.pubkey for peer in community.peers])
-        return community
-
-    @classmethod
-    def without_network(cls, json_data):
-        peers = []
+        network = Network.from_json(currency, json_data['peers'])
 
-        currency = json_data['currency']
-
-        for data in json_data['peers']:
-            endpoints = []
-            for e in data['endpoints']:
-                endpoints.append(Endpoint.from_inline(e))
-            peer = Peer(PROTOCOL_VERSION, currency, data['pubkey'],
-                        "0-DA39A3EE5E6B4B0D3255BFEF95601890AFD80709",
-                        endpoints, None)
-            peers.append(peer)
-
-        community = cls(currency, peers)
+        community = cls(currency, network)
         return community
 
     def load_cache(self, json_data):
@@ -206,48 +170,9 @@ class Community(object):
             if '404' in e:
                 return 0
 
-    def _peering_traversal(self, peer, found_peers, traversed_pubkeys):
-        logging.debug("Read {0} peering".format(peer.pubkey))
-        traversed_pubkeys.append(peer.pubkey)
-        if peer.currency == self.currency and \
-            peer.pubkey not in [p.pubkey for p in found_peers]:
-            found_peers.append(peer)
-        try:
-            e = next(e for e in peer.endpoints if type(e) is BMAEndpoint)
-            next_peers = bma.network.peering.Peers(e.conn_handler()).get()
-            for p in next_peers:
-                next_peer = Peer.from_signed_raw("{0}{1}\n".format(p['value']['raw'],
-                                                            p['value']['signature']))
-                logging.debug(traversed_pubkeys)
-                logging.debug("Traversing : next to read : {0} : {1}".format(next_peer.pubkey,
-                              (next_peer.pubkey not in traversed_pubkeys)))
-                if next_peer.pubkey not in traversed_pubkeys:
-                    self._peering_traversal(next_peer, found_peers, traversed_pubkeys)
-        except Timeout:
-            pass
-        except ConnectionError:
-            pass
-        except ValueError:
-            pass
-        except RequestException as e:
-            pass
-
-    def peering(self):
-        peers = []
-        traversed_pubkeys = []
-        for p in self.peers:
-            logging.debug(traversed_pubkeys)
-            logging.debug("Peering : next to read : {0} : {1}".format(p.pubkey,
-                          (p.pubkey not in traversed_pubkeys)))
-            if p.pubkey not in traversed_pubkeys:
-                self._peering_traversal(p, peers, traversed_pubkeys)
-
-        logging.debug("Peers found : {0}".format(peers))
-        return peers
-
     @property
     def nodes(self):
-        return self._nodes
+        return self._network.all_nodes
 
     def get_block(self, number=None):
         if number is None:
@@ -287,10 +212,10 @@ class Community(object):
         if cached:
             return self._cache.request(request, req_args, get_args)
         else:
-            for peer in self.peers.copy():
-                e = next(e for e in peer.endpoints if type(e) is BMAEndpoint)
+            nodes = self._network.online_nodes
+            for node in nodes:
                 try:
-                    req = request(e.conn_handler(), **req_args)
+                    req = request(node.endpoint.conn_handler(), **req_args)
                     data = req.get(**get_args)
 
                     if inspect.isgenerator(data):
@@ -306,40 +231,33 @@ class Community(object):
                     else:
                         raise
                 except RequestException:
-                    # Move the timeout peer to the end
-                    self.peers.remove(peer)
-                    self.peers.append(peer)
                     continue
 
-        raise NoPeerAvailable(self.currency, len(self.peers))
+        raise NoPeerAvailable(self.currency, len(nodes))
 
     def post(self, request, req_args={}, post_args={}):
-        for peer in self.peers:
-            e = next(e for e in peer.endpoints if type(e) is BMAEndpoint)
-            logging.debug("Trying to connect to : " + peer.pubkey)
-            req = request(e.conn_handler(), **req_args)
+        nodes = self._network.online_nodes
+        for node in nodes:
+            req = request(node.endpoint.conn_handler(), **req_args)
+            logging.debug("Trying to connect to : " + node.pubkey)
+            req = request(node.endpoint.conn_handler(), **req_args)
             try:
                 req.post(**post_args)
                 return
             except ValueError as e:
                 raise
             except RequestException:
-                # Move the timeout peer to the end
-                self.peers.remove(peer)
-                self.peers.append(peer)
                 continue
-            except:
-                raise
-        raise NoPeerAvailable(self.currency, len(self.peers))
+        raise NoPeerAvailable(self.currency, len(nodes))
 
     def broadcast(self, request, req_args={}, post_args={}):
         tries = 0
         ok = False
         value_error = None
-        for peer in self.peers:
-            e = next(e for e in peer.endpoints if type(e) is BMAEndpoint)
-            logging.debug("Trying to connect to : " + peer.pubkey)
-            req = request(e.conn_handler(), **req_args)
+        nodes = self._network.online_nodes
+        for node in nodes:
+            logging.debug("Trying to connect to : " + node.pubkey)
+            req = request(node.endpoint.conn_handler(), **req_args)
             try:
                 req.post(**post_args)
                 ok = True
@@ -348,32 +266,17 @@ class Community(object):
                 continue
             except RequestException:
                 tries = tries + 1
-                # Move the timeout peer to the end
-                self.peers.remove(peer)
-                self.peers.append(peer)
                 continue
-            except:
-                raise
 
         if not ok:
             raise value_error
 
         if tries == len(self.peers):
-            raise NoPeerAvailable(self.currency, len(self.peers))
-
-    def jsonify_peers_list(self):
-        data = []
-        for peer in self.peers:
-            endpoints_data = []
-            for e in peer.endpoints:
-                endpoints_data.append(e.inline())
-            data.append({'endpoints': endpoints_data,
-                         'pubkey': peer.pubkey})
-        return data
+            raise NoPeerAvailable(self.currency, len(nodes))
 
     def jsonify(self):
         data = {'currency': self.currency,
-                'peers': self.jsonify_peers_list()}
+                'peers': self.network.jsonify()}
         return data
 
     def get_parameters(self):

src/cutecoin/core/net/network.py

+'''
+Created on 24 févr. 2015
+
+@author: inso
+'''
+
+from ucoinpy.documents.peer import Peer, BMAEndpoint
+from ucoinpy.api import bma
+
+from .node import Node
+
+import logging
+import time
+
+from PyQt5.QtCore import QObject, pyqtSignal
+
+
+class Network(QObject):
+    '''
+    classdocs
+    '''
+    nodes_changed = pyqtSignal()
+
+    def __init__(self, currency, nodes):
+        '''
+        Constructor
+        '''
+        self.currency = currency
+        self._nodes = nodes
+        #TODO: Crawl nodes at startup
+
+    @classmethod
+    def from_json(cls, currency, json_data):
+        nodes = []
+        for data in json_data:
+            node = Node.from_json(currency, data)
+            nodes.append(node)
+        block_max = max([n.block for n in nodes])
+        for node in nodes:
+            node.check_sync(currency, block_max)
+        return cls(currency, nodes)
+
+    def jsonify(self):
+        data = []
+        for node in self.nodes:
+            data.append(node.jsonify())
+        return data
+
+    @property
+    def online_nodes(self):
+        return [n for n in self._nodes if n.state == Node.ONLINE]
+
+    @property
+    def all_nodes(self):
+        return self._nodes.copy()
+
+    def perpetual_crawling(self):
+        while self.must_crawl:
+            self.crawling(interval=10)
+
+    def crawling(self, interval=0):
+        nodes = []
+        traversed_pubkeys = []
+        for n in self._nodes:
+            logging.debug(traversed_pubkeys)
+            logging.debug("Peering : next to read : {0} : {1}".format(n.pubkey,
+                          (n.pubkey not in traversed_pubkeys)))
+            if n.pubkey not in traversed_pubkeys:
+                n.peering_traversal(self.currency, nodes,
+                                    traversed_pubkeys, interval)
+                time.sleep(interval)
+
+        block_max = max([n.block for n in nodes])
+        for node in [n for n in nodes if n.state == Node.ONLINE]:
+            node.check_sync(block_max)
+
+        #TODO: Offline nodes for too long have to be removed
+        #TODO: Corrupted nodes should maybe be removed faster ?
+
+        logging.debug("Nodes found : {0}".format(nodes))
+        return nodes

src/cutecoin/core/net/node.py

+'''
+Created on 21 févr. 2015
+
+@author: inso
+'''
+
+from ucoinpy.documents.peer import Peer, BMAEndpoint, Endpoint
+from ucoinpy.api import bma
+from requests.exceptions import RequestException
+from ...core.person import Person
+from ...tools.exceptions import PersonNotFoundError
+import logging
+import time
+
+
+class Node(object):
+    '''
+    classdocs
+    '''
+
+    ONLINE = 1
+    OFFLINE = 2
+    DESYNCED = 3
+    CORRUPTED = 4
+
+    def __init__(self, endpoints, pubkey, block, state):
+        '''
+        Constructor
+        '''
+        self._endpoints = endpoints
+        self._pubkey = pubkey
+        self._block = block
+        self._state = state
+
+    @classmethod
+    def from_peer(cls, currency, peer):
+        node = cls(peer.endpoints, "", 0, Node.ONLINE)
+        node.refresh_state(currency)
+        return node
+
+    @classmethod
+    def from_json(cls, currency, data):
+        endpoints = []
+        for endpoint_data in data['endpoints']:
+            endpoints.append(Endpoint.from_inline(endpoint_data))
+
+        node = cls(endpoints, "", 0, Node.ONLINE)
+        node.refresh_state(currency)
+        return node
+
+    @property
+    def pubkey(self):
+        return self._pubkey
+
+    @property
+    def endpoint(self):
+        return next((e for e in self._endpoints if type(e) is BMAEndpoint))
+
+    @property
+    def block(self):
+        return self._block
+
+    @property
+    def state(self):
+        return self._state
+
+    def check_sync(self, currency, block):
+        if self._block < block:
+            self._state = Node.DESYNCED
+        else:
+            self._state = Node.ONLINE
+
+    def refresh_state(self, currency):
+        try:
+            informations = bma.network.Peering(self.endpoint.conn_handler()).get()
+            block = bma.blockchain.Current(self.endpoint.conn_handler()).get()
+            block_number = block["number"]
+            node_pubkey = informations["pubkey"]
+            node_currency = informations["currency"]
+        except ValueError as e:
+            if '404' in e:
+                block_number = 0
+        except RequestException:
+            self._state = Node.OFFLINE
+
+        if node_currency != currency:
+            self.state = Node.CORRUPTED
+
+        self._block = block_number
+        self._pubkey = node_pubkey
+
+    def peering_traversal(self, currency, found_nodes, traversed_pubkeys, interval):
+        logging.debug("Read {0} peering".format(self.pubkey))
+        traversed_pubkeys.append(self.pubkey)
+        self.refresh_state(currency)
+        if self.pubkey not in [n.pubkey for n in found_nodes]:
+            found_nodes.append(self)
+
+        try:
+            next_peers = bma.network.peering.Peers(self.endpoint.conn_handler()).get()
+            for p in next_peers:
+                next_peer = Peer.from_signed_raw("{0}{1}\n".format(p['value']['raw'],
+                                                            p['value']['signature']))
+                logging.debug(traversed_pubkeys)
+                logging.debug("Traversing : next to read : {0} : {1}".format(next_peer.pubkey,
+                              (next_peer.pubkey not in traversed_pubkeys)))
+                next_node = Node.from_peer(next_peer)
+                if next_node.pubkey not in traversed_pubkeys:
+                    next_node.peering_traversal(currency, found_nodes, traversed_pubkeys)
+                    time.sleep(interval)
+        except RequestException as e:
+            self._state = Node.OFFLINE

src/cutecoin/core/network/node.py

-'''
-Created on 21 févr. 2015
-
-@author: inso
-'''
-
-from ucoinpy.documents.peer import Peer, BMAEndpoint
-from ucoinpy.api import bma
-from ...core.person import Person
-from ...tools.exceptions import PersonNotFoundError
-
-
-class Node(object):
-    '''
-    classdocs
-    '''
-
-    def __init__(self, endpoint, pubkey, block):
-        '''
-        Constructor
-        '''
-        self._endpoint = endpoint
-        self._pubkey = pubkey
-        self._block = block
-
-    @classmethod
-    def from_peer(cls, peer):
-        e = next((e for e in peer.endpoints if type(e) is BMAEndpoint))
-        informations = bma.network.Peering(e.conn_handler()).get()
-        try:
-            block = bma.blockchain.Current(e.conn_handler()).get()
-            block_number = block["number"]
-        except ValueError as e:
-            if '404' in e:
-                block_number = 0
-        node_pubkey = informations["pubkey"]
-
-        return cls(e, node_pubkey, block_number)
-
-    @property
-    def pubkey(self):
-        return self._pubkey
-
-    @property
-    def endpoint(self):
-        return self._endpoint
-
-    @property
-    def block(self):
-        return self._block

src/cutecoin/models/network.py

@@ -117,6 +117,7 @@ class NetworkTableModel(QAbstractTableModel):
         node = self.nodes[row]
         if role == Qt.DisplayRole:
             return self.data_node(node)[col]
+        #TODO: Display colors depending on node state
 
     def flags(self, index):
         return Qt.ItemIsSelectable | Qt.ItemIsEnabled