forked from LBRYCommunity/lbry-sdk
6fae07d29e
-add {‘sd_blob_hash’: …, ‘sd_blob_size ‘: …} query type with
{‘send_sd_blob’: True/False, ‘needed_blobs’: []} response
this allows the reflector client to know how much of a stream reflector
already has covered, as to minimize the number of subsequent requests
and prevent streams from being partially reflected
-remove empty {} request
47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
import logging
|
|
from twisted.internet import reactor
|
|
from twisted.internet.error import ConnectionLost, ConnectionDone
|
|
from lbrynet.reflector import BlobClientFactory, ClientFactory
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _check_if_reflector_has_stream(lbry_file, reflector_server):
|
|
reflector_address, reflector_port = reflector_server[0], reflector_server[1]
|
|
factory = BlobClientFactory(
|
|
lbry_file.blob_manager,
|
|
[lbry_file.sd_hash]
|
|
)
|
|
d = reactor.resolve(reflector_address)
|
|
d.addCallback(lambda ip: reactor.connectTCP(ip, reflector_port, factory))
|
|
d.addCallback(lambda _: factory.finished_deferred)
|
|
d.addCallback(lambda _: not factory.sent_blobs)
|
|
return d
|
|
|
|
|
|
def _reflect_stream(lbry_file, reflector_server):
|
|
reflector_address, reflector_port = reflector_server[0], reflector_server[1]
|
|
factory = ClientFactory(
|
|
lbry_file.blob_manager,
|
|
lbry_file.stream_info_manager,
|
|
lbry_file.stream_hash
|
|
)
|
|
d = reactor.resolve(reflector_address)
|
|
d.addCallback(lambda ip: reactor.connectTCP(ip, reflector_port, factory))
|
|
d.addCallback(lambda _: factory.finished_deferred)
|
|
d.addCallback(lambda reflected_blobs: log.info("Reflected %i blobs for lbry://%s",
|
|
len(reflected_blobs),
|
|
lbry_file.uri))
|
|
return d
|
|
|
|
|
|
def _catch_error(err, uri):
|
|
msg = "An error occurred while checking availability for lbry://%s: %s"
|
|
log.error(msg, uri, err.getTraceback())
|
|
|
|
|
|
def check_and_restore_availability(lbry_file, reflector_server):
|
|
d = _reflect_stream(lbry_file, reflector_server)
|
|
d.addErrback(lambda err: err.trap(ConnectionDone, ConnectionLost))
|
|
d.addErrback(_catch_error, lbry_file.uri)
|
|
return d
|