forked from LBRYCommunity/lbry-sdk
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
|
import logging
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import mimetypes
|
||
|
|
|
||
|
|
from appdirs import user_data_dir
|
||
|
|
from zope.interface import implements
|
||
|
|
from twisted.internet import defer, error, interfaces, abstract, task, reactor
|
||
|
|
|
||
|
|
|
||
|
|
# TODO: omg, this code is essentially duplicated in Daemon
|
||
|
|
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)
|
||
|
|
|
||
|
|
log = logging.getLogger(__name__)
|
||
|
|
STATUS_FINISHED = 'finished'
|
||
|
|
|
||
|
|
class EncryptedFileStreamer(object):
|
||
|
|
"""
|
||
|
|
Writes LBRY stream to request; will pause to wait for new data if the file
|
||
|
|
is downloading.
|
||
|
|
|
||
|
|
No support for range requests (some browser players can't handle it when
|
||
|
|
the full video data isn't available on request).
|
||
|
|
"""
|
||
|
|
implements(interfaces.IPushProducer)
|
||
|
|
|
||
|
|
bufferSize = abstract.FileDescriptor.bufferSize
|
||
|
|
|
||
|
|
|
||
|
|
# How long to wait between sending blocks (needed because some
|
||
|
|
# video players freeze up if you try to send data too fast)
|
||
|
|
stream_interval = 0.005
|
||
|
|
|
||
|
|
# How long to wait before checking if new data has been appended to the file
|
||
|
|
new_data_check_interval = 0.25
|
||
|
|
|
||
|
|
|
||
|
|
def __init__(self, request, path, stream, file_manager):
|
||
|
|
def _set_content_length_header(length):
|
||
|
|
self._request.setHeader('content-length', length)
|
||
|
|
return defer.succeed(None)
|
||
|
|
|
||
|
|
self._request = request
|
||
|
|
self._file = open(path, 'rb')
|
||
|
|
self._stream = stream
|
||
|
|
self._file_manager = file_manager
|
||
|
|
self._headers_sent = False
|
||
|
|
|
||
|
|
self._running = True
|
||
|
|
|
||
|
|
self._request.setResponseCode(200)
|
||
|
|
self._request.setHeader('accept-ranges', 'none')
|
||
|
|
self._request.setHeader('content-type', mimetypes.guess_type(path)[0])
|
||
|
|
self._request.setHeader("Content-Security-Policy", "sandbox")
|
||
|
|
|
||
|
|
self._deferred = stream.get_total_bytes()
|
||
|
|
self._deferred.addCallback(_set_content_length_header)
|
||
|
|
self._deferred.addCallback(lambda _: self.resumeProducing())
|
||
|
|
|
||
|
|
def _check_for_new_data(self):
|
||
|
|
def _recurse_or_stop(stream_status):
|
||
|
|
if not self._running:
|
||
|
|
return
|
||
|
|
|
||
|
|
if stream_status != STATUS_FINISHED:
|
||
|
|
self._deferred.addCallback(lambda _: task.deferLater(reactor, self.new_data_check_interval, self._check_for_new_data))
|
||
|
|
else:
|
||
|
|
self.stopProducing()
|
||
|
|
|
||
|
|
if not self._running:
|
||
|
|
return
|
||
|
|
|
||
|
|
# Clear the file's EOF indicator by seeking to current position
|
||
|
|
self._file.seek(self._file.tell())
|
||
|
|
|
||
|
|
data = self._file.read(self.bufferSize)
|
||
|
|
if data:
|
||
|
|
self._request.write(data)
|
||
|
|
if self._running: # .write() can trigger a pause
|
||
|
|
self._deferred.addCallback(lambda _: task.deferLater(reactor, self.stream_interval, self._check_for_new_data))
|
||
|
|
else:
|
||
|
|
self._deferred.addCallback(lambda _: self._file_manager.get_lbry_file_status(self._stream))
|
||
|
|
self._deferred.addCallback(_recurse_or_stop)
|
||
|
|
|
||
|
|
def pauseProducing(self):
|
||
|
|
self._running = False
|
||
|
|
|
||
|
|
def resumeProducing(self):
|
||
|
|
self._running = True
|
||
|
|
self._check_for_new_data()
|
||
|
|
|
||
|
|
def stopProducing(self):
|
||
|
|
self._running = False
|
||
|
|
self._file.close()
|
||
|
|
self._deferred.addErrback(lambda err: err.trap(defer.CancelledError))
|
||
|
|
self._deferred.addErrback(lambda err: err.trap(error.ConnectionDone))
|
||
|
|
self._deferred.cancel()
|
||
|
|
self._request.unregisterProducer()
|
||
|
|
self._request.finish()
|