Use new IPv8 constructs for PopularityCommunity

src/tribler-core/tribler_core/modules/popularity/payload.py

@@ -1,54 +1,43 @@
-import struct
+from ipv8.messaging.lazy_payload import VariablePayload, vp_compile
+from ipv8.messaging.serialization import default_serializer
 
-from ipv8.messaging.payload import Payload
 
-TORRENT_INFO_FORMAT = '20sIIQ'  # Infohash, seeders, leechers and a timestamp
+@vp_compile
+class TorrentInfoFormat(VariablePayload):
+    format_list = ['20s', 'I', 'I', 'Q']
+    names = ['infohash', 'seeders', 'leechers', 'timestamp']
+    length = 36
 
+    def to_tuple(self):
+        return self.infohash, self.seeders, self.leechers, self.timestamp
+
+    @classmethod
+    def from_list_bytes(cls, serialized):
+        return default_serializer.unpack_to_serializables([cls] * (len(serialized)//cls.length), serialized)[:-1]
 
-class TorrentsHealthPayload(Payload):
 
+@vp_compile
+class TorrentsHealthPayload(VariablePayload):
+
+    msg_id = 1
     format_list = ['I', 'I', 'varlenI', 'raw']  # Number of random torrents, number of torrents checked by you
+    names = ['random_torrents_length', 'torrents_checked_length', 'random_torrents', 'torrents_checked']
+
+    def fix_pack_random_torrents(self, value):
+        return b''.join(default_serializer.ez_pack_serializables([TorrentInfoFormat(*sublist)]) for sublist in value)
 
-    def __init__(self, random_torrents, torrents_checked):
-        """
-        Initialize a TorrentsHealthPayload, containing information on the health of both random torrents and popular
-        torrents that have been checked by you.
-        :param random_torrents: List of tuple of (infohash, seeders, leechers, checked_timestamp)
-        :param torrents_checked: List of tuple of (infohash, seeders, leechers, checked_timestamp)
-        """
-        super(TorrentsHealthPayload, self).__init__()
-        self.random_torrents = random_torrents
-        self.torrents_checked = torrents_checked
-
-    def to_pack_list(self):
-        random_torrents_items = [item for sublist in self.random_torrents for item in sublist]
-        checked_torrents_items = [item for sublist in self.torrents_checked for item in sublist]
-        data = [('I', len(self.random_torrents)),
-                ('I', len(self.torrents_checked)),
-                ('varlenI', struct.pack("!" + TORRENT_INFO_FORMAT * len(self.random_torrents), *random_torrents_items)),
-                ('raw', struct.pack("!" + TORRENT_INFO_FORMAT * len(self.torrents_checked), *checked_torrents_items))]
-
-        return data
+    def fix_pack_torrents_checked(self, value):
+        return b''.join(default_serializer.ez_pack_serializables([TorrentInfoFormat(*sublist)]) for sublist in value)
+
+    @classmethod
+    def fix_unpack_random_torrents(cls, value):
+        return [payload.to_tuple() for payload in TorrentInfoFormat.from_list_bytes(value)]
+
+    @classmethod
+    def fix_unpack_torrents_checked(cls, value):
+        return [payload.to_tuple() for payload in TorrentInfoFormat.from_list_bytes(value)]
 
     @classmethod
-    def from_unpack_list(cls, *args):
-        num_random_torrents, num_checked_torrents, raw_random_torrents, raw_checked_torrents = args
-
-        random_torrents_list = struct.unpack("!" + TORRENT_INFO_FORMAT * num_random_torrents, raw_random_torrents)
-        checked_torrents_list = struct.unpack("!" + TORRENT_INFO_FORMAT * num_checked_torrents, raw_checked_torrents)
-
-        random_torrents = []
-        checked_torrents = []
-        for ind in range(num_random_torrents):
-            random_torrents.append((random_torrents_list[ind * 4],
-                                    random_torrents_list[ind * 4 + 1],
-                                    random_torrents_list[ind * 4 + 2],
-                                    random_torrents_list[ind * 4 + 3]))
-
-        for ind in range(num_checked_torrents):
-            checked_torrents.append((checked_torrents_list[ind * 4],
-                                     checked_torrents_list[ind * 4 + 1],
-                                     checked_torrents_list[ind * 4 + 2],
-                                     checked_torrents_list[ind * 4 + 3]))
-
-        return TorrentsHealthPayload(random_torrents, checked_torrents)
+    def create(cls, random_torrents_checked, popular_torrents_checked):
+        return cls(len(random_torrents_checked), len(popular_torrents_checked),
+                   random_torrents_checked, popular_torrents_checked)

src/tribler-core/tribler_core/modules/popularity/popularity_community.py

@@ -4,7 +4,6 @@
 
 from ipv8.community import Community
 from ipv8.lazy_community import lazy_wrapper
-from ipv8.messaging.payload_headers import BinMemberAuthenticationPayload
 from ipv8.peer import Peer
 
 from pony.orm import db_session
@@ -14,8 +13,6 @@
 
 PUBLISH_INTERVAL = 5
 
-MSG_TORRENTS_HEALTH = 1
-
 
 class PopularityCommunity(Community):
     """
@@ -32,9 +29,7 @@ def __init__(self, *args, **kwargs):
 
         super(PopularityCommunity, self).__init__(*args, **kwargs)
 
-        self.decode_map.update({
-            chr(MSG_TORRENTS_HEALTH): self.on_torrents_health
-        })
+        self.add_message_handler(TorrentsHealthPayload, self.on_torrents_health)
 
         self.logger.info('Popularity Community initialized (peer mid %s)', hexlify(self.my_peer.mid))
         self.register_task("publish", self.gossip_torrents_health, interval=PUBLISH_INTERVAL)
@@ -54,17 +49,12 @@ def gossip_torrents_health(self):
 
         random_peer = random.choice(self.get_peers())
 
-        auth = BinMemberAuthenticationPayload(self.my_peer.public_key.key_to_bin()).to_pack_list()
-        payload = TorrentsHealthPayload(random_torrents_checked, popular_torrents_checked).to_pack_list()
-
-        packet = self._ez_pack(self._prefix, MSG_TORRENTS_HEALTH, [auth, payload])
-        self.endpoint.send(random_peer.address, packet)
+        self.ez_send(random_peer, TorrentsHealthPayload.create(random_torrents_checked, popular_torrents_checked))
 
     @lazy_wrapper(TorrentsHealthPayload)
     async def on_torrents_health(self, _, payload):
         self.logger.info("Received torrent health information for %d random torrents and %d checked torrents",
                          len(payload.random_torrents), len(payload.torrents_checked))
-
         all_torrents = payload.random_torrents + payload.torrents_checked
 
         def _put_health_entries_in_db():