Repository: streamlink/streamlink
Pull request #3882 (2021-07-24 06:21:54)

plugins.nimotv: use 'mStreamPkg'
using regex seems easier than https://github.com/streamlink/streamlink/issues/3875#issuecomment-882002731

might break at some point, but should work for now

closes https://github.com/streamlink/streamlink/issues/3875

---

@Trung0246 could you test it


Patch:
diff --git a/src/streamlink/buffers.py b/src/streamlink/buffers.py
index af56b9b4..f98d8fac 100644
--- a/src/streamlink/buffers.py
+++ b/src/streamlink/buffers.py
@@ -103,10 +103,7 @@ class RingBuffer(Buffer):
 
     def read(self, size=-1, block=True, timeout=None):
         if block and not self.closed:
-            self.event_used.wait(timeout)
-
-            # If the event is still not set it's a timeout
-            if not self.event_used.is_set() and self.length == 0:
+            if not self.event_used.wait(timeout) and self.length == 0:
                 raise OSError("Read timeout")
 
         return self._read(size)
diff --git a/src/streamlink/plugins/nimotv.py b/src/streamlink/plugins/nimotv.py
index a7123f9c..a7681bbe 100644
--- a/src/streamlink/plugins/nimotv.py
+++ b/src/streamlink/plugins/nimotv.py
@@ -10,7 +10,7 @@ log = logging.getLogger(__name__)
 
 
 @pluginmatcher(re.compile(
-    r'https?://(?:www\.)?nimo\.tv/(?P<username>.*)'
+    r'https?://(?:www\.|m\.)?nimo\.tv/(?P<username>.*)'
 ))
 class NimoTV(Plugin):
     data_url = 'https://m.nimo.tv/{0}'
@@ -28,16 +28,8 @@ class NimoTV(Plugin):
                 'title': str,
                 'nickname': str,
                 'game': str,
-                'roomLineInfo': validate.any(None, {
-                    'vCodeLines2': [{
-                        'iBitRate': int,
-                        'vCdns': [{
-                            'vCdnUrls': [{
-                                'smediaUrl': validate.url(),
-                            }],
-                        }],
-                    }],
-                }),
+                'liveStreamStatus': int,
+                validate.optional('mStreamPkg'): str,
             },
         )),
     )
@@ -46,21 +38,23 @@ class NimoTV(Plugin):
         250: '240p',
         500: '360p',
         1000: '480p',
-        2000: '720p',
+        2500: '720p',
         6000: '1080p',
     }
 
+    _re_appid = re.compile(br'appid=(\d+)')
+    _re_domain = re.compile(br'(https?:\/\/[A-Za-z]{2,3}.hls[A-Za-z\.\/]+)(?:V|&)')
+    _re_id = re.compile(br'id=([^|\\]+)')
+    _re_tp = re.compile(br'tp=(\d+)')
+
     def get_author(self):
-        if self.author is not None:
-            return self.author
+        return self.author
 
     def get_category(self):
-        if self.category is not None:
-            return self.category
+        return self.category
 
     def get_title(self):
-        if self.title is not None:
-            return self.title
+        return self.title
 
     def _get_streams(self):
         username = self.match.group('username')
@@ -74,20 +68,50 @@ class NimoTV(Plugin):
             schema=self.data_schema,
         )
 
-        if data is None or data['roomLineInfo'] is None:
+        if data['liveStreamStatus'] == 0:
+            log.info('This stream is currently offline')
+            return
+
+        mStreamPkg = data.get('mStreamPkg')
+        if not mStreamPkg:
+            log.debug('missing mStreamPkg')
             return
 
