diff --git a/squeaknode/sync/peer_sync_controller.py b/squeaknode/sync/peer_sync_controller.py index dad3e456..3d3d64ec 100644 --- a/squeaknode/sync/peer_sync_controller.py +++ b/squeaknode/sync/peer_sync_controller.py @@ -1,9 +1,6 @@ import logging from contextlib import contextmanager -from squeak.core import CSqueak - -from squeaknode.core.util import get_hash from squeaknode.network.peer_client import PeerClient from squeaknode.sync.util import parse_buy_offer @@ -36,20 +33,24 @@ class PeerSyncController: min_block, max_block, ): - # Get list of followed addresses. + # Get list of followed addresses followed_addresses = self.squeak_controller.get_followed_addresses() # Get remote hashes - lookup_result = self._get_remote_hashes( - followed_addresses, min_block, max_block) + lookup_result = self.peer_client.lookup_squeaks( + followed_addresses, + min_block, + max_block, + ) remote_hashes = lookup_result.hashes - # Get local hashes of downloaded squeaks - local_hashes = self._get_local_hashes( - followed_addresses, min_block, max_block) + # Get local hashes of saved squeaks + local_hashes = self.squeak_controller.lookup_squeaks_include_locked( + followed_addresses, + min_block, + max_block, + ) # Get hashes to download hashes_to_download = set(remote_hashes) - set(local_hashes) - # Download squeaks for the hashes - # TODO: catch exception downloading individual squeak. # TODO: check if hash belongs to correct range after downloading. for hash in hashes_to_download: # if self.peer_connection.stopped(): @@ -57,8 +58,12 @@ class PeerSyncController: self._download_squeak(hash) # Get local hashes of locked squeaks that don't have an offer from this peer. - locked_hashes = self._get_locked_hashes( - followed_addresses, min_block, max_block) + locked_hashes = self.squeak_controller.lookup_squeaks_needing_offer( + followed_addresses, + min_block, + max_block, + self.peer.peer_id, + ) # Get hashes to get offer hashes_to_get_offer = set(remote_hashes) & set(locked_hashes) # Download offers for the hashes @@ -75,91 +80,50 @@ class PeerSyncController: ): # Get list of sharing addresses. sharing_addresses = self.squeak_controller.get_sharing_addresses() - # Get remote hashes - lookup_result = self._get_remote_hashes( - sharing_addresses, min_block, max_block) + lookup_result = self.peer_client.lookup_squeaks( + sharing_addresses, + min_block, + max_block, + ) remote_hashes = lookup_result.hashes allowed_addresses = lookup_result.allowed_addresses peer_latest_block = lookup_result.latest_block_height - max_block = min(max_block, peer_latest_block) if max_block < min_block: return - # Get local hashes addresses_to_search = set(allowed_addresses) & set(sharing_addresses) - local_hashes = self._get_local_unlocked_hashes( - addresses_to_search, min_block, max_block) - + local_hashes = self.squeak_controller.lookup_squeaks( + addresses_to_search, + min_block, + max_block, + ) # Get hashes to upload hashes_to_upload = set(local_hashes) - set(remote_hashes) - # Upload squeaks for the hashes # TODO: catch exception uploading individual squeak for hash in hashes_to_upload: # if self.peer_connection.stopped(): # return - squeak = self._get_local_squeak(hash) - self._upload_squeak(squeak) + self._upload_squeak(hash) def download_single_squeak(self, squeak_hash: bytes): - # Download squeak if not already present. - saved_squeak = self._get_local_squeak(squeak_hash) + """Downloads a single squeak and the corresponding offer. """ + saved_squeak = self.squeak_controller.get_squeak(squeak_hash) if not saved_squeak: self._download_squeak(squeak_hash) - - # Download offer from peer if not already present. saved_offer = self._get_saved_offer(squeak_hash) if not saved_offer: self._download_offer(squeak_hash) def upload_single_squeak(self, squeak_hash: bytes): - # Upload the squeak if it exists locally. - local_squeak = self._get_local_squeak(squeak_hash) + """Uploads a single squeak. """ + local_squeak = self.squeak_controller.get_squeak(squeak_hash) if local_squeak and local_squeak.HasDecryptionKey(): - self._upload_squeak(local_squeak) - - def get_offer(self, squeak_hash: bytes): - # Get the squeak from the squeak hash - squeak = self._get_local_squeak(squeak_hash) - - # Download the buy offer - offer_msg = self._download_offer_msg(squeak_hash) - buy_offer = parse_buy_offer(offer_msg) - - decoded_offer = self.squeak_controller.get_offer( - squeak, buy_offer, self.peer) - - # Save the offer - self.squeak_controller.save_offer(decoded_offer) - - def _get_local_hashes(self, addresses, min_block, max_block): - return self.squeak_controller.lookup_squeaks_include_locked( - addresses, - min_block, - max_block, - ) - - def _get_local_unlocked_hashes(self, addresses, min_block, max_block): - return self.squeak_controller.lookup_squeaks(addresses, min_block, max_block) - - def _get_locked_hashes(self, addresses, min_block, max_block): - return self.squeak_controller.lookup_squeaks_needing_offer( - addresses, - min_block, - max_block, - self.peer.peer_id, - ) - - def _get_remote_hashes(self, addresses, min_block, max_block): - return self.peer_client.lookup_squeaks(addresses, min_block, max_block) - - def _save_squeak(self, squeak: CSqueak): - self.squeak_controller.save_downloaded_squeak(squeak) + self._upload_squeak(squeak_hash) def _get_saved_offer(self, squeak_hash: bytes): - # offers = self.squeak_db.get_offers_with_peer(squeak_hash) offers = self.squeak_controller.get_buy_offers_with_peer(squeak_hash) for offer_with_peer in offers: if offer_with_peer.offer.peer_id == self.peer.peer_id: @@ -167,25 +131,25 @@ class PeerSyncController: def _download_squeak(self, squeak_hash: bytes): squeak = self.peer_client.get_squeak(squeak_hash) - self._save_squeak(squeak) + self.squeak_controller.save_downloaded_squeak(squeak) logger.info("Downloaded squeak {} from peer {}".format( squeak_hash.hex(), self.peer )) def _download_offer(self, squeak_hash: bytes): - self.get_offer(squeak_hash) + squeak = self.squeak_controller.get_squeak(squeak_hash) + offer_msg = self.peer_client.buy_squeak(squeak_hash) + buy_offer = parse_buy_offer(offer_msg) + decoded_offer = self.squeak_controller.get_offer( + squeak, buy_offer, self.peer) + self.squeak_controller.save_offer(decoded_offer) logger.info("Downloaded offer for squeak {} from peer {}".format( squeak_hash.hex(), self.peer )) - def _get_local_squeak(self, squeak_hash: bytes): - return self.squeak_controller.get_squeak(squeak_hash) - - def _upload_squeak(self, squeak: CSqueak): + def _upload_squeak(self, squeak_hash: bytes): + squeak = self.squeak_controller.get_squeak(squeak_hash) self.peer_client.post_squeak(squeak) logger.info("Uploaded squeak {} to peer {}".format( - get_hash(squeak).hex(), self.peer + squeak_hash.hex(), self.peer )) - - def _download_offer_msg(self, squeak_hash: bytes): - return self.peer_client.buy_squeak(squeak_hash)