Make update subscriptions async (#1723)

* Update subscriptions asynchronously when follows list changes

* Change log level of lines in update follows worker

* Remove old commented lines for update follows subscription

* Use update subscription event class to represent new subscription change required

squeaknode/core/update_subscriptions_event.py

@@ -0,0 +1,25 @@
+# MIT License
+#
+# Copyright (c) 2020 Jonathan Zernik
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+
+class UpdateSubscriptionsEvent():
+    """Represents an event that requires an update of subscriptions."""

squeaknode/node/squeak_controller.py

@@ -59,6 +59,7 @@ from squeaknode.core.squeak_entry import SqueakEntry
 from squeaknode.core.squeak_peer import SqueakPeer
 from squeaknode.core.squeak_profile import SqueakProfile
 from squeaknode.core.squeaks import get_hash
+from squeaknode.core.update_subscriptions_event import UpdateSubscriptionsEvent
 from squeaknode.node.listener_subscription_client import EventListener
 from squeaknode.node.received_payments_subscription_client import ReceivedPaymentsSubscriptionClient
 from squeaknode.node.temporary_interest_manager import TemporaryInterest
@@ -85,6 +86,7 @@ class SqueakController:
         self.new_squeak_listener = EventListener()
         self.new_received_offer_listener = EventListener()
         self.new_secret_key_listener = EventListener()
+        self.new_follow_listener = EventListener()
         self.temporary_interest_manager = TemporaryInterestManager()
         self.config = config
 
@@ -253,7 +255,7 @@ class SqueakController:
             profile_name,
         )
         profile_id = self.squeak_db.insert_profile(squeak_profile)
-        self.update_subscriptions()
+        self.create_update_subscriptions_event()
         return profile_id
 
     def import_signing_profile(self, profile_name: str, private_key: str) -> int:
@@ -262,7 +264,7 @@ class SqueakController:
             private_key,
         )
         profile_id = self.squeak_db.insert_profile(squeak_profile)
-        self.update_subscriptions()
+        self.create_update_subscriptions_event()
         return profile_id
 
     def create_contact_profile(self, profile_name: str, squeak_address: str) -> int:
@@ -271,7 +273,7 @@ class SqueakController:
             squeak_address,
         )
         profile_id = self.squeak_db.insert_profile(squeak_profile)
-        self.update_subscriptions()
+        self.create_update_subscriptions_event()
         return profile_id
 
     def get_profiles(self) -> List[SqueakProfile]:
@@ -294,7 +296,7 @@ class SqueakController:
 
     def set_squeak_profile_following(self, profile_id: int, following: bool) -> None:
         self.squeak_db.set_profile_following(profile_id, following)
-        self.update_subscriptions()
+        self.create_update_subscriptions_event()
 
     def set_squeak_profile_use_custom_price(self, profile_id: int, use_custom_price: bool) -> None:
         self.squeak_db.set_profile_use_custom_price(
@@ -309,7 +311,7 @@ class SqueakController:
 
     def delete_squeak_profile(self, profile_id: int) -> None:
         self.squeak_db.delete_profile(profile_id)
-        self.update_subscriptions()
+        self.create_update_subscriptions_event()
 
     def set_squeak_profile_image(self, profile_id: int, profile_image: bytes) -> None:
         self.squeak_db.set_profile_image(profile_id, profile_image)
@@ -798,10 +800,16 @@ class SqueakController:
     def subscribe_new_secret_keys(self, stopped: threading.Event):
         yield from self.new_secret_key_listener.yield_items(stopped)
 
+    def subscribe_follows(self, stopped: threading.Event):
+        yield from self.new_follow_listener.yield_items(stopped)
+
     def update_subscriptions(self):
         locator = self.get_interested_locator()
         self.network_manager.update_local_subscriptions(locator)
 
+    def create_update_subscriptions_event(self):
+        self.new_follow_listener.handle_new_item(UpdateSubscriptionsEvent())
+
     def subscribe_received_offers_for_squeak(self, squeak_hash: bytes, stopped: threading.Event):
         for received_offer in self.new_received_offer_listener.yield_items(stopped):
             if received_offer.squeak_hash == squeak_hash:

squeaknode/node/squeak_node.py

@@ -42,6 +42,7 @@ from squeaknode.node.process_received_payments_worker import ProcessReceivedPaym
 from squeaknode.node.squeak_controller import SqueakController
 from squeaknode.node.squeak_deletion_worker import SqueakDeletionWorker
 from squeaknode.node.squeak_offer_expiry_worker import SqueakOfferExpiryWorker
+from squeaknode.node.update_follows_worker import UpdateFollowsWorker
 from squeaknode.node.update_subscribed_secret_key_worker import UpdateSubscribedSecretKeysWorker
 from squeaknode.node.update_subscribed_squeak_worker import UpdateSubscribedSqueaksWorker
 
@@ -72,6 +73,7 @@ class SqueakNode:
         self.initialize_offer_expiry_worker()
         self.initialize_new_squeak_worker()
         self.initialize_new_secret_key_worker()
+        self.initialize_new_follow_worker()
         self.initialize_peer_subscription_update_worker()
 
     def start_running(self):
@@ -88,6 +90,7 @@ class SqueakNode:
         self.offer_expiry_worker.start()
         self.new_squeak_worker.start_running()
         self.new_secret_key_worker.start_running()
+        self.new_follow_worker.start_running()
         self.new_bitcoin_block_worker.start_running()
 
     def stop_running(self):
@@ -223,6 +226,11 @@ class SqueakNode:
             self.squeak_controller,
         )
 
+    def initialize_new_follow_worker(self):
+        self.new_follow_worker = UpdateFollowsWorker(
+            self.squeak_controller,
+        )
+
     def initialize_peer_subscription_update_worker(self):
         self.new_bitcoin_block_worker = PeerSubscriptionUpdateWorker(
             self.squeak_controller,

squeaknode/node/update_follows_worker.py

@@ -0,0 +1,53 @@
+# MIT License
+#
+# Copyright (c) 2020 Jonathan Zernik
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+import logging
+import threading
+
+from squeaknode.node.squeak_controller import SqueakController
+
+
+logger = logging.getLogger(__name__)
+
+
+class UpdateFollowsWorker:
+
+    def __init__(self, squeak_controller: SqueakController):
+        self.squeak_controller = squeak_controller
+        self.stopped = threading.Event()
+
+    def start_running(self):
+        threading.Thread(
+            target=self.handle_new_follow,
+            name="new_follows_thread",
+            daemon=True,
+        ).start()
+
+    def stop_running(self):
+        self.stopped.set()
+
+    def handle_new_follow(self):
+        logger.debug("Starting UpdateFollowsWorker...")
+        for _ in self.squeak_controller.subscribe_follows(
+                self.stopped,
+        ):
+            logger.debug("Handling update subscriptions event")
+            self.squeak_controller.update_subscriptions()