+        mStreamPkg = bytes.fromhex(mStreamPkg)
+        try:
+            _appid = self._re_appid.search(mStreamPkg).group(1).decode('utf-8')
+            _domain = self._re_domain.search(mStreamPkg).group(1).decode('utf-8')
+            _id = self._re_id.search(mStreamPkg).group(1).decode('utf-8')
+            _tp = self._re_tp.search(mStreamPkg).group(1).decode('utf-8')
+        except AttributeError:
+            log.error('invalid mStreamPkg')
+            return
+
+        params = {
+            'appid': _appid,
+            'id': _id,
+            'tp': _tp,
+            'u': '0',
+            't': '100',
+            'needwm': 1,
+        }
+        url = f'{_domain}{_id}.m3u8'
+        log.debug(f'URL={url}')
+        for k, v in self.video_qualities.items():
+            _params = params.copy()
+            _params.update({'ratio': k})
+            if v == '1080p':
+                _params.update({'needwm': 0})
+            elif v in ('720p', '480p', '360p'):
+                _params.update({'sphd': 1})
+
+            log.trace(f'{v} params={_params!r}')
+            # some qualities might not exist, but it will select a different lower quality
+            yield v, HLSStream(self.session, url, params=_params)
+
         self.author = data['nickname']
         self.category = data['game']
         self.title = data['title']
 
-        for vcl2 in data['roomLineInfo']['vCodeLines2']:
-            q = self.video_qualities[vcl2['iBitRate']]
-            for vcdn in vcl2['vCdns']:
-                for vcdnurl in vcdn['vCdnUrls']:
-                    if 'tx.hls.nimo.tv' in vcdnurl['smediaUrl']:
-                        log.debug(f"HLS URL={vcdnurl['smediaUrl']} ({q})")
-                        yield q, HLSStream(self.session, vcdnurl['smediaUrl'])
-
 
 __plugin__ = NimoTV
