mirror of
https://github.com/yzernik/squeaknode.git
synced 2026-08-15 12:50:47 +02:00
Create peer sync task class (#350)
* Create peer sync task class * Convery peer upload to subclass of peer sync tsk * Use network task class for download task * Use network task class for single squeak download * comment unused network sync methods * Remove old network controller methods * Move all peer task methods into peer task class * Move get offer method into peer sync class * Delete get offer class * Upload timeline squeaks using network task subclass * Fix upload squeaks to only upload unlocked squeaks * Remove unused synctask class
This commit is contained in:
Jonathan Zernik
GitHub
parent
d42f2362f3
commit
e66dbdeebd
8 changed files with 663 additions and 530 deletions
|
|
@ -370,7 +370,7 @@ class SqueakAdminServerHandler(object):
|
|||
)
|
||||
|
||||
def handle_sync_squeaks(self, request):
|
||||
logger.info("Handle get sync squeaks")
|
||||
logger.info("Handle sync squeaks")
|
||||
self.squeak_node.sync_squeaks()
|
||||
return squeak_admin_pb2.SyncSqueaksReply()
|
||||
|
||||
|
|
|
|||
129
squeaknode/node/network_task.py
Normal file
129
squeaknode/node/network_task.py
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
import logging
|
||||
import threading
|
||||
|
||||
from squeaknode.network.peer_client import PeerClient
|
||||
from squeaknode.node.peer_task import PeerSyncTask
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
LOOKUP_BLOCK_INTERVAL = 1008 # 1 week
|
||||
|
||||
|
||||
class NetworkSyncTask:
|
||||
def __init__(
|
||||
self,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lightning_client,
|
||||
):
|
||||
self.squeak_store = squeak_store
|
||||
self.postgres_db = postgres_db
|
||||
self.lightning_client = lightning_client
|
||||
|
||||
def sync(self, peers):
|
||||
logger.info("Network sync for class {}".format(
|
||||
self.__class__,
|
||||
))
|
||||
for peer in peers:
|
||||
sync_peer_thread = threading.Thread(
|
||||
target=self.sync_peer,
|
||||
args=(peer,),
|
||||
)
|
||||
sync_peer_thread.start()
|
||||
|
||||
def sync_peer(self, peer):
|
||||
# peer_upload = PeerUpload(
|
||||
# peer,
|
||||
# self.squeak_store,
|
||||
# self.postgres_db,
|
||||
# self.lightning_client,
|
||||
# )
|
||||
# try:
|
||||
# logger.debug("Trying to upload to peer: {}".format(peer))
|
||||
# with self.UploadingContextManager(
|
||||
# peer, peer_upload, self.squeak_sync_status
|
||||
# ) as uploading_manager:
|
||||
# peer_upload.upload(block_height)
|
||||
# except Exception as e:
|
||||
# logger.error("Upload from peer failed.", exc_info=True)
|
||||
pass
|
||||
|
||||
|
||||
class DownloadTimelineNetworkSyncTask(NetworkSyncTask):
|
||||
def __init__(
|
||||
self,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lightning_client,
|
||||
block_height,
|
||||
):
|
||||
super().__init__(squeak_store, postgres_db, lightning_client)
|
||||
self.block_height = block_height
|
||||
|
||||
def sync_peer(self, peer):
|
||||
if not peer.downloading:
|
||||
return
|
||||
peer_sync_task = PeerSyncTask(
|
||||
peer,
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
)
|
||||
try:
|
||||
logger.info("Trying to download with block height: {}".format(self.block_height))
|
||||
peer_sync_task.download(self.block_height)
|
||||
except Exception as e:
|
||||
logger.error("Download from peer failed.", exc_info=True)
|
||||
|
||||
class UploadTimelineNetworkSyncTask(NetworkSyncTask):
|
||||
def __init__(
|
||||
self,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lightning_client,
|
||||
block_height,
|
||||
):
|
||||
super().__init__(squeak_store, postgres_db, lightning_client)
|
||||
self.block_height = block_height
|
||||
|
||||
def sync_peer(self, peer):
|
||||
if not peer.uploading:
|
||||
return
|
||||
peer_sync_task = PeerSyncTask(
|
||||
peer,
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
)
|
||||
try:
|
||||
logger.info("Trying to upload with block height: {}".format(self.block_height))
|
||||
peer_sync_task.upload(self.block_height)
|
||||
except Exception as e:
|
||||
logger.error("Upload from peer failed.", exc_info=True)
|
||||
|
||||
class SingleSqueakNetworkSyncTask(NetworkSyncTask):
|
||||
def __init__(
|
||||
self,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lightning_client,
|
||||
squeak_hash,
|
||||
):
|
||||
super().__init__(squeak_store, postgres_db, lightning_client)
|
||||
self.squeak_hash = squeak_hash
|
||||
|
||||
def sync_peer(self, peer):
|
||||
if not peer.downloading:
|
||||
return
|
||||
peer_sync_task = PeerSyncTask(
|
||||
peer,
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
)
|
||||
try:
|
||||
logger.info("Trying to download single squeak {}".format(self.squeak_hash))
|
||||
peer_sync_task.download_single_squeak(self.squeak_hash)
|
||||
except Exception as e:
|
||||
logger.error("Download single squeak from peer failed.", exc_info=True)
|
||||
|
|
@ -1,154 +0,0 @@
|
|||
import logging
|
||||
import threading
|
||||
|
||||
from squeaknode.network.peer_client import PeerClient
|
||||
from squeaknode.node.peer_get_offer import PeerGetOffer
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
LOOKUP_BLOCK_INTERVAL = 1008 # 1 week
|
||||
|
||||
|
||||
class PeerDownload:
|
||||
def __init__(
|
||||
self,
|
||||
peer,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lightning_client,
|
||||
):
|
||||
self.peer = peer
|
||||
self.squeak_store = squeak_store
|
||||
self.postgres_db = postgres_db
|
||||
self.lightning_client = lightning_client
|
||||
|
||||
self.peer_client = PeerClient(
|
||||
self.peer.host,
|
||||
self.peer.port,
|
||||
)
|
||||
|
||||
self._stop_event = threading.Event()
|
||||
|
||||
def download(
|
||||
self,
|
||||
block_height,
|
||||
lookup_block_interval=LOOKUP_BLOCK_INTERVAL,
|
||||
):
|
||||
# Get list of followed addresses.
|
||||
addresses = self._get_followed_addresses()
|
||||
logger.debug("Followed addresses: {}".format(addresses))
|
||||
min_block = block_height - lookup_block_interval
|
||||
max_block = block_height
|
||||
|
||||
# Get remote hashes
|
||||
remote_hashes = self._get_remote_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got remote hashes: {}".format(len(remote_hashes)))
|
||||
for hash in remote_hashes:
|
||||
logger.debug("remote hash: {}".format(hash.hex()))
|
||||
|
||||
# Get local hashes of downloaded squeaks
|
||||
local_hashes = self._get_local_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got local hashes: {}".format(len(local_hashes)))
|
||||
for hash in local_hashes:
|
||||
logger.debug("local hash: {}".format(hash.hex()))
|
||||
|
||||
# Get hashes to download
|
||||
hashes_to_download = set(remote_hashes) - set(local_hashes)
|
||||
logger.debug("Hashes to download: {}".format(len(hashes_to_download)))
|
||||
for hash in hashes_to_download:
|
||||
logger.debug("hash to download: {}".format(hash.hex()))
|
||||
|
||||
# Download squeaks for the hashes
|
||||
# TODO: catch exception downloading individual squeak
|
||||
for hash in hashes_to_download:
|
||||
if self.stopped():
|
||||
return
|
||||
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(addresses, min_block, max_block)
|
||||
logger.debug("Got locked hashes: {}".format(len(locked_hashes)))
|
||||
for hash in locked_hashes:
|
||||
logger.debug("locked hash: {}".format(hash.hex()))
|
||||
|
||||
# Get hashes to get offer
|
||||
hashes_to_get_offer = set(remote_hashes) & set(locked_hashes)
|
||||
logger.debug("Hashes to get offer: {}".format(len(hashes_to_get_offer)))
|
||||
for hash in hashes_to_get_offer:
|
||||
logger.debug("hash to get offer: {}".format(hash.hex()))
|
||||
|
||||
# Download offers for the hashes
|
||||
# TODO: catch exception downloading individual squeak
|
||||
for hash in hashes_to_get_offer:
|
||||
if self.stopped():
|
||||
return
|
||||
self._download_offer(hash)
|
||||
|
||||
def download_single_squeak(self, squeak_hash):
|
||||
# Download squeak if not already present.
|
||||
saved_squeak = self._get_saved_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 stop(self):
|
||||
self._stop_event.set()
|
||||
|
||||
def stopped(self):
|
||||
return self._stop_event.is_set()
|
||||
|
||||
def _get_local_hashes(self, addresses, min_block, max_block):
|
||||
return self.squeak_store.lookup_squeaks_include_locked(
|
||||
addresses,
|
||||
min_block,
|
||||
max_block,
|
||||
)
|
||||
|
||||
def _get_locked_hashes(self, addresses, min_block, max_block):
|
||||
return self.squeak_store.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):
|
||||
self.squeak_store.save_downloaded_squeak(squeak)
|
||||
|
||||
def _get_saved_squeak(self, squeak_hash):
|
||||
return self.squeak_store.get_squeak(squeak_hash)
|
||||
return self.postgres_db.get_offers_with_peer(squeak_hash_str)
|
||||
|
||||
def _get_saved_offer(self, squeak_hash):
|
||||
offers = self.postgres_db.get_offers_with_peer(squeak_hash)
|
||||
for offer in offers:
|
||||
if offer.peer_id == peer_id:
|
||||
return offer
|
||||
|
||||
def _download_squeak(self, squeak_hash):
|
||||
logger.info("Downloading squeak: {} from peer: {}".format(squeak_hash.hex(), self.peer.peer_id))
|
||||
squeak = self.peer_client.get_squeak(squeak_hash)
|
||||
self._save_squeak(squeak)
|
||||
|
||||
def _get_followed_addresses(self):
|
||||
followed_profiles = self.postgres_db.get_following_profiles()
|
||||
return [profile.address for profile in followed_profiles]
|
||||
|
||||
def _download_offer(self, squeak_hash):
|
||||
logger.info("Downloading offer for hash: {}".format(squeak_hash.hex()))
|
||||
peer_get_offer = PeerGetOffer(
|
||||
self.peer,
|
||||
squeak_hash,
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
)
|
||||
peer_get_offer.get_offer()
|
||||
|
|
@ -1,147 +0,0 @@
|
|||
import logging
|
||||
import threading
|
||||
|
||||
from squeak.core.encryption import generate_data_key
|
||||
|
||||
from squeaknode.core.offer import Offer
|
||||
from squeaknode.network.peer_client import PeerClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PeerGetOffer:
|
||||
def __init__(self, peer, squeak_hash, squeak_store, postgres_db, lightning_client):
|
||||
self.peer = peer
|
||||
self.squeak_hash = squeak_hash
|
||||
self.squeak_store = squeak_store
|
||||
self.postgres_db = postgres_db
|
||||
self.lightning_client = lightning_client
|
||||
|
||||
self.peer_client = PeerClient(
|
||||
self.peer.host,
|
||||
self.peer.port,
|
||||
)
|
||||
|
||||
self._stop_event = threading.Event()
|
||||
|
||||
def get_offer(self):
|
||||
logger.info("Getting offer for squeak hash: {}".format(self.squeak_hash.hex()))
|
||||
|
||||
# Get the squeak from the squeak hash
|
||||
squeak = self._get_squeak()
|
||||
|
||||
# Get the encryption key
|
||||
encryption_key = squeak.GetEncryptionKey()
|
||||
|
||||
# Create a new challenge
|
||||
challenge_proof = self._generate_challenge_proof()
|
||||
challenge = self._get_challenge(challenge_proof, encryption_key)
|
||||
|
||||
# Download the buy offer
|
||||
offer = self._download_buy_offer(challenge)
|
||||
|
||||
# Check the proof
|
||||
proof = offer.proof
|
||||
logger.info("Proof: {}".format(proof.hex()))
|
||||
logger.info("Expected proof: {}".format(challenge_proof.hex()))
|
||||
if proof != challenge_proof:
|
||||
raise Exception(
|
||||
"Invalid offer proof: {}, expected: {}".format(
|
||||
proof.hex(),
|
||||
challenge_proof.hex(),
|
||||
)
|
||||
)
|
||||
|
||||
# Get the decoded offer from the payment request string
|
||||
decoded_offer = self._get_decoded_offer(offer)
|
||||
|
||||
# Save the offer
|
||||
self._save_offer(decoded_offer)
|
||||
|
||||
def stop(self):
|
||||
self._stop_event.set()
|
||||
|
||||
def stopped(self):
|
||||
return self._stop_event.is_set()
|
||||
|
||||
def _get_squeak(self):
|
||||
squeak_entry = self.squeak_store.get_squeak(self.squeak_hash)
|
||||
return squeak_entry.squeak
|
||||
|
||||
def _generate_challenge_proof(self):
|
||||
return generate_data_key()
|
||||
|
||||
def _get_challenge(self, challenge_proof, encryption_key):
|
||||
return encryption_key.encrypt(challenge_proof)
|
||||
|
||||
def _download_buy_offer(self, challenge):
|
||||
logger.info(
|
||||
"Downloading buy offer for squeak hash: {}".format(self.squeak_hash.hex())
|
||||
)
|
||||
offer_msg = self.peer_client.buy_squeak(self.squeak_hash, challenge)
|
||||
offer = self._offer_from_msg(offer_msg)
|
||||
return offer
|
||||
|
||||
def _save_offer(self, offer):
|
||||
logger.info("Saving offer: {}".format(offer))
|
||||
self.postgres_db.insert_offer(offer)
|
||||
|
||||
def _offer_from_msg(self, offer_msg):
|
||||
if not offer_msg:
|
||||
return None
|
||||
return Offer(
|
||||
offer_id=None,
|
||||
squeak_hash=offer_msg.squeak_hash,
|
||||
key_cipher=offer_msg.key_cipher,
|
||||
iv=offer_msg.iv,
|
||||
price_msat=None,
|
||||
payment_hash=offer_msg.preimage_hash,
|
||||
invoice_timestamp=None,
|
||||
invoice_expiry=None,
|
||||
payment_request=offer_msg.payment_request,
|
||||
destination=None,
|
||||
node_host=offer_msg.host,
|
||||
node_port=offer_msg.port,
|
||||
proof=offer_msg.proof,
|
||||
peer_id=self.peer.peer_id,
|
||||
)
|
||||
|
||||
def _decode_payment_request(self, payment_request):
|
||||
return self.lightning_client.decode_pay_req(payment_request)
|
||||
|
||||
def _get_decoded_offer(self, offer):
|
||||
pay_req = self._decode_payment_request(offer.payment_request)
|
||||
logger.info("Decoded payment request: {}".format(pay_req))
|
||||
|
||||
price_msat = pay_req.num_msat
|
||||
destination = pay_req.destination
|
||||
invoice_timestamp = pay_req.timestamp
|
||||
invoice_expiry = pay_req.expiry
|
||||
node_host = offer.node_host or self.peer.host
|
||||
node_port = offer.node_port
|
||||
|
||||
logger.info("price_msat: {}".format(price_msat))
|
||||
logger.info("destination: {}".format(destination))
|
||||
logger.info("invoice_timestamp: {}".format(invoice_timestamp))
|
||||
logger.info("invoice_expiry: {}".format(invoice_expiry))
|
||||
logger.info("node_host: {}".format(node_host))
|
||||
logger.info("node_port: {}".format(node_port))
|
||||
|
||||
decoded_offer = Offer(
|
||||
offer_id=offer.offer_id,
|
||||
squeak_hash=offer.squeak_hash,
|
||||
key_cipher=offer.key_cipher,
|
||||
iv=offer.iv,
|
||||
price_msat=price_msat,
|
||||
payment_hash=offer.payment_hash,
|
||||
invoice_timestamp=invoice_timestamp,
|
||||
invoice_expiry=invoice_expiry,
|
||||
payment_request=offer.payment_request,
|
||||
destination=destination,
|
||||
node_host=node_host,
|
||||
node_port=node_port,
|
||||
proof=offer.proof,
|
||||
peer_id=offer.peer_id,
|
||||
)
|
||||
|
||||
return decoded_offer
|
||||
313
squeaknode/node/peer_task.py
Normal file
313
squeaknode/node/peer_task.py
Normal file
|
|
@ -0,0 +1,313 @@
|
|||
import logging
|
||||
import threading
|
||||
|
||||
|
||||
from squeak.core.encryption import generate_data_key
|
||||
|
||||
from squeaknode.core.offer import Offer
|
||||
from squeaknode.network.peer_client import PeerClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
LOOKUP_BLOCK_INTERVAL = 1008 # 1 week
|
||||
|
||||
|
||||
class PeerSyncTask:
|
||||
def __init__(
|
||||
self,
|
||||
peer,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lightning_client,
|
||||
):
|
||||
self.peer = peer
|
||||
self.squeak_store = squeak_store
|
||||
self.postgres_db = postgres_db
|
||||
self.lightning_client = lightning_client
|
||||
|
||||
self.peer_client = PeerClient(
|
||||
self.peer.host,
|
||||
self.peer.port,
|
||||
)
|
||||
|
||||
self._stop_event = threading.Event()
|
||||
|
||||
def download(
|
||||
self,
|
||||
block_height,
|
||||
lookup_block_interval=LOOKUP_BLOCK_INTERVAL,
|
||||
):
|
||||
# Get list of followed addresses.
|
||||
addresses = self._get_followed_addresses()
|
||||
logger.debug("Followed addresses: {}".format(addresses))
|
||||
min_block = block_height - lookup_block_interval
|
||||
max_block = block_height
|
||||
|
||||
# Get remote hashes
|
||||
remote_hashes = self._get_remote_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got remote hashes: {}".format(len(remote_hashes)))
|
||||
for hash in remote_hashes:
|
||||
logger.debug("remote hash: {}".format(hash.hex()))
|
||||
|
||||
# Get local hashes of downloaded squeaks
|
||||
local_hashes = self._get_local_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got local hashes: {}".format(len(local_hashes)))
|
||||
for hash in local_hashes:
|
||||
logger.debug("local hash: {}".format(hash.hex()))
|
||||
|
||||
# Get hashes to download
|
||||
hashes_to_download = set(remote_hashes) - set(local_hashes)
|
||||
logger.debug("Hashes to download: {}".format(len(hashes_to_download)))
|
||||
for hash in hashes_to_download:
|
||||
logger.debug("hash to download: {}".format(hash.hex()))
|
||||
|
||||
# Download squeaks for the hashes
|
||||
# TODO: catch exception downloading individual squeak
|
||||
for hash in hashes_to_download:
|
||||
if self.stopped():
|
||||
return
|
||||
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(addresses, min_block, max_block)
|
||||
logger.debug("Got locked hashes: {}".format(len(locked_hashes)))
|
||||
for hash in locked_hashes:
|
||||
logger.debug("locked hash: {}".format(hash.hex()))
|
||||
|
||||
# Get hashes to get offer
|
||||
hashes_to_get_offer = set(remote_hashes) & set(locked_hashes)
|
||||
logger.debug("Hashes to get offer: {}".format(len(hashes_to_get_offer)))
|
||||
for hash in hashes_to_get_offer:
|
||||
logger.debug("hash to get offer: {}".format(hash.hex()))
|
||||
|
||||
# Download offers for the hashes
|
||||
# TODO: catch exception downloading individual squeak
|
||||
for hash in hashes_to_get_offer:
|
||||
if self.stopped():
|
||||
return
|
||||
self._download_offer(hash)
|
||||
|
||||
def upload(
|
||||
self,
|
||||
block_height,
|
||||
lookup_block_interval=LOOKUP_BLOCK_INTERVAL,
|
||||
):
|
||||
# Get list of sharing addresses.
|
||||
addresses = self._get_sharing_addresses()
|
||||
logger.debug("Sharing addresses: {}".format(addresses))
|
||||
min_block = block_height - lookup_block_interval
|
||||
max_block = block_height
|
||||
|
||||
# Get remote hashes
|
||||
remote_hashes = self._get_remote_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got remote hashes: {}".format(len(remote_hashes)))
|
||||
for hash in remote_hashes:
|
||||
logger.debug("remote hash: {}".format(hash.hex()))
|
||||
|
||||
# Get local hashes
|
||||
local_hashes = self._get_local_unlocked_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got local hashes: {}".format(len(local_hashes)))
|
||||
for hash in local_hashes:
|
||||
logger.debug("local hash: {}".format(hash.hex()))
|
||||
|
||||
# Get hashes to upload
|
||||
hashes_to_upload = set(local_hashes) - set(remote_hashes)
|
||||
logger.debug("Hashes to upload: {}".format(len(hashes_to_upload)))
|
||||
for hash in hashes_to_upload:
|
||||
logger.debug("hash to upload: {}".format(hash.hex()))
|
||||
|
||||
# Upload squeaks for the hashes
|
||||
# TODO: catch exception uploading individual squeak
|
||||
for hash in hashes_to_upload:
|
||||
if self.stopped():
|
||||
return
|
||||
self._upload_squeak(hash)
|
||||
|
||||
def download_single_squeak(self, squeak_hash):
|
||||
# Download squeak if not already present.
|
||||
saved_squeak = self._get_saved_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 get_offer(self, squeak_hash):
|
||||
logger.info("Getting offer for squeak hash: {}".format(squeak_hash.hex()))
|
||||
|
||||
# Get the squeak from the squeak hash
|
||||
squeak = self._get_local_squeak(squeak_hash)
|
||||
|
||||
# Get the encryption key
|
||||
encryption_key = squeak.GetEncryptionKey()
|
||||
|
||||
# Create a new challenge
|
||||
challenge_proof = self._generate_challenge_proof()
|
||||
challenge = self._get_challenge(challenge_proof, encryption_key)
|
||||
|
||||
# Download the buy offer
|
||||
offer = self._download_buy_offer(squeak_hash, challenge)
|
||||
|
||||
# Check the proof
|
||||
proof = offer.proof
|
||||
logger.info("Proof: {}".format(proof.hex()))
|
||||
logger.info("Expected proof: {}".format(challenge_proof.hex()))
|
||||
if proof != challenge_proof:
|
||||
raise Exception(
|
||||
"Invalid offer proof: {}, expected: {}".format(
|
||||
proof.hex(),
|
||||
challenge_proof.hex(),
|
||||
)
|
||||
)
|
||||
|
||||
# Get the decoded offer from the payment request string
|
||||
decoded_offer = self._get_decoded_offer(offer)
|
||||
|
||||
# Save the offer
|
||||
self._save_offer(decoded_offer)
|
||||
|
||||
def stop(self):
|
||||
self._stop_event.set()
|
||||
|
||||
def stopped(self):
|
||||
return self._stop_event.is_set()
|
||||
|
||||
def _get_local_hashes(self, addresses, min_block, max_block):
|
||||
return self.squeak_store.lookup_squeaks_include_locked(
|
||||
addresses,
|
||||
min_block,
|
||||
max_block,
|
||||
)
|
||||
|
||||
def _get_local_unlocked_hashes(self, addresses, min_block, max_block):
|
||||
return self.squeak_store.lookup_squeaks(addresses, min_block, max_block)
|
||||
|
||||
def _get_locked_hashes(self, addresses, min_block, max_block):
|
||||
return self.squeak_store.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):
|
||||
self.squeak_store.save_downloaded_squeak(squeak)
|
||||
|
||||
def _get_saved_squeak(self, squeak_hash):
|
||||
return self.squeak_store.get_squeak(squeak_hash)
|
||||
|
||||
def _get_saved_offer(self, squeak_hash):
|
||||
offers = self.postgres_db.get_offers_with_peer(squeak_hash)
|
||||
for offer in offers:
|
||||
if offer.peer_id == peer_id:
|
||||
return offer
|
||||
|
||||
def _download_squeak(self, squeak_hash):
|
||||
logger.info("Downloading squeak: {} from peer: {}".format(squeak_hash.hex(), self.peer.peer_id))
|
||||
squeak = self.peer_client.get_squeak(squeak_hash)
|
||||
self._save_squeak(squeak)
|
||||
|
||||
def _get_followed_addresses(self):
|
||||
followed_profiles = self.postgres_db.get_following_profiles()
|
||||
return [profile.address for profile in followed_profiles]
|
||||
|
||||
def _download_offer(self, squeak_hash):
|
||||
logger.info("Downloading offer for hash: {}".format(squeak_hash.hex()))
|
||||
self.get_offer(squeak_hash)
|
||||
|
||||
def _get_local_squeak(self, squeak_hash):
|
||||
squeak_entry = self.squeak_store.get_squeak(squeak_hash)
|
||||
return squeak_entry.squeak
|
||||
|
||||
def _upload_squeak(self, squeak_hash):
|
||||
logger.info("Uploading squeak: {}".format(squeak_hash.hex()))
|
||||
squeak = self._get_local_squeak(squeak_hash)
|
||||
self.peer_client.post_squeak(squeak)
|
||||
|
||||
def _get_sharing_addresses(self):
|
||||
sharing_profiles = self.postgres_db.get_sharing_profiles()
|
||||
return [profile.address for profile in sharing_profiles]
|
||||
|
||||
def _generate_challenge_proof(self):
|
||||
return generate_data_key()
|
||||
|
||||
def _get_challenge(self, challenge_proof, encryption_key):
|
||||
return encryption_key.encrypt(challenge_proof)
|
||||
|
||||
def _download_buy_offer(self, squeak_hash, challenge):
|
||||
logger.info(
|
||||
"Downloading buy offer for squeak hash: {}".format(squeak_hash.hex())
|
||||
)
|
||||
offer_msg = self.peer_client.buy_squeak(squeak_hash, challenge)
|
||||
offer = self._offer_from_msg(offer_msg)
|
||||
return offer
|
||||
|
||||
def _save_offer(self, offer):
|
||||
logger.info("Saving offer: {}".format(offer))
|
||||
self.postgres_db.insert_offer(offer)
|
||||
|
||||
def _offer_from_msg(self, offer_msg):
|
||||
if not offer_msg:
|
||||
return None
|
||||
return Offer(
|
||||
offer_id=None,
|
||||
squeak_hash=offer_msg.squeak_hash,
|
||||
key_cipher=offer_msg.key_cipher,
|
||||
iv=offer_msg.iv,
|
||||
price_msat=None,
|
||||
payment_hash=offer_msg.preimage_hash,
|
||||
invoice_timestamp=None,
|
||||
invoice_expiry=None,
|
||||
payment_request=offer_msg.payment_request,
|
||||
destination=None,
|
||||
node_host=offer_msg.host,
|
||||
node_port=offer_msg.port,
|
||||
proof=offer_msg.proof,
|
||||
peer_id=self.peer.peer_id,
|
||||
)
|
||||
|
||||
def _decode_payment_request(self, payment_request):
|
||||
return self.lightning_client.decode_pay_req(payment_request)
|
||||
|
||||
def _get_decoded_offer(self, offer):
|
||||
pay_req = self._decode_payment_request(offer.payment_request)
|
||||
logger.info("Decoded payment request: {}".format(pay_req))
|
||||
|
||||
price_msat = pay_req.num_msat
|
||||
destination = pay_req.destination
|
||||
invoice_timestamp = pay_req.timestamp
|
||||
invoice_expiry = pay_req.expiry
|
||||
node_host = offer.node_host or self.peer.host
|
||||
node_port = offer.node_port
|
||||
|
||||
logger.info("price_msat: {}".format(price_msat))
|
||||
logger.info("destination: {}".format(destination))
|
||||
logger.info("invoice_timestamp: {}".format(invoice_timestamp))
|
||||
logger.info("invoice_expiry: {}".format(invoice_expiry))
|
||||
logger.info("node_host: {}".format(node_host))
|
||||
logger.info("node_port: {}".format(node_port))
|
||||
|
||||
decoded_offer = Offer(
|
||||
offer_id=offer.offer_id,
|
||||
squeak_hash=offer.squeak_hash,
|
||||
key_cipher=offer.key_cipher,
|
||||
iv=offer.iv,
|
||||
price_msat=price_msat,
|
||||
payment_hash=offer.payment_hash,
|
||||
invoice_timestamp=invoice_timestamp,
|
||||
invoice_expiry=invoice_expiry,
|
||||
payment_request=offer.payment_request,
|
||||
destination=destination,
|
||||
node_host=node_host,
|
||||
node_port=node_port,
|
||||
proof=offer.proof,
|
||||
peer_id=offer.peer_id,
|
||||
)
|
||||
|
||||
return decoded_offer
|
||||
|
|
@ -2,6 +2,7 @@ import logging
|
|||
import threading
|
||||
|
||||
from squeaknode.network.peer_client import PeerClient
|
||||
from squeaknode.node.peer_task import PeerSyncTask
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -9,90 +10,48 @@ logger = logging.getLogger(__name__)
|
|||
LOOKUP_BLOCK_INTERVAL = 1008 # 1 week
|
||||
|
||||
|
||||
class PeerUpload:
|
||||
def __init__(
|
||||
self,
|
||||
peer,
|
||||
block_height,
|
||||
squeak_store,
|
||||
postgres_db,
|
||||
lookup_block_interval=LOOKUP_BLOCK_INTERVAL,
|
||||
):
|
||||
self.peer = peer
|
||||
self.block_height = block_height
|
||||
self.squeak_store = squeak_store
|
||||
self.postgres_db = postgres_db
|
||||
self.lookup_block_interval = lookup_block_interval
|
||||
# class PeerUpload(PeerSyncTask):
|
||||
# def __init__(
|
||||
# self,
|
||||
# peer,
|
||||
# squeak_store,
|
||||
# postgres_db,
|
||||
# lightning_client,
|
||||
# ):
|
||||
# super().__init__(peer, squeak_store, postgres_db, lightning_client)
|
||||
|
||||
self.peer_client = PeerClient(
|
||||
self.peer.host,
|
||||
self.peer.port,
|
||||
)
|
||||
# def upload(
|
||||
# self,
|
||||
# block_height,
|
||||
# lookup_block_interval=LOOKUP_BLOCK_INTERVAL,
|
||||
# ):
|
||||
# # Get list of sharing addresses.
|
||||
# addresses = self._get_sharing_addresses()
|
||||
# logger.debug("Sharing addresses: {}".format(addresses))
|
||||
# min_block = block_height - lookup_block_interval
|
||||
# max_block = block_height
|
||||
|
||||
self._stop_event = threading.Event()
|
||||
# # Get remote hashes
|
||||
# remote_hashes = self._get_remote_hashes(addresses, min_block, max_block)
|
||||
# logger.debug("Got remote hashes: {}".format(len(remote_hashes)))
|
||||
# for hash in remote_hashes:
|
||||
# logger.debug("remote hash: {}".format(hash.hex()))
|
||||
|
||||
def upload(self):
|
||||
# Get list of sharing addresses.
|
||||
addresses = self._get_sharing_addresses()
|
||||
logger.debug("Sharing addresses: {}".format(addresses))
|
||||
min_block = self.block_height - self.lookup_block_interval
|
||||
max_block = self.block_height
|
||||
# # Get local hashes
|
||||
# local_hashes = self._get_local_hashes(addresses, min_block, max_block)
|
||||
# logger.debug("Got local hashes: {}".format(len(local_hashes)))
|
||||
# for hash in local_hashes:
|
||||
# logger.debug("local hash: {}".format(hash.hex()))
|
||||
|
||||
if self.stopped():
|
||||
return
|
||||
# # Get hashes to upload
|
||||
# hashes_to_upload = set(local_hashes) - set(remote_hashes)
|
||||
# logger.debug("Hashes to upload: {}".format(len(hashes_to_upload)))
|
||||
# for hash in hashes_to_upload:
|
||||
# logger.debug("hash to upload: {}".format(hash.hex()))
|
||||
|
||||
# Get remote hashes
|
||||
remote_hashes = self._get_remote_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got remote hashes: {}".format(len(remote_hashes)))
|
||||
for hash in remote_hashes:
|
||||
logger.debug("remote hash: {}".format(hash.hex()))
|
||||
|
||||
if self.stopped():
|
||||
return
|
||||
|
||||
# Get local hashes
|
||||
local_hashes = self._get_local_hashes(addresses, min_block, max_block)
|
||||
logger.debug("Got local hashes: {}".format(len(local_hashes)))
|
||||
for hash in local_hashes:
|
||||
logger.debug("local hash: {}".format(hash.hex()))
|
||||
|
||||
if self.stopped():
|
||||
return
|
||||
|
||||
# Get hashes to upload
|
||||
hashes_to_upload = set(local_hashes) - set(remote_hashes)
|
||||
logger.debug("Hashes to upload: {}".format(len(hashes_to_upload)))
|
||||
for hash in hashes_to_upload:
|
||||
logger.debug("hash to upload: {}".format(hash.hex()))
|
||||
|
||||
# Upload squeaks for the hashes
|
||||
# TODO: catch exception uploading individual squeak
|
||||
for hash in hashes_to_upload:
|
||||
if self.stopped():
|
||||
return
|
||||
self._upload_squeak(hash)
|
||||
|
||||
def stop(self):
|
||||
self._stop_event.set()
|
||||
|
||||
def stopped(self):
|
||||
return self._stop_event.is_set()
|
||||
|
||||
def _get_local_hashes(self, addresses, min_block, max_block):
|
||||
return self.squeak_store.lookup_squeaks(addresses, min_block, max_block)
|
||||
|
||||
def _get_remote_hashes(self, addresses, min_block, max_block):
|
||||
return self.peer_client.lookup_squeaks(addresses, min_block, max_block)
|
||||
|
||||
def _get_local_squeak(self, squeak_hash):
|
||||
squeak_entry = self.squeak_store.get_squeak(squeak_hash)
|
||||
return squeak_entry.squeak
|
||||
|
||||
def _upload_squeak(self, squeak_hash):
|
||||
logger.info("Uploading squeak: {}".format(squeak_hash.hex()))
|
||||
squeak = self._get_local_squeak(squeak_hash)
|
||||
self.peer_client.post_squeak(squeak)
|
||||
|
||||
def _get_sharing_addresses(self):
|
||||
sharing_profiles = self.postgres_db.get_sharing_profiles()
|
||||
return [profile.address for profile in sharing_profiles]
|
||||
# # Upload squeaks for the hashes
|
||||
# # TODO: catch exception uploading individual squeak
|
||||
# for hash in hashes_to_upload:
|
||||
# if self.stopped():
|
||||
# return
|
||||
# self._upload_squeak(hash)
|
||||
|
|
|
|||
|
|
@ -9,7 +9,6 @@ from squeak.core.signing import CSigningKey, CSqueakAddress
|
|||
from squeak.core import CheckSqueak
|
||||
|
||||
from squeaknode.core.squeak_address_validator import SqueakAddressValidator
|
||||
from squeaknode.node.peer_download import PeerDownload
|
||||
from squeaknode.node.squeak_block_periodic_worker import SqueakBlockPeriodicWorker
|
||||
from squeaknode.node.squeak_block_queue_worker import SqueakBlockQueueWorker
|
||||
from squeaknode.node.squeak_block_verifier import SqueakBlockVerifier
|
||||
|
|
|
|||
|
|
@ -3,8 +3,9 @@ import threading
|
|||
|
||||
from collections import defaultdict
|
||||
|
||||
from squeaknode.node.peer_download import PeerDownload
|
||||
from squeaknode.node.peer_upload import PeerUpload
|
||||
from squeaknode.node.network_task import DownloadTimelineNetworkSyncTask
|
||||
from squeaknode.node.network_task import UploadTimelineNetworkSyncTask
|
||||
from squeaknode.node.network_task import SingleSqueakNetworkSyncTask
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -12,55 +13,58 @@ logger = logging.getLogger(__name__)
|
|||
HOUR_IN_SECONDS = 3600
|
||||
|
||||
|
||||
class SqueakSyncStatus:
|
||||
def __init__(self):
|
||||
self.downloads = {}
|
||||
self.uploads = {}
|
||||
self.single_squeak_downloads = defaultdict(dict)
|
||||
# class SqueakSyncStatus:
|
||||
# def __init__(self):
|
||||
# self.downloads = {}
|
||||
# self.uploads = {}
|
||||
# self.single_squeak_downloads = defaultdict(dict)
|
||||
|
||||
def add_download(self, peer, peer_download):
|
||||
self.downloads[peer.peer_id] = peer_download
|
||||
# def add_download(self, peer, peer_download):
|
||||
# self.downloads[peer.peer_id] = peer_download
|
||||
|
||||
def add_upload(self, peer, peer_upload):
|
||||
self.uploads[peer.peer_id] = peer_upload
|
||||
# def add_upload(self, peer, peer_upload):
|
||||
# self.uploads[peer.peer_id] = peer_upload
|
||||
|
||||
def add_single_squeak_download(self, squeak_hash, peer, peer_download):
|
||||
self.single_squeak_downloads[squeak_hash][peer.peer_id] = peer_download
|
||||
# def add_single_squeak_download(self, squeak_hash, peer, peer_download):
|
||||
# self.single_squeak_downloads[squeak_hash][peer.peer_id] = peer_download
|
||||
|
||||
def is_downloading(self, peer):
|
||||
return peer.peer_id in self.downloads
|
||||
# def is_downloading(self, peer):
|
||||
# return peer.peer_id in self.downloads
|
||||
|
||||
def is_uploading(self, peer):
|
||||
return peer.peer_id in self.uploads
|
||||
# def is_uploading(self, peer):
|
||||
# return peer.peer_id in self.uploads
|
||||
|
||||
def is_downloading_single_squeak(self, squeak_hash, peer):
|
||||
return peer.peer_id in self.single_squeak_downloads[squeak_hash]
|
||||
# def is_downloading_single_squeak(self, squeak_hash, peer):
|
||||
# return peer.peer_id in self.single_squeak_downloads[squeak_hash]
|
||||
|
||||
def remove_download(self, peer):
|
||||
del self.downloads[peer.peer_id]
|
||||
# def remove_download(self, peer):
|
||||
# del self.downloads[peer.peer_id]
|
||||
|
||||
def remove_upload(self, peer):
|
||||
del self.uploads[peer.peer_id]
|
||||
# def remove_upload(self, peer):
|
||||
# del self.uploads[peer.peer_id]
|
||||
|
||||
def remove_single_peer_download(self, squeak_hash, peer):
|
||||
del self.single_squeak_downloads[squeak_hash][peer.peer_id]
|
||||
# def remove_single_peer_download(self, squeak_hash, peer):
|
||||
# del self.single_squeak_downloads[squeak_hash][peer.peer_id]
|
||||
|
||||
def get_current_downloads(self):
|
||||
return self.downloads.values()
|
||||
# def get_current_downloads(self):
|
||||
# return self.downloads.values()
|
||||
|
||||
def get_current_uploads(self):
|
||||
return self.uploads.values()
|
||||
# def get_current_uploads(self):
|
||||
# return self.uploads.values()
|
||||
|
||||
|
||||
class SqueakSyncController:
|
||||
def __init__(self, blockchain_client, squeak_store, postgres_db, lightning_client):
|
||||
self.squeak_sync_status = SqueakSyncStatus()
|
||||
self.blockchain_client = blockchain_client
|
||||
self.squeak_store = squeak_store
|
||||
self.postgres_db = postgres_db
|
||||
self.lightning_client = lightning_client
|
||||
|
||||
def sync_peers(self, peers):
|
||||
self.download_timeline(peers)
|
||||
self.upload_timeline(peers)
|
||||
|
||||
def download_timeline(self, peers):
|
||||
try:
|
||||
block_info = self.blockchain_client.get_best_block_info()
|
||||
block_height = block_info.block_height
|
||||
|
|
@ -69,138 +73,168 @@ class SqueakSyncController:
|
|||
"Failed to sync because unable to get blockchain info.", exc_info=False
|
||||
)
|
||||
return
|
||||
self._download_from_peers(peers, block_height)
|
||||
self._upload_to_peers(peers, block_height)
|
||||
dowload_timeline_task = DownloadTimelineNetworkSyncTask(
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
block_height,
|
||||
)
|
||||
dowload_timeline_task.sync(peers)
|
||||
|
||||
def upload_timeline(self, peers):
|
||||
try:
|
||||
block_info = self.blockchain_client.get_best_block_info()
|
||||
block_height = block_info.block_height
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Failed to sync because unable to get blockchain info.", exc_info=False
|
||||
)
|
||||
return
|
||||
upload_timeline_task = UploadTimelineNetworkSyncTask(
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
block_height,
|
||||
)
|
||||
upload_timeline_task.sync(peers)
|
||||
|
||||
def download_single_squeak_from_peers(self, squeak_hash, peers):
|
||||
for peer in peers:
|
||||
if peer.downloading:
|
||||
download_thread = threading.Thread(
|
||||
target=self._download_single_squeak_from_peer,
|
||||
args=(
|
||||
squeak_hash,
|
||||
peer,
|
||||
),
|
||||
)
|
||||
download_thread.start()
|
||||
|
||||
def _download_from_peers(self, peers, block_height):
|
||||
for peer in peers:
|
||||
if peer.downloading:
|
||||
download_thread = threading.Thread(
|
||||
target=self._download_from_peer,
|
||||
args=(
|
||||
peer,
|
||||
block_height,
|
||||
),
|
||||
)
|
||||
download_thread.start()
|
||||
|
||||
def _upload_to_peers(self, peers, block_height):
|
||||
for peer in peers:
|
||||
if peer.uploading:
|
||||
upload_thread = threading.Thread(
|
||||
target=self._upload_to_peer,
|
||||
args=(
|
||||
peer,
|
||||
block_height,
|
||||
),
|
||||
)
|
||||
upload_thread.start()
|
||||
|
||||
def _download_from_peer(self, peer, block_height):
|
||||
peer_download = PeerDownload(
|
||||
peer,
|
||||
# for peer in peers:
|
||||
# if peer.downloading:
|
||||
# download_thread = threading.Thread(
|
||||
# target=self._download_single_squeak_from_peer,
|
||||
# args=(
|
||||
# squeak_hash,
|
||||
# peer,
|
||||
# ),
|
||||
# )
|
||||
# download_thread.start()
|
||||
timeline_sync_task = SingleSqueakNetworkSyncTask(
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
squeak_hash,
|
||||
)
|
||||
try:
|
||||
logger.debug("Trying to download from peer: {}".format(peer))
|
||||
with self.DownloadingContextManager(
|
||||
peer, peer_download, self.squeak_sync_status
|
||||
) as downloading_manager:
|
||||
peer_download.download(block_height)
|
||||
except Exception as e:
|
||||
logger.error("Download from peer failed.", exc_info=True)
|
||||
timeline_sync_task.sync(peers)
|
||||
|
||||
def _upload_to_peer(self, peer, block_height):
|
||||
peer_upload = PeerUpload(
|
||||
peer,
|
||||
block_height,
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
)
|
||||
try:
|
||||
logger.debug("Trying to upload to peer: {}".format(peer))
|
||||
with self.UploadingContextManager(
|
||||
peer, peer_upload, self.squeak_sync_status
|
||||
) as uploading_manager:
|
||||
peer_upload.upload()
|
||||
except Exception as e:
|
||||
logger.error("Upload from peer failed.", exc_info=True)
|
||||
|
||||
def _download_single_squeak_from_peer(self, squeak_hash, peer):
|
||||
peer_download = PeerDownload(
|
||||
peer,
|
||||
self.squeak_store,
|
||||
self.postgres_db,
|
||||
self.lightning_client,
|
||||
)
|
||||
try:
|
||||
logger.debug("Trying to download single squeak {} from peer: {}".format(squeak_hash, peer))
|
||||
with self.SingleSqueakDownloadingContextManager(
|
||||
squeak_hash, peer, peer_download, self.squeak_sync_status
|
||||
) as downloading_manager:
|
||||
peer_download.download_single_squeak(squeak_hash)
|
||||
except Exception as e:
|
||||
logger.error("Download single squeak from peer failed.", exc_info=True)
|
||||
# def _download_from_peers(self, peers, block_height):
|
||||
# for peer in peers:
|
||||
# if peer.downloading:
|
||||
# download_thread = threading.Thread(
|
||||
# target=self._download_from_peer,
|
||||
# args=(
|
||||
# peer,
|
||||
# block_height,
|
||||
# ),
|
||||
# )
|
||||
# download_thread.start()
|
||||
|
||||
class DownloadingContextManager:
|
||||
def __init__(self, peer, peer_download, squeak_sync_status):
|
||||
self.peer = peer
|
||||
self.peer_download = peer_download
|
||||
self.squeak_sync_status = squeak_sync_status
|
||||
# def _upload_to_peers(self, peers, block_height):
|
||||
# for peer in peers:
|
||||
# if peer.uploading:
|
||||
# upload_thread = threading.Thread(
|
||||
# target=self._upload_to_peer,
|
||||
# args=(
|
||||
# peer,
|
||||
# block_height,
|
||||
# ),
|
||||
# )
|
||||
# upload_thread.start()
|
||||
|
||||
if self.squeak_sync_status.is_downloading(self.peer):
|
||||
raise Exception("Peer is already downloading: {}".format(self.peer))
|
||||
# def _download_from_peer(self, peer, block_height):
|
||||
# peer_download = PeerDownload(
|
||||
# peer,
|
||||
# self.squeak_store,
|
||||
# self.postgres_db,
|
||||
# self.lightning_client,
|
||||
# )
|
||||
# try:
|
||||
# logger.debug("Trying to download from peer: {}".format(peer))
|
||||
# with self.DownloadingContextManager(
|
||||
# peer, peer_download, self.squeak_sync_status
|
||||
# ) as downloading_manager:
|
||||
# peer_download.download(block_height)
|
||||
# except Exception as e:
|
||||
# logger.error("Download from peer failed.", exc_info=True)
|
||||
|
||||
def __enter__(self):
|
||||
self.squeak_sync_status.add_download(self.peer, self.peer_download)
|
||||
return self
|
||||
# def _upload_to_peer(self, peer, block_height):
|
||||
# peer_upload = PeerUpload(
|
||||
# peer,
|
||||
# self.squeak_store,
|
||||
# self.postgres_db,
|
||||
# self.lightning_client,
|
||||
# )
|
||||
# try:
|
||||
# logger.debug("Trying to upload to peer: {}".format(peer))
|
||||
# with self.UploadingContextManager(
|
||||
# peer, peer_upload, self.squeak_sync_status
|
||||
# ) as uploading_manager:
|
||||
# peer_upload.upload(block_height)
|
||||
# except Exception as e:
|
||||
# logger.error("Upload from peer failed.", exc_info=True)
|
||||
|
||||
def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
self.squeak_sync_status.remove_download(self.peer)
|
||||
# def _download_single_squeak_from_peer(self, squeak_hash, peer):
|
||||
# peer_download = PeerSingleSqueakDownload(
|
||||
# peer,
|
||||
# self.squeak_store,
|
||||
# self.postgres_db,
|
||||
# self.lightning_client,
|
||||
# )
|
||||
# try:
|
||||
# logger.debug("Trying to download single squeak {} from peer: {}".format(squeak_hash, peer))
|
||||
# with self.SingleSqueakDownloadingContextManager(
|
||||
# squeak_hash, peer, peer_download, self.squeak_sync_status
|
||||
# ) as downloading_manager:
|
||||
# peer_download.download_single_squeak(squeak_hash)
|
||||
# except Exception as e:
|
||||
# logger.error("Download single squeak from peer failed.", exc_info=True)
|
||||
|
||||
class UploadingContextManager:
|
||||
def __init__(self, peer, peer_upload, squeak_sync_status):
|
||||
self.peer = peer
|
||||
self.peer_upload = peer_upload
|
||||
self.squeak_sync_status = squeak_sync_status
|
||||
# class DownloadingContextManager:
|
||||
# def __init__(self, peer, peer_download, squeak_sync_status):
|
||||
# self.peer = peer
|
||||
# self.peer_download = peer_download
|
||||
# self.squeak_sync_status = squeak_sync_status
|
||||
|
||||
if self.squeak_sync_status.is_uploading(self.peer):
|
||||
raise Exception("Peer is already uploading: {}".format(self.peer))
|
||||
# if self.squeak_sync_status.is_downloading(self.peer):
|
||||
# raise Exception("Peer is already downloading: {}".format(self.peer))
|
||||
|
||||
def __enter__(self):
|
||||
self.squeak_sync_status.add_upload(self.peer, self.peer_upload)
|
||||
return self
|
||||
# def __enter__(self):
|
||||
# self.squeak_sync_status.add_download(self.peer, self.peer_download)
|
||||
# return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
self.squeak_sync_status.remove_upload(self.peer)
|
||||
# def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
# self.squeak_sync_status.remove_download(self.peer)
|
||||
|
||||
class SingleSqueakDownloadingContextManager:
|
||||
def __init__(self, squeak_hash, peer, peer_download, squeak_sync_status):
|
||||
self.squeak_hash = squeak_hash
|
||||
self.peer = peer
|
||||
self.peer_download = peer_download
|
||||
self.squeak_sync_status = squeak_sync_status
|
||||
# class UploadingContextManager:
|
||||
# def __init__(self, peer, peer_upload, squeak_sync_status):
|
||||
# self.peer = peer
|
||||
# self.peer_upload = peer_upload
|
||||
# self.squeak_sync_status = squeak_sync_status
|
||||
|
||||
if self.squeak_sync_status.is_downloading_single_squeak(self.squeak_hash, self.peer):
|
||||
raise Exception("Peer {} is already downloading hash: {}".format(self.peer, self.squeak_hash))
|
||||
# if self.squeak_sync_status.is_uploading(self.peer):
|
||||
# raise Exception("Peer is already uploading: {}".format(self.peer))
|
||||
|
||||
def __enter__(self):
|
||||
self.squeak_sync_status.add_single_squeak_download(self.squeak_hash, self.peer, self.peer_download)
|
||||
return self
|
||||
# def __enter__(self):
|
||||
# self.squeak_sync_status.add_upload(self.peer, self.peer_upload)
|
||||
# return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
self.squeak_sync_status.remove_single_peer_download(self.squeak_hash, self.peer)
|
||||
# def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
# self.squeak_sync_status.remove_upload(self.peer)
|
||||
|
||||
# class SingleSqueakDownloadingContextManager:
|
||||
# def __init__(self, squeak_hash, peer, peer_download, squeak_sync_status):
|
||||
# self.squeak_hash = squeak_hash
|
||||
# self.peer = peer
|
||||
# self.peer_download = peer_download
|
||||
# self.squeak_sync_status = squeak_sync_status
|
||||
|
||||
# if self.squeak_sync_status.is_downloading_single_squeak(self.squeak_hash, self.peer):
|
||||
# raise Exception("Peer {} is already downloading hash: {}".format(self.peer, self.squeak_hash))
|
||||
|
||||
# def __enter__(self):
|
||||
# self.squeak_sync_status.add_single_squeak_download(self.squeak_hash, self.peer, self.peer_download)
|
||||
# return self
|
||||
|
||||
# def __exit__(self, exc_type, exc_value, exc_traceback):
|
||||
# self.squeak_sync_status.remove_single_peer_download(self.squeak_hash, self.peer)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue