lbry-sdk/lbrynet/core/server/BlobRequestHandler.py

import logging

from twisted.internet import defer
from twisted.protocols.basic import FileSender
from twisted.python.failure import Failure
from zope.interface import implements

from lbrynet import analytics
from lbrynet.core.Offer import Offer
from lbrynet.interfaces import IQueryHandlerFactory, IQueryHandler, IBlobSender

log = logging.getLogger(__name__)


class BlobRequestHandlerFactory(object):
    #implements(IQueryHandlerFactory)

    def __init__(self, blob_manager, wallet, payment_rate_manager, analytics_manager):
        self.blob_manager = blob_manager
        self.wallet = wallet
        self.payment_rate_manager = payment_rate_manager
        self.analytics_manager = analytics_manager

    ######### IQueryHandlerFactory #########

    def build_query_handler(self):
        q_h = BlobRequestHandler(
            self.blob_manager, self.wallet, self.payment_rate_manager, self.analytics_manager)
        return q_h

    def get_primary_query_identifier(self):
        return 'requested_blob'

    def get_description(self):
        return "Blob Uploader - uploads blobs"


class BlobRequestHandler(object):
    #implements(IQueryHandler, IBlobSender)
    PAYMENT_RATE_QUERY = 'blob_data_payment_rate'
    BLOB_QUERY = 'requested_blob'
    AVAILABILITY_QUERY = 'requested_blobs'

    def __init__(self, blob_manager, wallet, payment_rate_manager, analytics_manager):
        self.blob_manager = blob_manager
        self.payment_rate_manager = payment_rate_manager
        self.wallet = wallet
        self.query_identifiers = [self.PAYMENT_RATE_QUERY, self.BLOB_QUERY, self.AVAILABILITY_QUERY]
        self.analytics_manager = analytics_manager
        self.peer = None
        self.blob_data_payment_rate = None
        self.read_handle = None
        self.currently_uploading = None
        self.file_sender = None
        self.blob_bytes_uploaded = 0
        self._blobs_requested = []

    ######### IQueryHandler #########

    def register_with_request_handler(self, request_handler, peer):
        self.peer = peer
        request_handler.register_query_handler(self, self.query_identifiers)
        request_handler.register_blob_sender(self)

    def handle_queries(self, queries):
        response = defer.succeed({})
        log.debug("Handle query: %s", str(queries))

        if self.AVAILABILITY_QUERY in queries:
            self._blobs_requested = queries[self.AVAILABILITY_QUERY]
            response.addCallback(lambda r: self._reply_to_availability(r, self._blobs_requested))
        if self.PAYMENT_RATE_QUERY in queries:
            offered_rate = queries[self.PAYMENT_RATE_QUERY]
            offer = Offer(offered_rate)
            if offer.rate is None:
                log.warning("Empty rate offer")
            response.addCallback(lambda r: self._handle_payment_rate_query(offer, r))
        if self.BLOB_QUERY in queries:
            incoming = queries[self.BLOB_QUERY]
            response.addCallback(lambda r: self._reply_to_send_request(r, incoming))
        return response

    ######### IBlobSender #########

    def send_blob_if_requested(self, consumer):
        if self.currently_uploading is not None:
            return self.send_file(consumer)
        return defer.succeed(True)

    def cancel_send(self, err):
        if self.currently_uploading is not None:
            self.read_handle.close()
        self.read_handle = None
        self.currently_uploading = None
        return err

    ######### internal #########

    def _reply_to_availability(self, request, blobs):
        d = self._get_available_blobs(blobs)

        def set_available(available_blobs):
            log.debug("available blobs: %s", str(available_blobs))
            request.update({'available_blobs': available_blobs})
            return request

        d.addCallback(set_available)
        return d

    def _handle_payment_rate_query(self, offer, request):
        blobs = self._blobs_requested
        log.debug("Offered rate %f LBC/mb for %i blobs", offer.rate, len(blobs))
        reply = self.payment_rate_manager.reply_to_offer(self.peer, blobs, offer)
        if reply.is_accepted:
            self.blob_data_payment_rate = offer.rate
            request[self.PAYMENT_RATE_QUERY] = "RATE_ACCEPTED"
            log.debug("Accepted rate: %f", offer.rate)
        elif reply.is_too_low:
            request[self.PAYMENT_RATE_QUERY] = "RATE_TOO_LOW"
            log.debug("Reject rate: %f", offer.rate)
        elif reply.is_unset:
            log.warning("Rate unset")
            request['incoming_blob'] = {'error': 'RATE_UNSET'}
        log.debug("Returning rate query result: %s", str(request))

        return request

    def _handle_blob_query(self, response, query):
        log.debug("Received the client's request to send a blob")
        response['incoming_blob'] = {}

        if self.blob_data_payment_rate is None:
            response['incoming_blob'] = {'error': "RATE_UNSET"}
            return response
        else:
            return self._send_blob(response, query)

    def _send_blob(self, response, query):
        d = self.blob_manager.get_blob(query)
        d.addCallback(self.open_blob_for_reading, response)
        return d

    def open_blob_for_reading(self, blob, response):
        response_fields = {}
        d = defer.succeed(None)
        if blob.get_is_verified():
            read_handle = blob.open_for_reading()
            if read_handle is not None:
                self.currently_uploading = blob
                self.read_handle = read_handle
                log.info("Sending %s to %s", str(blob), self.peer)
                response_fields['blob_hash'] = blob.blob_hash
                response_fields['length'] = blob.length
                response['incoming_blob'] = response_fields
                d.addCallback(lambda _: response)
                return d
        log.debug("We can not send %s", str(blob))
        response['incoming_blob'] = {'error': 'BLOB_UNAVAILABLE'}
        d.addCallback(lambda _: response)
        return d

    def _reply_to_send_request(self, response, incoming):
        response_fields = {}
        response['incoming_blob'] = response_fields

        if self.blob_data_payment_rate is None:
            log.debug("Rate not set yet")
            response['incoming_blob'] = {'error': 'RATE_UNSET'}
            return defer.succeed(response)
        else:
            log.debug("Requested blob: %s", str(incoming))
            d = self.blob_manager.get_blob(incoming)
            d.addCallback(lambda blob: self.open_blob_for_reading(blob, response))
            return d

    def _get_available_blobs(self, requested_blobs):
        d = self.blob_manager.completed_blobs(requested_blobs)
        return d

    def send_file(self, consumer):

        def _send_file():
            inner_d = start_transfer()
            # TODO: if the transfer fails, check if it's because the connection was cut off.
            # TODO: if so, perhaps bill the client
            inner_d.addCallback(lambda _: set_expected_payment())
            inner_d.addBoth(set_not_uploading)
            return inner_d

        def count_bytes(data):
            uploaded = len(data)
            self.blob_bytes_uploaded += uploaded
            self.peer.update_stats('blob_bytes_uploaded', uploaded)
            if self.analytics_manager is not None:
                self.analytics_manager.add_observation(analytics.BLOB_BYTES_UPLOADED, uploaded)
            return data

        def start_transfer():
            self.file_sender = FileSender()
            log.debug("Starting the file upload")
            assert self.read_handle is not None, \
                "self.read_handle was None when trying to start the transfer"
            d = self.file_sender.beginFileTransfer(self.read_handle, consumer, count_bytes)
            return d

        def set_expected_payment():
            log.debug("Setting expected payment")
            if (
                self.blob_bytes_uploaded != 0 and self.blob_data_payment_rate is not None
                and self.blob_data_payment_rate > 0
            ):
                # TODO: explain why 2**20
                self.wallet.add_expected_payment(self.peer,
                                                 self.currently_uploading.length * 1.0 *
                                                 self.blob_data_payment_rate / 2 ** 20)
                self.blob_bytes_uploaded = 0
            self.peer.update_stats('blobs_uploaded', 1)
            return None

        def set_not_uploading(reason=None):
            if self.currently_uploading is not None:
                self.read_handle.close()
                self.read_handle = None
                self.currently_uploading = None
            self.file_sender = None
            if reason is not None and isinstance(reason, Failure):
                log.warning("Upload has failed. Reason: %s", reason.getErrorMessage())

        return _send_file()