diff --git a/src/streamlink/plugins/tv4play.py b/src/streamlink/plugins/tv4play.py
index 6aa176d1..d53fc39f 100644
--- a/src/streamlink/plugins/tv4play.py
+++ b/src/streamlink/plugins/tv4play.py
@@ -12,7 +12,7 @@ log = logging.getLogger(__name__)
 @pluginmatcher(re.compile(r"""
     https?://(?:www\.)?
     (?:
-        tv4play\.se/program/[^?/]+
+        tv4play\.se/program/[^?/]+/[^?/]+
         |
         fotbollskanalen\.se/video
     )
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
index 5c25bc92..2300a376 100644
--- a/src/streamlink/plugins/twitch.py
+++ b/src/streamlink/plugins/twitch.py
@@ -118,6 +118,16 @@ class TwitchHLSStreamReader(HLSStreamReader):
     __worker__ = TwitchHLSStreamWorker
     __writer__ = TwitchHLSStreamWriter
 
+    def __init__(self, stream):
+        if stream.disable_ads:
+            log.info("Will skip ad segments")
+        if stream.low_latency:
+            live_edge = max(1, min(LOW_LATENCY_MAX_LIVE_EDGE, stream.session.options.get("hls-live-edge")))
+            stream.session.options.set("hls-live-edge", live_edge)
+            stream.session.options.set("hls-segment-stream-data", True)
+            log.info(f"Low latency streaming (HLS live edge: {live_edge})")
+        super().__init__(stream)
+
 
 class TwitchHLSStream(HLSStream):
     __reader__ = TwitchHLSStreamReader
@@ -127,17 +137,6 @@ class TwitchHLSStream(HLSStream):
         self.disable_ads = self.session.get_plugin_option("twitch", "disable-ads")
         self.low_latency = self.session.get_plugin_option("twitch", "low-latency")
 
-    def open(self):
-        if self.disable_ads:
-            log.info("Will skip ad segments")
-        if self.low_latency:
-            live_edge = max(1, min(LOW_LATENCY_MAX_LIVE_EDGE, self.session.options.get("hls-live-edge")))
-            self.session.options.set("hls-live-edge", live_edge)
-            self.session.options.set("hls-segment-stream-data", True)
-            log.info("Low latency streaming (HLS live edge: {0})".format(live_edge))
-
-        return super().open()
-
     @classmethod
     def _get_variant_playlist(cls, res):
         return load_hls_playlist(res.text, base_uri=res.url)
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index a1ed1665..e6babb9c 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -123,11 +123,11 @@ class HLSStreamWriter(SegmentedStreamWriter):
                 if future is None or sequence.segment.discontinuity:
                     future = self.executor.submit(self.fetch_map, sequence)
                     self.map_cache.set(sequence.segment.map.uri, future)
-                self.queue(future, sequence, True)
+                self.queue(sequence, future, True)
 
             # regular segment request
             future = self.executor.submit(self.fetch, sequence)
-            self.queue(future, sequence, False)
+            self.queue(sequence, future, False)
 
     def fetch(self, sequence: Sequence) -> Optional[Response]:
         try:
@@ -231,25 +231,6 @@ class HLSStreamWorker(SegmentedStreamWorker):
         elif self.playlist_reload_time_override not in ["segment", "live-edge"]:
             self.playlist_reload_time_override = 0
 
-        self.reload_playlist()
-
-        if self.playlist_end is None:
-            if self.duration_offset_start > 0:
-                log.debug(f"Time offsets negative for live streams, skipping back {self.duration_offset_start} seconds")
-            # live playlist, force offset durations back to None
-            self.duration_offset_start = -self.duration_offset_start
-
-        if self.duration_offset_start != 0:
-            self.playlist_sequence = self.duration_to_sequence(self.duration_offset_start, self.playlist_sequences)
-
-        if self.playlist_sequences:
-            log.debug(f"First Sequence: {self.playlist_sequences[0].num}; "
-                      f"Last Sequence: {self.playlist_sequences[-1].num}")
-            log.debug(f"Start offset: {self.duration_offset_start}; "
-                      f"Duration: {self.duration_limit}; "
-                      f"Start Sequence: {self.playlist_sequence}; "
-                      f"End Sequence: {self.playlist_end}")
-
     def _reload_playlist(self, text, url):
         return hls_playlist.load(text, url)
 
@@ -289,7 +270,7 @@ class HLSStreamWorker(SegmentedStreamWorker):
             return sequences[-1].segment.duration
         if self.playlist_reload_time_override == "live-edge" and sequences:
             return sum([s.segment.duration for s in sequences[-max(1, self.live_edge - 1):]])
-        if self.playlist_reload_time_override > 0:
+        if type(self.playlist_reload_time_override) is float and self.playlist_reload_time_override > 0:
             return self.playlist_reload_time_override
         if playlist.target_duration:
             return playlist.target_duration
@@ -341,6 +322,25 @@ class HLSStreamWorker(SegmentedStreamWorker):
         return default
 
     def iter_segments(self):
+        self.reload_playlist()
+
+        if self.playlist_end is None:
+            if self.duration_offset_start > 0:
+                log.debug(f"Time offsets negative for live streams, skipping back {self.duration_offset_start} seconds")
+            # live playlist, force offset durations back to None
+            self.duration_offset_start = -self.duration_offset_start
+
+        if self.duration_offset_start != 0:
+            self.playlist_sequence = self.duration_to_sequence(self.duration_offset_start, self.playlist_sequences)
+
+        if self.playlist_sequences:
+            log.debug(f"First Sequence: {self.playlist_sequences[0].num}; "
+                      f"Last Sequence: {self.playlist_sequences[-1].num}")
+            log.debug(f"Start offset: {self.duration_offset_start}; "
+                      f"Duration: {self.duration_limit}; "
+                      f"Start Sequence: {self.playlist_sequence}; "
+                      f"End Sequence: {self.playlist_end}")
+
         total_duration = 0
         while not self.closed:
             for sequence in filter(self.valid_sequence, self.playlist_sequences):
@@ -369,20 +369,21 @@ class HLSStreamReader(SegmentedStreamReader):
     __worker__ = HLSStreamWorker
     __writer__ = HLSStreamWriter
 
-    def __init__(self, stream, *args, **kwargs):
-        super().__init__(stream, *args, **kwargs)
+    def __init__(self, stream):
         self.request_params = dict(stream.args)
-        self.timeout = stream.session.options.get("hls-timeout")
-
-        self.filter_event = Event()
-        self.filter_event.set()
-
         # These params are reserved for internal use
         self.request_params.pop("exception", None)
         self.request_params.pop("stream", None)
         self.request_params.pop("timeout", None)
         self.request_params.pop("url", None)
 
+        self.filter_event = Event()
+        self.filter_event.set()
+
+        timeout = stream.session.options.get("hls-timeout")
+
+        super().__init__(stream, timeout)
+
     def read(self, size):
         while True:
             try:
diff --git a/src/streamlink/stream/segmented.py b/src/streamlink/stream/segmented.py
index 4812dd03..51e1de11 100644
--- a/src/streamlink/stream/segmented.py
+++ b/src/streamlink/stream/segmented.py
@@ -48,20 +48,21 @@ class SegmentedStreamWorker(Thread):
         self.reader = reader
         self.writer = reader.writer
         self.stream = reader.stream
-        self.session = reader.stream.session
+        self.session = reader.session
 
-        self._wait = None
+        self._wait = Event()
 
         super().__init__(daemon=True, name=f"Thread-{self.__class__.__name__}")
 
     def close(self):
         """Shuts down the thread."""
-        if not self.closed:
-            log.debug("Closing worker thread")
+        if self.closed:  # pragma: no cover
+            return
+
+        log.debug("Closing worker thread")
 
         self.closed = True
-        if self._wait:
-            self._wait.set()
+        self._wait.set()
 
     def wait(self, time):
         """Pauses the thread for a specified time.
@@ -69,7 +70,6 @@ class SegmentedStreamWorker(Thread):
         Returns False if interrupted by another thread and True if the
         time runs out normally.
         """
-        self._wait = Event()
         return not self._wait.wait(time)
 
     def iter_segments(self):
@@ -82,7 +82,7 @@ class SegmentedStreamWorker(Thread):
 
     def run(self):
         for segment in self.iter_segments():
-            if self.closed:
+            if self.closed:  # pragma: no cover
                 break
             self.writer.put(segment)
 
@@ -102,7 +102,7 @@ class SegmentedStreamWriter(Thread):
         self.closed = False
         self.reader = reader
         self.stream = reader.stream
-        self.session = reader.stream.session
+        self.session = reader.session
 
         if not retries:
             retries = self.session.options.get("stream-segment-attempts")
@@ -121,17 +121,19 @@ class SegmentedStreamWriter(Thread):
         super().__init__(daemon=True, name=f"Thread-{self.__class__.__name__}")
 
     def close(self):
-        """Shuts down the thread."""
-        if not self.closed:
-            log.debug("Closing writer thread")
+        """Shuts down the thread, its executor and closes the reader (worker thread and buffer)."""
+        if self.closed:  # pragma: no cover
+            return
+
+        log.debug("Closing writer thread")
 
         self.closed = True
-        self.reader.buffer.close()
+        self.reader.close()
         self.executor.shutdown(wait=True, cancel_futures=True)
 
     def put(self, segment):
         """Adds a segment to the download pool and write queue."""
-        if self.closed:
+        if self.closed:  # pragma: no cover
             return
 
         if segment is None:
@@ -139,17 +141,27 @@ class SegmentedStreamWriter(Thread):
         else:
             future = self.executor.submit(self.fetch, segment, retries=self.retries)
 
-        self.queue(future, segment)
+        self.queue(segment, future)
 
-    def queue(self, future: Optional[Future], segment: Any, *data):
+    def queue(self, segment: Any, future: Optional[Future], *data):
         """Puts values into a queue but aborts if this thread is closed."""
-        while not self.closed:
+        while not self.closed:  # pragma: no branch
             try:
-                self.futures.put((future, segment, *data), block=True, timeout=1)
+                self._futures_put((segment, future, *data))
                 return
             except queue.Full:  # pragma: no cover
                 continue
 
+    def _futures_put(self, item):
+        self.futures.put(item, block=True, timeout=1)
+
+    def _futures_get(self):
+        return self.futures.get(block=True, timeout=0.5)
+
+    @staticmethod
+    def _future_result(future: Future):
+        return future.result(timeout=0.5)
+
     def fetch(self, segment):
         """Fetches a segment.
 
@@ -167,7 +179,7 @@ class SegmentedStreamWriter(Thread):
     def run(self):
         while not self.closed:
             try:
-                future, segment, *data = self.futures.get(block=True, timeout=0.5)
+                segment, future, *data = self._futures_get()
             except queue.Empty:  # pragma: no cover
                 continue
 
@@ -177,7 +189,7 @@ class SegmentedStreamWriter(Thread):
 
             while not self.closed:  # pragma: no branch
                 try:
-                    result = future.result(timeout=0.5)
+                    result = self._future_result(future)
                 except futures.TimeoutError:  # pragma: no cover
                     continue
                 except futures.CancelledError:  # pragma: no cover
@@ -199,18 +211,14 @@ class SegmentedStreamReader(StreamIO):
         super().__init__()
         self.session = stream.session
         self.stream = stream
+        self.timeout = timeout or self.session.options.get("stream-timeout")
 
-        if not timeout:
-            timeout = self.session.options.get("stream-timeout")
-
-        self.timeout = timeout
-
-    def open(self):
         buffer_size = self.session.get_option("ringbuffer-size")
         self.buffer = RingBuffer(buffer_size)
         self.writer = self.__writer__(self)
         self.worker = self.__worker__(self)
 
+    def open(self):
         self.writer.start()
         self.worker.start()
 
@@ -220,8 +228,8 @@ class SegmentedStreamReader(StreamIO):
         self.buffer.close()
 
     def read(self, size):
-        if not self.buffer:
-            return b""
-
-        return self.buffer.read(size, block=self.writer.is_alive(),
-                                timeout=self.timeout)
+        return self.buffer.read(
+            size,
+            block=self.writer.is_alive(),
+            timeout=self.timeout
+        )


Test patch:
diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py
index 55f29b26..364939d4 100644
--- a/tests/mixins/stream_hls.py
+++ b/tests/mixins/stream_hls.py
@@ -10,6 +10,11 @@ from streamlink import Streamlink
 from streamlink.stream.hls import HLSStream, HLSStreamWriter as _HLSStreamWriter
 
 
+TIMEOUT_AWAIT_READ = 5
+TIMEOUT_AWAIT_READ_ONCE = 5
+TIMEOUT_AWAIT_WRITE = 5
+
+
 class HLSItemBase:
     path = ""
 
@@ -89,6 +94,16 @@ class EventedHLSStreamWriter(_HLSStreamWriter):
         self.write_done = Event()
         self.write_error = None
 
+    def _futures_put(self, item):
+        self.futures.put_nowait(item)
+
+    def _futures_get(self):
+        return self.futures.get_nowait()
+
+    @staticmethod
+    def _future_result(future):
+        return future.result(timeout=0)
+
     def write(self, *args, **kwargs):
         # only write once per step
         self.write_wait.wait()
@@ -110,13 +125,8 @@ class HLSStreamReadThread(Thread):
     """
     Run the reader on a separate thread, so that each read can be controlled from within the main thread
     """
-    def __init__(self, session, stream, *args, **kwargs):
-        """
-        :param Streamlink session:
-        :param HLSStream stream:
-        """
-        Thread.__init__(self, *args, **kwargs)
-        self.daemon = True
+    def __init__(self, session: Streamlink, stream: HLSStream, *args, **kwargs):
+        super().__init__(*args, **kwargs, daemon=True)
 
         self.read_wait = Event()
         self.read_once = Event()
@@ -127,12 +137,12 @@ class HLSStreamReadThread(Thread):
 
         self.session = session
         self.stream = stream
-        self.reader = stream.open()
+        self.reader = stream.__reader__(stream)
 
         # ensure that at least one read was attempted before closing the writer thread early
         # otherwise, the writer will close the reader's buffer, making it not block on read and yielding empty results
         def _await_read_then_close():
-            self.read_once.wait(timeout=5)
+            self.read_once.wait(timeout=TIMEOUT_AWAIT_READ_ONCE)
             return self.writer_close()
 
         self.writer_close = self.reader.writer.close
@@ -168,18 +178,24 @@ class HLSStreamReadThread(Thread):
         self.data[:] = []
         self.error = None
 
+    def close(self):
+        self.reader.buffer.close()
+        self.read_once.set()
+        self.read_wait.set()
+
 
 class TestMixinStreamHLS(unittest.TestCase):
     __stream__ = HLSStream
     __readthread__ = HLSStreamReadThread
 
+    session: Streamlink
+    stream: HLSStream
+    thread: HLSStreamReadThread
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.mocker = requests_mock.Mocker()
         self.mocks = {}
-        self.session = None
-        self.stream = None
-        self.thread = None
 
     def setUp(self):
         super().setUp()
@@ -187,12 +203,19 @@ class TestMixinStreamHLS(unittest.TestCase):
 
     def tearDown(self):
         super().tearDown()
-        self.close_thread()
+
+        # close read thread and make sure that all threads have terminated before moving on
+        thread = self.thread
+        thread.reader.close()
+        if isinstance(thread.reader.writer, EventedHLSStreamWriter):
+            thread.reader.writer.write_wait.set()
+        thread.read_wait.set()
+        thread.reader.writer.join()
+        thread.reader.worker.join()
+        thread.join()
+
         self.mocker.stop()
         self.mocks.clear()
-        self.session = None
-        self.stream = None
-        self.thread = None
 
     def mock(self, method, url, *args, **kwargs):
         self.mocks[url] = self.mocker.request(method, url, *args, **kwargs)
@@ -213,47 +236,46 @@ class TestMixinStreamHLS(unittest.TestCase):
             segments = segments.values()
         return b"".join([getattr(segment, prop) for segment in segments if cond is None or cond(segment)])
 
-    # close read thread and make sure that all threads have terminated before moving on
-    def close_thread(self):
-        thread = self.thread
-        if isinstance(thread.reader.writer, EventedHLSStreamWriter):
-            thread.reader.writer.write_wait.set()
-        thread.reader.close()
-        thread.read_wait.set()
-        thread.reader.writer.join()
-        thread.reader.worker.join()
-        thread.join()
-
     # make one write call on the write thread and wait until it has finished
-    def await_write(self, write_calls=1, timeout=5):
+    def await_write(self, write_calls=1, timeout=TIMEOUT_AWAIT_WRITE):
         writer = self.thread.reader.writer
-        for _ in range(write_calls):
+        if not writer.is_alive():  # pragma: no cover
+            raise RuntimeError("Write thread is not alive")
+
+        for write_call in range(write_calls):
             writer.write_wait.set()
-            writer.write_done.wait(timeout)
-            writer.write_done.clear()
+            done = writer.write_done.wait(timeout)
             if writer.write_error:
                 raise writer.write_error
+            if not done:  # pragma: no cover
+                raise RuntimeError(f"Await write timeout: write_call={write_call + 1}")
+            writer.write_done.clear()
 
     # make one read call on the read thread and wait until it has finished
-    def await_read(self, read_all=False, timeout=5):
+    def await_read(self, read_all=False, timeout=TIMEOUT_AWAIT_READ):
         thread = self.thread
+        if not thread.is_alive():  # pragma: no cover
+            raise RuntimeError("Read thread is not alive")
+
         thread.read_all = read_all
         thread.read_wait.set()
-        thread.read_done.wait(timeout)
-        thread.read_done.clear()
+        done = thread.read_done.wait(timeout)
 
         try:
             if thread.error:
                 raise thread.error
+            if not done:  # pragma: no cover
+                raise RuntimeError(f"Await read timeout: read_all={read_all}")
             return b"".join(thread.data)
         finally:
+            thread.read_done.clear()
             thread.reset()
 
     def get_session(self, options=None, *args, **kwargs):
         return Streamlink(options)
 
     # set up HLS responses, create the session and read thread and start it
-    def subject(self, playlists, options=None, streamoptions=None, threadoptions=None, *args, **kwargs):
+    def subject(self, playlists, options=None, streamoptions=None, threadoptions=None, start=True, *args, **kwargs):
         # filter out tags and duplicate segments between playlist responses while keeping index order
         segments_all = [item for playlist in playlists for item in playlist.items if isinstance(item, Segment)]
         segments = OrderedDict([(segment.num, segment) for segment in segments_all])
@@ -264,7 +286,13 @@ class TestMixinStreamHLS(unittest.TestCase):
 
         self.session = self.get_session(options, *args, **kwargs)
         self.stream = self.__stream__(self.session, self.url(playlists[0]), **(streamoptions or {}))
-        self.thread = self.__readthread__(self.session, self.stream, **(threadoptions or {}))
-        self.thread.start()
+        self.thread = self.__readthread__(self.session, self.stream, name=f"ReadThread-{self.id()}", **(threadoptions or {}))
+
+        if start:
+            self.start()
 
         return self.thread, segments
+
+    def start(self):
+        self.thread.reader.open()
+        self.thread.start()
diff --git a/tests/plugins/test_nimotv.py b/tests/plugins/test_nimotv.py
index 46b06b1e..f7bd190c 100644
--- a/tests/plugins/test_nimotv.py
+++ b/tests/plugins/test_nimotv.py
@@ -10,4 +10,5 @@ class TestPluginNimoTV(PluginCanHandleUrl):
         'https://www.nimo.tv/live/737614',
         'http://www.nimo.tv/sanz',
         'https://www.nimo.tv/sanz',
+        'https://m.nimo.tv/user',
     ]
diff --git a/tests/plugins/test_tv4play.py b/tests/plugins/test_tv4play.py
index f1bd77a5..12f14018 100644
--- a/tests/plugins/test_tv4play.py
+++ b/tests/plugins/test_tv4play.py
@@ -6,8 +6,8 @@ class TestPluginCanHandleUrlTV4Play(PluginCanHandleUrl):
     __plugin__ = TV4Play
 
     should_match = [
-        'https://www.tv4play.se/program/fridas-vm-resa/10000884',
-        'https://www.tv4play.se/program/monk/3938989',
-        'https://www.tv4play.se/program/nyheterna/10378590',
+        'https://www.tv4play.se/program/robinson/del-26-sasong-2021/13299862',
+        'https://www.tv4play.se/program/sverige-mot-norge/del-1-sasong-1/12490380',
+        'https://www.tv4play.se/program/nyheterna/live/10378590',
         'https://www.fotbollskanalen.se/video/10395484/ghoddos-fullbordar-vandningen---ger-ofk-ledningen/',
     ]
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index ac3d4753..1970fe5f 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -1,5 +1,6 @@
 import os
 import unittest
+from threading import Event
 from unittest.mock import Mock, call, patch
 
 import pytest
@@ -248,6 +249,9 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
             Playlist(0, [key, SegmentEnc(0, aesKey, aesIv, padding=padding)], end=True)
         ])
 
