Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

stream manager component becomes file manager component

Browse source
This commit is contained in:
Victor Shyba 2020-01-27 19:02:31 -03:00
parent 543c75b293
commit 698ee271d6
6 changed files with 46 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
lbry/extras/daemon/components.py
View file

@ -28,6 +28,7 @@ try:
from lbry.torrent.session import TorrentSession from lbry.torrent.session import TorrentSession
except ImportError: except ImportError:
libtorrent = None libtorrent = None
TorrentSession = None
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -39,7 +40,7 @@ WALLET_COMPONENT = "wallet"
WALLET_SERVER_PAYMENTS_COMPONENT = "wallet_server_payments" WALLET_SERVER_PAYMENTS_COMPONENT = "wallet_server_payments"
DHT_COMPONENT = "dht" DHT_COMPONENT = "dht"
HASH_ANNOUNCER_COMPONENT = "hash_announcer" HASH_ANNOUNCER_COMPONENT = "hash_announcer"
STREAM_MANAGER_COMPONENT = "stream_manager" FILE_MANAGER_COMPONENT = "file_manager"
PEER_PROTOCOL_SERVER_COMPONENT = "peer_protocol_server" PEER_PROTOCOL_SERVER_COMPONENT = "peer_protocol_server"
UPNP_COMPONENT = "upnp" UPNP_COMPONENT = "upnp"
EXCHANGE_RATE_MANAGER_COMPONENT = "exchange_rate_manager" EXCHANGE_RATE_MANAGER_COMPONENT = "exchange_rate_manager"
@ -326,8 +327,8 @@ class HashAnnouncerComponent(Component):
} }
class StreamManagerComponent(Component): class FileManagerComponent(Component):
component_name = STREAM_MANAGER_COMPONENT component_name = FILE_MANAGER_COMPONENT
depends_on = [BLOB_COMPONENT, DATABASE_COMPONENT, WALLET_COMPONENT] depends_on = [BLOB_COMPONENT, DATABASE_COMPONENT, WALLET_COMPONENT]
def __init__(self, component_manager): def __init__(self, component_manager):

60
lbry/extras/daemon/daemon.py
View file

