LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases 282 Packages Wiki Activity

fix setting finished event

Browse source
This commit is contained in:
Jack Robison 2019-02-06 12:18:24 -05:00
parent f22b5da170
commit ae11c5bb4b
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 6 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
lbrynet/stream/assembler.py
View file

@ -109,12 +109,13 @@ class StreamAssembler:
self.descriptor.sd_hash) self.descriptor.sd_hash)
continue continue
finally: finally:
if written_blobs == len(self.descriptor.blobs) - 1: if written_blobs == len(self.descriptor.blobs) - 2:
log.debug("finished decrypting and assembling stream") log.debug("finished decrypting and assembling stream")
self.stream_finished_event.set() self.stream_finished_event.set()
await self.after_finished() await self.after_finished()
else: else:
log.debug("stream decryption and assembly did not finish") log.debug("stream decryption and assembly did not finish (%i/%i blobs are done)", written_blobs,
len(self.descriptor.blobs) - 2)
async def get_blob(self, blob_hash: str, length: typing.Optional[int] = None) -> 'BlobFile': async def get_blob(self, blob_hash: str, length: typing.Optional[int] = None) -> 'BlobFile':
return self.blob_manager.get_blob(blob_hash, length) return self.blob_manager.get_blob(blob_hash, length)

5
tests/unit/stream/test_downloader.py
View file

@ -37,15 +37,16 @@ class TestStreamDownloader(BlobExchangeTestBase):
return q2, self.loop.create_task(_task()) return q2, self.loop.create_task(_task())
mock_node.accumulate_peers = mock_accumulate_peers or _mock_accumulate_peers mock_node.accumulate_peers = mock_accumulate_peers or _mock_accumulate_peers
self.downloader.download(mock_node) self.downloader.download(mock_node)
await self.downloader.stream_finished_event.wait() await self.downloader.stream_finished_event.wait()
self.assertTrue(self.downloader.stream_handle.closed)
self.assertTrue(os.path.isfile(self.downloader.output_path))
self.downloader.stop() self.downloader.stop()
self.assertIs(self.downloader.stream_handle, None)
self.assertTrue(os.path.isfile(self.downloader.output_path)) self.assertTrue(os.path.isfile(self.downloader.output_path))
with open(self.downloader.output_path, 'rb') as f: with open(self.downloader.output_path, 'rb') as f:
self.assertEqual(f.read(), self.stream_bytes) self.assertEqual(f.read(), self.stream_bytes)
await asyncio.sleep(0.01) await asyncio.sleep(0.01)
self.assertTrue(self.downloader.stream_handle.closed)
async def test_transfer_stream(self): async def test_transfer_stream(self):
await self._test_transfer_stream(10) await self._test_transfer_stream(10)