+        # close read thread early
+        self.thread.close()
+
         with self.assertRaises(ValueError) as cm:
             self.await_write()
         self.assertEqual(str(cm.exception), "Padding is incorrect.", "Crypto.Util.Padding.unpad exception")
@@ -260,15 +264,24 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
             Playlist(0, [key, SegmentEnc(0, aesKey, aesIv, padding=padding)], end=True)
         ])
 
+        # close read thread early
+        self.thread.close()
+
         with self.assertRaises(ValueError) as cm:
             self.await_write()
         self.assertEqual(str(cm.exception), "PKCS#7 padding is incorrect.", "Crypto.Util.Padding.unpad exception")
 
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
-@patch("streamlink.stream.hls.HLSStreamWriter.run", Mock(return_value=True))
 class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
-    segments = [Segment(0, "", 11), Segment(1, "", 7), Segment(2, "", 5), Segment(3, "", 3)]
+    __stream__ = EventedHLSStream
+
+    segments = [
+        Segment(0, duration=11),
+        Segment(1, duration=7),
+        Segment(2, duration=5),
+        Segment(3, duration=3)
+    ]
 
     def get_session(self, options=None, reload_time=None, *args, **kwargs):
         return super().get_session(dict(options or {}, **{
@@ -277,10 +290,24 @@ class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
         }))
 
     def subject(self, *args, **kwargs):