@ -40,7 +40,7 @@ from lbry.error import (
from lbry.extras import system_info from lbry.extras import system_info
from lbry.extras.daemon import analytics from lbry.extras.daemon import analytics
from lbry.extras.daemon.components import WALLET_COMPONENT, DATABASE_COMPONENT, DHT_COMPONENT, BLOB_COMPONENT from lbry.extras.daemon.components import WALLET_COMPONENT, DATABASE_COMPONENT, DHT_COMPONENT, BLOB_COMPONENT
from lbry.extras.daemon.components import STREAM_MANAGER_COMPONENT from lbry.extras.daemon.components import FILE_MANAGER_COMPONENT
from lbry.extras.daemon.components import EXCHANGE_RATE_MANAGER_COMPONENT, UPNP_COMPONENT from lbry.extras.daemon.components import EXCHANGE_RATE_MANAGER_COMPONENT, UPNP_COMPONENT
from lbry.extras.daemon.componentmanager import RequiredCondition from lbry.extras.daemon.componentmanager import RequiredCondition
from lbry.extras.daemon.componentmanager import ComponentManager from lbry.extras.daemon.componentmanager import ComponentManager
@ -372,8 +372,8 @@ class Daemon(metaclass=JSONRPCServerType):
return self.component_manager.get_component(DATABASE_COMPONENT) return self.component_manager.get_component(DATABASE_COMPONENT)
@property @property
def stream_manager(self) -> typing.Optional['StreamManager']: def file_manager(self) -> typing.Optional['StreamManager']:
return self.component_manager.get_component(STREAM_MANAGER_COMPONENT) return self.component_manager.get_component(FILE_MANAGER_COMPONENT)
@property @property
def exchange_rate_manager(self) -> typing.Optional['ExchangeRateManager']: def exchange_rate_manager(self) -> typing.Optional['ExchangeRateManager']:
@ -609,8 +609,8 @@ class Daemon(metaclass=JSONRPCServerType):
else: else:
name, claim_id = name_and_claim_id.split("/") name, claim_id = name_and_claim_id.split("/")
uri = f"lbry://{name}#{claim_id}" uri = f"lbry://{name}#{claim_id}"
if not self.stream_manager.started.is_set(): if not self.file_manager.started.is_set():
await self.stream_manager.started.wait() await self.file_manager.started.wait()
stream = await self.jsonrpc_get(uri) stream = await self.jsonrpc_get(uri)
if isinstance(stream, dict): if isinstance(stream, dict):
raise web.HTTPServerError(text=stream['error']) raise web.HTTPServerError(text=stream['error'])
@ -634,11 +634,11 @@ class Daemon(metaclass=JSONRPCServerType):
async def _handle_stream_range_request(self, request: web.Request): async def _handle_stream_range_request(self, request: web.Request):
sd_hash = request.path.split("/stream/")[1] sd_hash = request.path.split("/stream/")[1]
if not self.stream_manager.started.is_set(): if not self.file_manager.started.is_set():
await self.stream_manager.started.wait() await self.file_manager.started.wait()
if sd_hash not in self.stream_manager.streams: if sd_hash not in self.file_manager.streams:
return web.HTTPNotFound() return web.HTTPNotFound()
return await self.stream_manager.stream_partial_content(request, sd_hash) return await self.file_manager.stream_partial_content(request, sd_hash)
async def _process_rpc_call(self, data): async def _process_rpc_call(self, data):
args = data.get('params', {}) args = data.get('params', {})
@ -1077,7 +1077,7 @@ class Daemon(metaclass=JSONRPCServerType):
return results return results
@requires(WALLET_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT, @requires(WALLET_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT,
STREAM_MANAGER_COMPONENT) FILE_MANAGER_COMPONENT)
async def jsonrpc_get( async def jsonrpc_get(
self, uri, file_name=None, download_directory=None, timeout=None, save_file=None, wallet_id=None): self, uri, file_name=None, download_directory=None, timeout=None, save_file=None, wallet_id=None):
""" """
@ -1103,7 +1103,7 @@ class Daemon(metaclass=JSONRPCServerType):
if download_directory and not os.path.isdir(download_directory): if download_directory and not os.path.isdir(download_directory):
return {"error": f"specified download directory \"{download_directory}\" does not exist"} return {"error": f"specified download directory \"{download_directory}\" does not exist"}
try: try:
stream = await self.stream_manager.download_from_uri( stream = await self.file_manager.download_from_uri(
uri, self.exchange_rate_manager, timeout, file_name, download_directory, uri, self.exchange_rate_manager, timeout, file_name, download_directory,
save_file=save_file, wallet=wallet save_file=save_file, wallet=wallet
) )
@ -1949,7 +1949,7 @@ class Daemon(metaclass=JSONRPCServerType):
File management. File management.
""" """
@requires(STREAM_MANAGER_COMPONENT) @requires(FILE_MANAGER_COMPONENT)
async def jsonrpc_file_list(self, sort=None, reverse=False, comparison=None, wallet_id=None, page=None, async def jsonrpc_file_list(self, sort=None, reverse=False, comparison=None, wallet_id=None, page=None,
page_size=None, **kwargs): page_size=None, **kwargs):
""" """
@ -1994,7 +1994,7 @@ class Daemon(metaclass=JSONRPCServerType):
comparison = comparison or 'eq' comparison = comparison or 'eq'
paginated = paginate_list( paginated = paginate_list(
self.stream_manager.get_filtered_streams(sort, reverse, comparison, **kwargs), page, page_size self.file_manager.get_filtered_streams(sort, reverse, comparison, **kwargs), page, page_size
) )
if paginated['items']: if paginated['items']:
receipts = { receipts = {
@ -2008,7 +2008,7 @@ class Daemon(metaclass=JSONRPCServerType):
stream.purchase_receipt = receipts.get(stream.claim_id) stream.purchase_receipt = receipts.get(stream.claim_id)
return paginated return paginated
@requires(STREAM_MANAGER_COMPONENT) @requires(FILE_MANAGER_COMPONENT)
async def jsonrpc_file_set_status(self, status, **kwargs): async def jsonrpc_file_set_status(self, status, **kwargs):
""" """
Start or stop downloading a file Start or stop downloading a file
@ -2032,12 +2032,12 @@ class Daemon(metaclass=JSONRPCServerType):
if status not in ['start', 'stop']: if status not in ['start', 'stop']:
raise Exception('Status must be "start" or "stop".') raise Exception('Status must be "start" or "stop".')
streams = self.stream_manager.get_filtered_streams(**kwargs) streams = self.file_manager.get_filtered_streams(**kwargs)
if not streams: if not streams:
raise Exception(f'Unable to find a file for {kwargs}') raise Exception(f'Unable to find a file for {kwargs}')
stream = streams[0] stream = streams[0]
if status == 'start' and not stream.running: if status == 'start' and not stream.running:
await stream.save_file(node=self.stream_manager.node) await stream.save_file(node=self.file_manager.node)
msg = "Resumed download" msg = "Resumed download"
elif status == 'stop' and stream.running: elif status == 'stop' and stream.running:
await stream.stop() await stream.stop()
@ -2049,7 +2049,7 @@ class Daemon(metaclass=JSONRPCServerType):
) )
return msg return msg
@requires(STREAM_MANAGER_COMPONENT) @requires(FILE_MANAGER_COMPONENT)
async def jsonrpc_file_delete(self, delete_from_download_dir=False, delete_all=False, **kwargs): async def jsonrpc_file_delete(self, delete_from_download_dir=False, delete_all=False, **kwargs):
""" """
Delete a LBRY file Delete a LBRY file
@ -2081,7 +2081,7 @@ class Daemon(metaclass=JSONRPCServerType):
(bool) true if deletion was successful (bool) true if deletion was successful
""" """
streams = self.stream_manager.get_filtered_streams(**kwargs) streams = self.file_manager.get_filtered_streams(**kwargs)
if len(streams) > 1: if len(streams) > 1:
if not delete_all: if not delete_all:
@ -2098,12 +2098,12 @@ class Daemon(metaclass=JSONRPCServerType):
else: else:
for stream in streams: for stream in streams:
message = f"Deleted file {stream.file_name}" message = f"Deleted file {stream.file_name}"
await self.stream_manager.delete_stream(stream, delete_file=delete_from_download_dir) await self.file_manager.delete_stream(stream, delete_file=delete_from_download_dir)
log.info(message) log.info(message)
result = True result = True
return result return result
@requires(STREAM_MANAGER_COMPONENT) @requires(FILE_MANAGER_COMPONENT)
async def jsonrpc_file_save(self, file_name=None, download_directory=None, **kwargs): async def jsonrpc_file_save(self, file_name=None, download_directory=None, **kwargs):
""" """
Start saving a file to disk. Start saving a file to disk.
@ -2130,7 +2130,7 @@ class Daemon(metaclass=JSONRPCServerType):
Returns: {File} Returns: {File}
""" """
streams = self.stream_manager.get_filtered_streams(**kwargs) streams = self.file_manager.get_filtered_streams(**kwargs)
if len(streams) > 1: if len(streams) > 1:
log.warning("There are %i matching files, use narrower filters to select one", len(streams)) log.warning("There are %i matching files, use narrower filters to select one", len(streams))
@ -2905,7 +2905,7 @@ class Daemon(metaclass=JSONRPCServerType):
Create, update, abandon, list and inspect your stream claims. Create, update, abandon, list and inspect your stream claims.
""" """
@requires(WALLET_COMPONENT, STREAM_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT) @requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT)
async def jsonrpc_publish(self, name, **kwargs): async def jsonrpc_publish(self, name, **kwargs):
""" """
Create or replace a stream claim at a given name (use 'stream create/update' for more control). Create or replace a stream claim at a given name (use 'stream create/update' for more control).
@ -3027,7 +3027,7 @@ class Daemon(metaclass=JSONRPCServerType):
f"to update a specific stream claim." f"to update a specific stream claim."
) )
@requires(WALLET_COMPONENT, STREAM_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT) @requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT)
async def jsonrpc_stream_repost(self, name, bid, claim_id, allow_duplicate_name=False, channel_id=None, async def jsonrpc_stream_repost(self, name, bid, claim_id, allow_duplicate_name=False, channel_id=None,
channel_name=None, channel_account_id=None, account_id=None, wallet_id=None, channel_name=None, channel_account_id=None, account_id=None, wallet_id=None,
claim_address=None, funding_account_ids=None, preview=False, blocking=False): claim_address=None, funding_account_ids=None, preview=False, blocking=False):
@ -3099,7 +3099,7 @@ class Daemon(metaclass=JSONRPCServerType):
return tx return tx
@requires(WALLET_COMPONENT, STREAM_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT) @requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT)
async def jsonrpc_stream_create( async def jsonrpc_stream_create(
self, name, bid, file_path, allow_duplicate_name=False, self, name, bid, file_path, allow_duplicate_name=False,
channel_id=None, channel_name=None, channel_account_id=None, channel_id=None, channel_name=None, channel_account_id=None,
@ -3237,7 +3237,7 @@ class Daemon(metaclass=JSONRPCServerType):
file_stream = None file_stream = None
if not preview: if not preview:
file_stream = await self.stream_manager.create_stream(file_path) file_stream = await self.file_manager.create_stream(file_path)
claim.stream.source.sd_hash = file_stream.sd_hash claim.stream.source.sd_hash = file_stream.sd_hash
new_txo.script.generate() new_txo.script.generate()
@ -3257,7 +3257,7 @@ class Daemon(metaclass=JSONRPCServerType):
return tx return tx
@requires(WALLET_COMPONENT, STREAM_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT) @requires(WALLET_COMPONENT, FILE_MANAGER_COMPONENT, BLOB_COMPONENT, DATABASE_COMPONENT)
async def jsonrpc_stream_update( async def jsonrpc_stream_update(
self, claim_id, bid=None, file_path=None, self, claim_id, bid=None, file_path=None,
channel_id=None, channel_name=None, channel_account_id=None, clear_channel=False, channel_id=None, channel_name=None, channel_account_id=None, clear_channel=False,
@ -4583,9 +4583,9 @@ class Daemon(metaclass=JSONRPCServerType):
""" """
if not blob_hash or not is_valid_blobhash(blob_hash): if not blob_hash or not is_valid_blobhash(blob_hash):
return f"Invalid blob hash to delete '{blob_hash}'" return f"Invalid blob hash to delete '{blob_hash}'"
streams = self.stream_manager.get_filtered_streams(sd_hash=blob_hash) streams = self.file_manager.get_filtered_streams(sd_hash=blob_hash)
if streams: if streams:
await self.stream_manager.delete_stream(streams[0]) await self.file_manager.delete_stream(streams[0])
else: else:
await self.blob_manager.delete_blobs([blob_hash]) await self.blob_manager.delete_blobs([blob_hash])
return "Deleted %s" % blob_hash return "Deleted %s" % blob_hash
@ -4758,7 +4758,7 @@ class Daemon(metaclass=JSONRPCServerType):
raise NotImplementedError() raise NotImplementedError()
@requires(STREAM_MANAGER_COMPONENT) @requires(FILE_MANAGER_COMPONENT)
async def jsonrpc_file_reflect(self, **kwargs): async def jsonrpc_file_reflect(self, **kwargs):
""" """
Reflect all the blobs in a file matching the filter criteria Reflect all the blobs in a file matching the filter criteria
@ -5334,7 +5334,7 @@ class Daemon(metaclass=JSONRPCServerType):
results = await self.ledger.resolve(accounts, urls, **kwargs) results = await self.ledger.resolve(accounts, urls, **kwargs)
if self.conf.save_resolved_claims and results: if self.conf.save_resolved_claims and results:
try: try:
claims = self.stream_manager._convert_to_old_resolve_output(self.wallet_manager, results) claims = self.file_manager._convert_to_old_resolve_output(self.wallet_manager, results)
await self.storage.save_claims_for_resolve([ await self.storage.save_claims_for_resolve([
value for value in claims.values() if 'error' not in value value for value in claims.values() if 'error' not in value
]) ])

12
tests/integration/datanetwork/test_file_commands.py
View file

@ -228,11 +228,11 @@ class FileCommands(CommandTestCase):
await self.daemon.jsonrpc_get('lbry://foo') await self.daemon.jsonrpc_get('lbry://foo')
with open(original_path, 'wb') as handle: with open(original_path, 'wb') as handle:
handle.write(b'some other stuff was there instead') handle.write(b'some other stuff was there instead')
self.daemon.stream_manager.stop() self.daemon.file_manager.stop()
await self.daemon.stream_manager.start() await self.daemon.file_manager.start()
await asyncio.wait_for(self.wait_files_to_complete(), timeout=5) # if this hangs, file didn't get set completed await asyncio.wait_for(self.wait_files_to_complete(), timeout=5) # if this hangs, file didn't get set completed
# check that internal state got through up to the file list API # check that internal state got through up to the file list API
stream = self.daemon.stream_manager.get_stream_by_stream_hash(file_info['stream_hash']) stream = self.daemon.file_manager.get_stream_by_stream_hash(file_info['stream_hash'])
file_info = (await self.file_list())[0] file_info = (await self.file_list())[0]
self.assertEqual(stream.file_name, file_info['file_name']) self.assertEqual(stream.file_name, file_info['file_name'])
# checks if what the API shows is what he have at the very internal level. # checks if what the API shows is what he have at the very internal level.
@ -255,7 +255,7 @@ class FileCommands(CommandTestCase):
resp = await self.out(self.daemon.jsonrpc_get('lbry://foo', timeout=2)) resp = await self.out(self.daemon.jsonrpc_get('lbry://foo', timeout=2))
self.assertNotIn('error', resp) self.assertNotIn('error', resp)
self.assertTrue(os.path.isfile(path)) self.assertTrue(os.path.isfile(path))
self.daemon.stream_manager.stop() self.daemon.file_manager.stop()
await asyncio.sleep(0.01, loop=self.loop) # FIXME: this sleep should not be needed await asyncio.sleep(0.01, loop=self.loop) # FIXME: this sleep should not be needed
self.assertFalse(os.path.isfile(path)) self.assertFalse(os.path.isfile(path))
@ -348,8 +348,8 @@ class FileCommands(CommandTestCase):
# restart the daemon and make sure the fee is still there # restart the daemon and make sure the fee is still there
self.daemon.stream_manager.stop() self.daemon.file_manager.stop()
await self.daemon.stream_manager.start() await self.daemon.file_manager.start()
self.assertItemCount(await self.daemon.jsonrpc_file_list(), 1) self.assertItemCount(await self.daemon.jsonrpc_file_list(), 1)
self.assertEqual((await self.daemon.jsonrpc_file_list())['items'][0].content_fee.raw, raw_content_fee) self.assertEqual((await self.daemon.jsonrpc_file_list())['items'][0].content_fee.raw, raw_content_fee)
await self.daemon.jsonrpc_file_delete(claim_name='icanpay') await self.daemon.jsonrpc_file_delete(claim_name='icanpay')

4
tests/integration/datanetwork/test_streaming.py
View file

@ -21,8 +21,8 @@ def get_random_bytes(n: int) -> bytes:
class RangeRequests(CommandTestCase): class RangeRequests(CommandTestCase):
async def _restart_stream_manager(self): async def _restart_stream_manager(self):
self.daemon.stream_manager.stop() self.daemon.file_manager.stop()
await self.daemon.stream_manager.start() await self.daemon.file_manager.start()
return return
async def _setup_stream(self, data: bytes, save_blobs: bool = True, save_files: bool = False, file_size=0): async def _setup_stream(self, data: bytes, save_blobs: bool = True, save_files: bool = False, file_size=0):

6
tests/integration/other/test_cli.py
View file

@ -6,7 +6,7 @@ from lbry.conf import Config
from lbry.extras import cli from lbry.extras import cli
from lbry.extras.daemon.components import ( from lbry.extras.daemon.components import (
DATABASE_COMPONENT, BLOB_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, DATABASE_COMPONENT, BLOB_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT,
HASH_ANNOUNCER_COMPONENT, STREAM_MANAGER_COMPONENT, PEER_PROTOCOL_SERVER_COMPONENT, HASH_ANNOUNCER_COMPONENT, FILE_MANAGER_COMPONENT, PEER_PROTOCOL_SERVER_COMPONENT,
UPNP_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, WALLET_SERVER_PAYMENTS_COMPONENT UPNP_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, WALLET_SERVER_PAYMENTS_COMPONENT
) )
from lbry.extras.daemon.daemon import Daemon from lbry.extras.daemon.daemon import Daemon
@ -21,7 +21,7 @@ class CLIIntegrationTest(AsyncioTestCase):
conf.api = 'localhost:5299' conf.api = 'localhost:5299'
conf.components_to_skip = ( conf.components_to_skip = (
DATABASE_COMPONENT, BLOB_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT, DATABASE_COMPONENT, BLOB_COMPONENT, WALLET_COMPONENT, DHT_COMPONENT,
HASH_ANNOUNCER_COMPONENT, STREAM_MANAGER_COMPONENT, PEER_PROTOCOL_SERVER_COMPONENT, HASH_ANNOUNCER_COMPONENT, FILE_MANAGER_COMPONENT, PEER_PROTOCOL_SERVER_COMPONENT,
UPNP_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, WALLET_SERVER_PAYMENTS_COMPONENT UPNP_COMPONENT, EXCHANGE_RATE_MANAGER_COMPONENT, WALLET_SERVER_PAYMENTS_COMPONENT
) )
Daemon.component_attributes = {} Daemon.component_attributes = {}
@ -34,4 +34,4 @@ class CLIIntegrationTest(AsyncioTestCase):
with contextlib.redirect_stdout(actual_output): with contextlib.redirect_stdout(actual_output):
cli.main(["--api", "localhost:5299", "status"]) cli.main(["--api", "localhost:5299", "status"])
actual_output = actual_output.getvalue() actual_output = actual_output.getvalue()
self.assertIn("connection_status", actual_output) self.assertIn("connection_status", actual_output)

2
tests/unit/components/test_component_manager.py
View file

@ -26,7 +26,7 @@ class TestComponentManager(AsyncioTestCase):
[ [
components.HashAnnouncerComponent, components.HashAnnouncerComponent,
components.PeerProtocolServerComponent, components.PeerProtocolServerComponent,
components.StreamManagerComponent, components.FileManagerComponent,
components.WalletServerPaymentsComponent components.WalletServerPaymentsComponent
] ]
] ]