lbry-sdk/lbrynet/lbrynet_daemon/LBRYDaemonServer.py

import logging
import os
import shutil
import json
import sys
import mimetypes

from datetime import datetime
from appdirs import user_data_dir
from twisted.web import server, static, resource
from twisted.internet import defer, interfaces, error, reactor, task, threads

from zope.interface import implements

from lbrynet.lbrynet_daemon.LBRYDaemon import LBRYDaemon
from lbrynet.conf import API_CONNECTION_STRING, API_ADDRESS, DEFAULT_WALLET, UI_ADDRESS, DEFAULT_UI_BRANCH, LOG_FILE_NAME


if sys.platform != "darwin":
    data_dir = os.path.join(os.path.expanduser("~"), ".lbrynet")
else:
    data_dir = user_data_dir("LBRY")
if not os.path.isdir(data_dir):
    os.mkdir(data_dir)

lbrynet_log = os.path.join(data_dir, LOG_FILE_NAME)
log = logging.getLogger(__name__)
handler = logging.handlers.RotatingFileHandler(lbrynet_log, maxBytes=2097152, backupCount=5)
log.addHandler(handler)
log.setLevel(logging.INFO)


class LBRYindex(resource.Resource):
    def __init__(self, ui_dir):
        resource.Resource.__init__(self)
        self.ui_dir = ui_dir

    isLeaf = False

    def _delayed_render(self, request, results):
        request.write(str(results))
        request.finish()

    def getChild(self, name, request):
        if name == '':
            return self
        return resource.Resource.getChild(self, name, request)

    def render_GET(self, request):
        return static.File(os.path.join(self.ui_dir, "index.html")).render_GET(request)


class LBRYFileStreamer(object):
    """
    Writes downloaded LBRY file to request as the download comes in, pausing and resuming as requested
    used for Chrome
    """

    implements(interfaces.IPushProducer)

    def __init__(self, request, path, start, stop, size):
        self._request = request
        self._fileObject = file(path)
        self._content_type = mimetypes.guess_type(path)[0]
        self._stop_pos = size - 1 if stop == '' else int(stop)  #chrome and firefox send range requests for "0-"
        self._cursor = self._start_pos = int(start)
        self._file_size = size
        self._depth = 0

        self._paused = self._sent_bytes = self._stopped = False
        self._delay = 0.25
        self._deferred = defer.succeed(None)

        self._request.setResponseCode(206)
        self._request.setHeader('accept-ranges', 'bytes')
        self._request.setHeader('content-type', self._content_type)

        self.resumeProducing()

    def pauseProducing(self):
        self._paused = True
        log.info("[" + str(datetime.now()) + "] Pausing producer")
        return defer.succeed(None)

    def resumeProducing(self):
        def _check_for_new_data():
            self._depth += 1
            self._fileObject.seek(self._start_pos, os.SEEK_END)
            readable_bytes = self._fileObject.tell()
            self._fileObject.seek(self._cursor)

            self._sent_bytes = False

            if (readable_bytes > self._cursor) and not (self._stopped or self._paused):
                read_length = min(readable_bytes, self._stop_pos) - self._cursor + 1
                self._request.setHeader('content-range', 'bytes %s-%s/%s' % (self._cursor, self._cursor + read_length - 1, self._file_size))
                self._request.setHeader('content-length', str(read_length))
                start_cur = self._cursor
                for i in range(read_length):
                    if self._paused or self._stopped:
                        break
                    else:
                        data = self._fileObject.read(1)
                        self._request.write(data)
                        self._cursor += 1

                log.info("[" + str(datetime.now()) + "] Wrote range %s-%s/%s, length: %s, readable: %s, depth: %s"  %
                         (start_cur, self._cursor, self._file_size, self._cursor - start_cur, readable_bytes, self._depth))
                self._sent_bytes = True

            if self._cursor == self._stop_pos + 1:
                self.stopProducing()
                return defer.succeed(None)
            elif self._paused or self._stopped:
                return defer.succeed(None)
            else:
                self._deferred.addCallback(lambda _: threads.deferToThread(reactor.callLater, self._delay, _check_for_new_data))
                return defer.succeed(None)

        log.info("[" + str(datetime.now()) + "] Resuming producer")
        self._paused = False
        self._deferred.addCallback(lambda _: _check_for_new_data())

    def stopProducing(self):
        log.info("[" + str(datetime.now()) + "] Stopping producer")
        self._stopped = True
        # self._fileObject.close()
        self._deferred.addErrback(lambda err: err.trap(defer.CancelledError))
        self._deferred.addErrback(lambda err: err.trap(error.ConnectionDone))
        self._deferred.cancel()
        # self._request.finish()
        self._request.unregisterProducer()
        return defer.succeed(None)


class HostedLBRYFile(resource.Resource):
    def __init__(self, api):
        self._api = api
        self._producer = None
        resource.Resource.__init__(self)

    # todo: fix LBRYFileStreamer and use it instead of static.File
    # def makeProducer(self, request, stream):
    #     def _save_producer(producer):
    #         self._producer = producer
    #         return defer.succeed(None)
    #
    #     range_header = request.getAllHeaders()['range'].replace('bytes=', '').split('-')
    #     start, stop = int(range_header[0]), range_header[1]
    #     log.info("[" + str(datetime.now()) + "] GET range %s-%s" % (start, stop))
    #     path = os.path.join(self._api.download_directory, stream.file_name)
    #
    #     d = stream.get_total_bytes()
    #     d.addCallback(lambda size: _save_producer(LBRYFileStreamer(request, path, start, stop, size)))
    #     d.addCallback(lambda _: request.registerProducer(self._producer, streaming=True))
    #     # request.notifyFinish().addCallback(lambda _: self._producer.stopProducing())
    #     request.notifyFinish().addErrback(self._responseFailed, d)
    #     return d

    def render_GET(self, request):
        if 'name' in request.args.keys():
            if request.args['name'][0] != 'lbry' and request.args['name'][0] not in self._api.waiting_on.keys():
                d = self._api._download_name(request.args['name'][0])
                # d.addCallback(lambda stream: self.makeProducer(request, stream))
                d.addCallback(lambda stream: static.File(os.path.join(self._api.download_directory,
                                                                          stream.file_name)).render_GET(request))

            elif request.args['name'][0] in self._api.waiting_on.keys():
                request.redirect(UI_ADDRESS + "/?watch=" + request.args['name'][0])
                request.finish()
            else:
                request.redirect(UI_ADDRESS)
                request.finish()
            return server.NOT_DONE_YET

    # def _responseFailed(self, err, call):
    #     call.addErrback(lambda err: err.trap(error.ConnectionDone))
    #     call.addErrback(lambda err: err.trap(defer.CancelledError))
    #     call.addErrback(lambda err: log.info("Error: " + str(err)))
    #     call.cancel()


class LBRYDaemonServer(object):
    def _setup_server(self, wallet):
        self.root = LBRYindex(os.path.join(os.path.join(data_dir, "lbry-ui"), "active"))
        self._api = LBRYDaemon(self.root, wallet_type="lbrycrd")
        self.root.putChild("view", HostedLBRYFile(self._api))
        self.root.putChild(API_ADDRESS, self._api)
        return defer.succeed(True)

    def start(self, branch=DEFAULT_UI_BRANCH, user_specified=False, branch_specified=False, wallet=None):
        d = self._setup_server(wallet)
        d.addCallback(lambda _: self._api.setup(branch, user_specified, branch_specified))
        return d