-        thread, _ = super().subject(*args, **kwargs)
-        self.await_read(read_all=True)
+        thread, segments = super().subject(start=False, *args, **kwargs)
+
+        # mock the worker thread's _playlist_reload_time method, so that the main thread can wait on its call
+        playlist_reload_time_called = Event()
+        orig_playlist_reload_time = thread.reader.worker._playlist_reload_time
+
+        def mocked_playlist_reload_time(*args, **kwargs):
+            playlist_reload_time_called.set()
+            return orig_playlist_reload_time(*args, **kwargs)
 
-        return thread.reader.worker.playlist_reload_time
+        with patch.object(thread.reader.worker, "_playlist_reload_time", side_effect=mocked_playlist_reload_time):
+            self.start()
+            self.await_write(len(segments))
+
+            if not playlist_reload_time_called.wait(timeout=5):  # pragma: no cover
+                raise RuntimeError("Missing _playlist_reload_time() call")
+
+            return thread.reader.worker.playlist_reload_time
 
     def test_hls_playlist_reload_time_default(self):
         time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="default")
@@ -290,10 +317,26 @@ class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
         time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="segment")
         self.assertEqual(time, 3, "segment sets the reload time to the playlist's last segment")
 
+    def test_hls_playlist_reload_time_segment_no_segments(self):
+        time = self.subject([Playlist(0, [], end=True, targetduration=6)], reload_time="segment")
+        self.assertEqual(time, 6, "segment sets the reload time to the targetduration if no segments are available")
+
+    def test_hls_playlist_reload_time_segment_no_segments_no_targetduration(self):
+        time = self.subject([Playlist(0, [], end=True, targetduration=0)], reload_time="segment")
+        self.assertEqual(time, 15, "sets reload time to 15 seconds when no segments and no targetduration are available")
+
     def test_hls_playlist_reload_time_live_edge(self):
         time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="live-edge")
         self.assertEqual(time, 8, "live-edge sets the reload time to the sum of the number of segments of the live-edge")
 
+    def test_hls_playlist_reload_time_live_edge_no_segments(self):
+        time = self.subject([Playlist(0, [], end=True, targetduration=6)], reload_time="live-edge")
+        self.assertEqual(time, 6, "live-edge sets the reload time to the targetduration if no segments are available")
+
+    def test_hls_playlist_reload_time_live_edge_no_segments_no_targetduration(self):
+        time = self.subject([Playlist(0, [], end=True, targetduration=0)], reload_time="live-edge")
+        self.assertEqual(time, 15, "sets reload time to 15 seconds when no segments and no targetduration are available")
+
     def test_hls_playlist_reload_time_number(self):
         time = self.subject([Playlist(0, self.segments, end=True, targetduration=6)], reload_time="4")
         self.assertEqual(time, 4, "number values override the reload time")
diff --git a/tests/streams/test_hls_filtered.py b/tests/streams/test_hls_filtered.py
index 73b487db..b683973c 100644
--- a/tests/streams/test_hls_filtered.py
+++ b/tests/streams/test_hls_filtered.py
@@ -132,8 +132,8 @@ class TestFilteredHLSStream(TestMixinStreamHLS, unittest.TestCase):
 
     @patch("streamlink.stream.hls.HLSStreamWriter.should_filter_sequence", new=filter_sequence)
     def test_filtered_closed(self):
-        thread, reader, writer, segments = self.subject([
-            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)])
+        thread, reader, writer, segments = self.subject(start=False, playlists=[
+            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)], end=True)
         ])
 
         # mock the reader thread's filter_event.wait method, so that the main thread can wait on its call
@@ -144,27 +144,26 @@ class TestFilteredHLSStream(TestMixinStreamHLS, unittest.TestCase):
             filter_event_wait_called.set()
             return orig_wait(*args, **kwargs)
 
-        mock = patch.object(reader.filter_event, "wait", side_effect=mocked_wait)
-        mock.start()
+        with patch.object(reader.filter_event, "wait", side_effect=mocked_wait):
+            self.start()
 
-        # write first filtered segment and trigger the filter_event's lock
-        self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
-        self.await_write()
-        self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
-
-        # make reader read (no data available yet)
-        thread.read_wait.set()
-        # before calling reader.close(), wait until reader thread's filter_event.wait was called
-        filter_event_wait_called.wait()
-
-        # close stream while reader is waiting for filtering to end
-        thread.reader.close()
-        thread.read_done.wait()
-        thread.read_done.clear()
-        self.assertEqual(thread.data, [b""], "Stops reading on stream close")
-        self.assertFalse(thread.error, "Is not a read timeout on stream close")
+            # write first filtered segment and trigger the filter_event's lock
+            self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
+            self.await_write()
+            self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
 
-        mock.stop()
+            # make reader read (no data available yet)
+            thread.read_wait.set()
+            # before calling reader.close(), wait until reader thread's filter_event.wait was called
+            if not filter_event_wait_called.wait(timeout=5):  # pragma: no cover
+                raise RuntimeError("Missing filter_event.wait() call")
+
+            # close stream while reader is waiting for filtering to end
+            thread.reader.close()
+            thread.read_done.wait()
+            thread.read_done.clear()
+            self.assertEqual(thread.data, [b""], "Stops reading on stream close")
+            self.assertFalse(thread.error, "Is not a read timeout on stream close")
 
     def test_hls_segment_ignore_names(self):
         thread, reader, writer, segments = self.subject([