Repository: streamlink/streamlink
Pull request #3236 (2020-10-08 17:13:13)

docs: fix working tree check in deploy script
This should fix the clean working tree check in the docs deploy script.

I've restarted the CI jobs earlier today because of an issue on codecov and wanted the reports to be re-uploaded. Then the deploy-docs job failed afterwards, because it tried to commit from a clean working tree. The `git diff-index --quiet HEAD --` check was supposed to catch that, but apparently it doesn't. Checking for an empty stdout of `git status --porcelain` catches all changes, untracked files, etc. and is therefore better, even though git's high-level "porcelain" commands are technically not meant for scripting.

- https://stackoverflow.com/questions/2657935/checking-for-a-dirty-index-or-untracked-files-with-git/5737794#5737794
- https://github.com/streamlink/streamlink/runs/1225045245#step:7:25
- https://github.com/streamlink/streamlink.github.io/tree/b9a60115179db13cb3900a5af0be51ca66155a9f

Also added a better root dir check...

gravyboat: Looks good, I'm going to merge as that codecov check isn't ever going to finish.

Patch:
diff --git a/script/deploy-docs.sh b/script/deploy-docs.sh
index e64a81ca..cd83b487 100755
--- a/script/deploy-docs.sh
+++ b/script/deploy-docs.sh
@@ -4,7 +4,7 @@ set -eo pipefail
 
 [[ -n "${GITHUB_ACTIONS}" ]] || exit 1
 
-ROOT=$(realpath "$(dirname "$(readlink -f "${0}")")/..")
+ROOT=$(git rev-parse --show-toplevel 2>/dev/null || realpath "$(dirname "$(readlink -f "${0}")")/..")
 
 DOCS_REPO=${DOCS_REPO:-streamlink/streamlink.github.io}
 DOCS_BRANCH=${DOCS_BRANCH:-master}
@@ -69,7 +69,7 @@ cp --archive "${SOURCE}/." "${DEST}/"
 echo Building a new file list in \'${DEST}\'
 ( cd "${SOURCE}"; find . -type f | sort | sed "s/^\\.\//${DEST}\//" > "${TEMP}/${DEST}/${FILELIST}" )
 
-if git diff-index --quiet HEAD --; then
+if [[ -z "$(git status --porcelain)" ]]; then
     echo No changes to be committed. Exiting...
     exit 0
 fi
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
index 616dbd11..fd5ca904 100644
--- a/src/streamlink/plugins/twitch.py
+++ b/src/streamlink/plugins/twitch.py
@@ -3,123 +3,25 @@ import argparse
 import json
 import logging
 import re
-import warnings
 from collections import namedtuple
 from random import random
 
 import requests
 
 from streamlink.compat import urlparse
-from streamlink.exceptions import NoStreamsError, PluginError, StreamError
+from streamlink.exceptions import NoStreamsError, PluginError
 from streamlink.plugin import Plugin, PluginArguments, PluginArgument
 from streamlink.plugin.api import validate
 from streamlink.plugin.api.utils import parse_json, parse_query
-from streamlink.stream import (
-    HTTPStream, HLSStream, FLVPlaylist, extract_flv_header_tags
-)
+from streamlink.stream import HTTPStream, HLSStream
 from streamlink.stream.hls import HLSStreamWorker
 from streamlink.stream.hls_filtered import FilteredHLSStreamWriter, FilteredHLSStreamReader
 from streamlink.stream.hls_playlist import M3U8, M3U8Parser, load as load_hls_playlist
 from streamlink.utils.times import hours_minutes_seconds
 
-try:
-    from itertools import izip as zip
-except ImportError:
-    pass
-
 
 log = logging.getLogger(__name__)
 
-QUALITY_WEIGHTS = {
-    "source": 1080,
-    "1080": 1080,
-    "high": 720,
-    "720": 720,
-    "medium": 480,
-    "480": 480,
-    "360": 360,
-    "low": 240,
-    "mobile": 120,
-}
-
-# Streamlink's client-id used for public API calls (don't steal this and register your own application on Twitch)
-TWITCH_CLIENT_ID = "pwkzresl8kj2rdj6g7bvxl9ys1wly3j"
-# Twitch's client-id used for private API calls (see issue #2680 for why we are doing this)
-TWITCH_CLIENT_ID_PRIVATE = "kimne78kx3ncx6brgo4mv6wki5h1ko"
-
-_url_re = re.compile(r"""
-    http(s)?://
-    (?:
-        (?P<subdomain>[\w\-]+)
-        \.
-    )?
-    twitch.tv/
-    (?:
-        videos/(?P<videos_id>\d+)|
-        (?P<channel>[^/]+)
-    )
-    (?:
-        /
-        (?P<video_type>[bcv])(?:ideo)?
-        /
-        (?P<video_id>\d+)
-    )?
-    (?:
-        /(?:clip/)?
-        (?P<clip_name>[\w]+)
-    )?
-""", re.VERBOSE)
-
-_access_token_schema = validate.Schema(
-    {
-        "token": validate.text,
-        "sig": validate.text
-    },
-    validate.union((
-        validate.get("sig"),
-        validate.get("token")
-    ))
-)
-_token_schema = validate.Schema(
-    {
-        "chansub": {
-            "restricted_bitrates": validate.all(
-                [validate.text],
-                validate.filter(
-                    lambda n: not re.match(r"(.+_)?archives|live|chunked", n)
-                )
-            )
-        }
-    },
-    validate.get("chansub")
-)
-_stream_schema = validate.Schema(
-    {
-        "stream": validate.any(None, {
-            "stream_type": validate.text,
-            "broadcast_platform": validate.text,
-            "channel": validate.any(None, {
-                "broadcaster_software": validate.text
-            })
-        })
-    },
-    validate.get("stream")
-)
-_video_schema = validate.Schema(
-    {
-        "chunks": {
-            validate.text: [{
-                "length": int,
-                "url": validate.any(None, validate.url(scheme="http")),
-                "upkeep": validate.any("pass", "fail", None)
-            }]
-        },
-        "restrictions": {validate.text: validate.text},
-        "start_offset": int,
-        "end_offset": int,
-    }
-)
-
 Segment = namedtuple("Segment", "uri duration title key discontinuity ad byterange date map prefetch")
 
 LOW_LATENCY_MAX_LIVE_EDGE = 2
@@ -262,11 +164,7 @@ class UsherService(object):
         params.update(extra_params)
 
         req = requests.Request("GET", url, params=params)
-        # prepare_request is only available in requests 2.0+
-        if hasattr(self.session.http, "prepare_request"):
-            req = self.session.http.prepare_request(req)
-        else:
-            req = req.prepare()
+        req = self.session.http.prepare_request(req)
 
         return req.url
 
@@ -279,69 +177,212 @@ class UsherService(object):
 
 
 class TwitchAPI(object):
-    def __init__(self, session, beta=False, version=3):
+    # Streamlink's client-id used for public API calls (don't steal this and register your own application on Twitch)
+    TWITCH_CLIENT_ID = "pwkzresl8kj2rdj6g7bvxl9ys1wly3j"
+    # Twitch's client-id used for private API calls (see issue #2680 for why we are doing this)
+    TWITCH_CLIENT_ID_PRIVATE = "kimne78kx3ncx6brgo4mv6wki5h1ko"
+
+    def __init__(self, session):
         self.session = session
-        self.subdomain = beta and "betaapi" or "api"
-        self.version = version
 
-    def call(self, path, format="json", schema=None, private=False, **extra_params):
-        params = dict(as3="t", **extra_params)
+    def _call(self, method="GET", subdomain="api", path="/", headers=None, private=False, data=None, **params):
+        url = "https://{0}.twitch.tv{1}".format(subdomain, path)
+        headers = headers or dict()
+        headers.update({
+            "Client-ID": self.TWITCH_CLIENT_ID if not private else self.TWITCH_CLIENT_ID_PRIVATE
+        })
 
-        if len(format) > 0:
-            url = "https://{0}.twitch.tv{1}.{2}".format(self.subdomain, path, format)
-        else:
-            url = "https://{0}.twitch.tv{1}".format(self.subdomain, path)
+        return self.session.http.request(method, url, data=data, params=params, headers=headers)
 
-        headers = {'Accept': 'application/vnd.twitchtv.v{0}+json'.format(self.version),
-                   'Client-ID': TWITCH_CLIENT_ID if not private else TWITCH_CLIENT_ID_PRIVATE}
+    def call(self, path, schema=None, **params):
+        headers = {"Accept": "application/vnd.twitchtv.v5+json"}
+        res = self._call(path=path, headers=headers, **params)
 
-        res = self.session.http.get(url, params=params, headers=headers)
+        return self.session.http.json(res, schema=schema)
 
-        if format == "json":
-            return self.session.http.json(res, schema=schema)
-        else:
-            return res
+    def call_gql(self, data, schema=None, **params):
+        res = self._call(method="POST", subdomain="gql", path="/gql", data=json.dumps(data), private=True, **params)
 
-    def call_subdomain(self, subdomain, path, format="json", schema=None, **extra_params):
-        subdomain_buffer = self.subdomain
-        self.subdomain = subdomain
-        try:
-            return self.call(path, format=format, schema=schema, **extra_params)
-        finally:
-            self.subdomain = subdomain_buffer
+        return self.session.http.json(res, schema=schema)
 
     # Public API calls
 
-    def users(self, **params):
-        return self.call("/kraken/users", **params)
-
-    def videos(self, video_id, **params):
-        return self.call("/kraken/videos/{0}".format(video_id), **params)
-
-    def channel_info(self, channel, **params):
-        return self.call("/kraken/channels/{0}".format(channel), **params)
-
-    def streams(self, channel_id, **params):
-        return self.call("/kraken/streams/{0}".format(channel_id), **params)
+    def channel_from_video_id(self, video_id):
+        return self.call("/kraken/videos/{0}".format(video_id), schema=validate.Schema(
+            {
+                "channel": {
+                    "_id": validate.any(int, validate.text),
+                    "name": validate.text
+                }
+            },
+            validate.get("channel"),
+            validate.union((
+                validate.all(validate.get("_id"), validate.transform(int)),
+                validate.all(validate.get("name"), validate.transform(lambda n: n.lower()))
+            ))
+        ))
+
+    def channel_from_login(self, channel):
+        return self.call("/kraken/users", login=channel, schema=validate.Schema(
+            {
+                "users": [{
+                    "_id": validate.any(int, validate.text)
+                }]
+            },
+            validate.get("users"),
+            validate.get(0),
+            validate.get("_id"),
+            validate.transform(int)
+        ))
+
+    def stream_rerun(self, channel_id):
+        return self.call("/kraken/streams/{0}".format(channel_id), schema=validate.Schema(
+            {
+                "stream": validate.any(None, validate.all({
+                    "stream_type": validate.text,
+                    "broadcast_platform": validate.text,
+                    "channel": validate.all({
+                        "broadcaster_software": validate.text
+                    }, validate.get("broadcaster_software"))
+                }, validate.union((
+                    validate.get("stream_type"),
+                    validate.get("broadcast_platform"),
+                    validate.get("channel")
+                ))))
+            },
+            validate.get("stream")
+        ))
+
+    def metadata_video(self, video_id):
+        return self.call("/kraken/videos/{0}".format(video_id), schema=validate.Schema(validate.any(
+            validate.all(
+                {
+                    "title": validate.text,
+                    "game": validate.text,
+                    "channel": validate.all(
+                        {"display_name": validate.text},
+                        validate.get("display_name")
+                    )
+                },
+                validate.transform(lambda data: (data["channel"], data["title"], data["game"]))
+            ),
+            validate.all({}, validate.transform(lambda _: (None,) * 3))
+        )))
+
+    def metadata_channel(self, channel_id):
+        return self.call("/kraken/streams/{0}".format(channel_id), schema=validate.Schema(
+            {
+                "stream": validate.any(
+                    validate.all(
+                        {"channel": {
+                            "display_name": validate.text,
+                            "game": validate.text,
+                            "status": validate.text
+                        }},
+                        validate.get("channel"),
+                        validate.transform(lambda ch: (ch["display_name"], ch["status"], ch["game"]))
+                    ),
+                    validate.all(None, validate.transform(lambda _: (None,) * 3))
+                )
+            },
+            validate.get("stream")
+        ))
 
     # Private API calls
 
-    def access_token(self, endpoint, asset, **params):
-        return self.call("/api/{0}/{1}/access_token".format(endpoint, asset), private=True, **params)
-
-    def hosted_channel(self, **params):
-        return self.call_subdomain("tmi", "/hosts", format="", **params)
-
-    # Unsupported/Removed private API calls
-
-    def channel_viewer_info(self, channel, **params):
-        warnings.warn("The channel_viewer_info API call is unsupported and may stop working at any time")
-        return self.call("/api/channels/{0}/viewer".format(channel), private=True, **params)
+    def access_token(self, endpoint, asset):
+        return self.call("/api/{0}/{1}/access_token".format(endpoint, asset), private=True, schema=validate.Schema(
+            {
+                "token": validate.text,
+                "sig": validate.text
+            },
+            validate.union((
+                validate.get("sig"),
+                validate.get("token")
+            ))
+        ))
+
+    def token(self, tokenstr):
+        return parse_json(tokenstr, schema=validate.Schema(
+            {
+                "chansub": {
+                    "restricted_bitrates": validate.all(
+                        [validate.text],
+                        validate.filter(
+                            lambda n: not re.match(r"(.+_)?archives|live|chunked", n)
+                        )
+                    )
+                }
+            },
+            validate.get("chansub"),
+            validate.get("restricted_bitrates")
+        ))
 
-    def channel_subscription(self, channel, **params):
-        warnings.warn("The channel_subscription API call has been removed and no longer works",
-                      category=DeprecationWarning)
-        return self.call("/api/channels/{0}/subscription".format(channel), private=True, **params)
+    def hosted_channel(self, channel_id):
+        return self.call("/hosts", subdomain="tmi", include_logins=1, host=channel_id, schema=validate.Schema(
+            {
+                "hosts": [{
+                    "host_id": int,
+                    "target_id": int,
+                    "target_login": validate.text,
+                    "target_display_name": validate.text
+                }]
+            },
+            validate.get("hosts"),
+            validate.get(0),
+            validate.union((
+                validate.get("host_id"),
+                validate.get("target_id"),
+                validate.get("target_login"),
+                validate.get("target_display_name")
+            ))
+        ))
+
+    # GraphQL API calls
+
+    def clips(self, clipname):
+        query = """{{
+            clip(slug: "{clipname}") {{
+                broadcaster {{
+                    displayName
+                }}
+                title
+                game {{
+                    name
+                }}
+                videoQualities {{
+                    quality
+                    sourceURL
+                }}
+            }}
+        }}""".format(clipname=clipname)
+
+        return self.call_gql({"query": query}, schema=validate.Schema(
+            {"data": {
+                "clip": validate.any(None, validate.all({
+                    "broadcaster": validate.all({"displayName": validate.text}, validate.get("displayName")),
+                    "title": validate.text,
+                    "game": validate.all({"name": validate.text}, validate.get("name")),
+                    "videoQualities": [validate.all({
+                        "quality": validate.all(
+                            validate.text,
+                            validate.transform(lambda q: "{0}p".format(q))
+                        ),
+                        "sourceURL": validate.url()
+                    }, validate.union((
+                        validate.get("quality"),
+                        validate.get("sourceURL")
+                    )))]
+                }, validate.union((
+                    validate.get("broadcaster"),
+                    validate.get("title"),
+                    validate.get("game"),
+                    validate.get("videoQualities")
+                ))))
+            }},
+            validate.get("data"),
+            validate.get("clip")
+        ))
 
 
 class Twitch(Plugin):
@@ -396,83 +437,31 @@ class Twitch(Plugin):
         )
     )
 
-    _schema_metadata_empty = validate.transform(lambda _: (None,) * 3)
-    _schema_metadata_channel = validate.Schema(
-        {
-            "stream": validate.any(
-                validate.all(
-                    {"channel": {
-                        "display_name": validate.text,
-                        "game": validate.text,
-                        "status": validate.text
-                    }},
-                    validate.get("channel"),
-                    validate.transform(lambda ch: (ch["display_name"], ch["status"], ch["game"]))
-                ),
-                validate.all(None, _schema_metadata_empty)
-            )
-        },
-        validate.get("stream")
-    )
-    _schema_metadata_video = validate.Schema(validate.any(
-        validate.all(
-            {
-                "title": validate.text,
-                "game": validate.text,
-                "channel": validate.all(
-                    {"display_name": validate.text},
-                    validate.get("display_name")
-                )
-            },
-            validate.transform(lambda data: (data["channel"], data["title"], data["game"]))
-        ),
-        validate.all({}, _schema_metadata_empty)
-    ))
-
-    @classmethod
-    def stream_weight(cls, key):
-        weight = QUALITY_WEIGHTS.get(key)
-        if weight:
-            return weight, "twitch"
-
-        return Plugin.stream_weight(key)
+    _re_url = re.compile(r"""
+        https?://(?:(?P<subdomain>[\w\-]+)\.)?twitch\.tv/
+        (?:
+            videos/(?P<videos_id>\d+)
+            |
+            (?P<channel>[^/]+)
+            (?:
+                /video/(?P<video_id>\d+)
+                |
+                /clip/(?P<clip_name>[\w]+)
+            )?
+        )
+    """, re.VERBOSE)
 
     @classmethod
     def can_handle_url(cls, url):
-        return _url_re.match(url)
-
-    def _get_metadata(self):
-        if self.video_id:
-            (self.author, self.title, self.category) = self.api.videos(self.video_id, schema=self._schema_metadata_video)
-        elif self.clip_name:
-            self._get_clips()
-        elif self._channel:
-            (self.author, self.title, self.category) = self.api.streams(self.channel_id, schema=self._schema_metadata_channel)
-
-    def get_title(self):
-        if self.title is None:
-            self._get_metadata()
-        return self.title
-
-    def get_author(self):
-        if self.author is None:
-            self._get_metadata()
-        return self.author
-
-    def get_category(self):
-        if self.category is None:
-            self._get_metadata()
-        return self.category
+        return cls._re_url.match(url)
 
     def __init__(self, url):
-        Plugin.__init__(self, url)
-        self._hosted_chain = []
-        match = _url_re.match(url).groupdict()
+        super(Twitch, self).__init__(url)
+        match = self._re_url.match(url).groupdict()
         parsed = urlparse(url)
         self.params = parse_query(parsed.query)
         self.subdomain = match.get("subdomain")
         self.video_id = None
-        self.video_type = None
         self._channel_id = None
         self._channel = None
         self.clip_name = None
@@ -483,270 +472,147 @@ class Twitch(Plugin):
         if self.subdomain == "player":
             # pop-out player
             if self.params.get("video"):
-                try:
-                    self.video_type = self.params["video"][0]
-                    self.video_id = self.params["video"][1:]
-                except IndexError:
-                    log.debug("Invalid video param: {0}".format(self.params["video"]))
+                self.video_id = self.params["video"]
             self._channel = self.params.get("channel")
         elif self.subdomain == "clips":
             # clip share URL
             self.clip_name = match.get("channel")
         else:
             self._channel = match.get("channel") and match.get("channel").lower()
-            self.video_type = match.get("video_type")
-            if match.get("videos_id"):
-                self.video_type = "v"
             self.video_id = match.get("video_id") or match.get("videos_id")
             self.clip_name = match.get("clip_name")
 
-        self.api = TwitchAPI(beta=self.subdomain == "beta",
-                             session=self.session,
-                             version=5)
+        self.api = TwitchAPI(session=self.session)
         self.usher = UsherService(session=self.session)
 
+    def get_title(self):
+        if self.title is None:
+            self._get_metadata()
+        return self.title
+
+    def get_author(self):
+        if self.author is None:
+            self._get_metadata()
+        return self.author
+
+    def get_category(self):
+        if self.category is None:
+            self._get_metadata()
+        return self.category
+
+    def _get_metadata(self):
+        if self.video_id:
+            (self.author, self.title, self.category) = self.api.metadata_video(self.video_id)
+        elif self.clip_name:
+            self._get_clips()
+        elif self._channel:
+            (self.author, self.title, self.category) = self.api.metadata_channel(self.channel_id)
+
     @property
     def channel(self):
         if not self._channel:
             if self.video_id:
-                cdata = self._channel_from_video_id(self.video_id)
-                self._channel = cdata["name"].lower()
-                self._channel_id = cdata["_id"]
+                self._channel_from_video_id(self.video_id)
         return self._channel
 
-    @channel.setter
-    def channel(self, channel):
-        self._channel = channel
-        # channel id becomes unknown
-        self._channel_id = None
-
     @property
     def channel_id(self):
         if not self._channel_id:
-            # If the channel name is set, use that to look up the ID
             if self._channel:
-                cdata = self._channel_from_login(self._channel)
-                self._channel_id = cdata["_id"]
-
-            # If the channel name is not set but the video ID is,
-            # use that to look up both ID and name
+                self._channel_from_login(self._channel)
             elif self.video_id:
-                cdata = self._channel_from_video_id(self.video_id)
-                self._channel = cdata["name"].lower()
-                self._channel_id = cdata["_id"]
+                self._channel_from_video_id(self.video_id)
         return self._channel_id
 
     def _channel_from_video_id(self, video_id):
-        vdata = self.api.videos(video_id)
-        if "channel" not in vdata:
+        try:
+            self._channel_id, self._channel = self.api.channel_from_video_id(video_id)
+        except PluginError:
             raise PluginError("Unable to find video: {0}".format(video_id))
-        return vdata["channel"]
 
     def _channel_from_login(self, channel):
-        cdata = self.api.users(login=channel)
-        if len(cdata["users"]):
-            return cdata["users"][0]
-        else:
+        try:
+            self._channel_id = self.api.channel_from_login(channel)
+        except PluginError:
             raise PluginError("Unable to find channel: {0}".format(channel))
 
-    def _create_playlist_streams(self, videos):
-        start_offset = int(videos.get("start_offset", 0))
-        stop_offset = int(videos.get("end_offset", 0))
-        streams = {}
-
-        for quality, chunks in videos.get("chunks").items():
-            if not chunks:
-                if videos.get("restrictions", {}).get(quality) == "chansub":
-                    log.warning("The quality '{0}' is not available since it requires a subscription.".format(quality))
-                continue
-
-            # Rename 'live' to 'source'
-            if quality == "live":
-                quality = "source"
-
-            chunks_filtered = list(filter(lambda c: c["url"], chunks))
-            if len(chunks) != len(chunks_filtered):
-                log.warning("The video '{0}' contains invalid chunks. There will be missing data.".format(quality))
-                chunks = chunks_filtered
-
-            chunks_duration = sum(c.get("length") for c in chunks)
-
-            # If it's a full broadcast we just use all the chunks
-            if start_offset == 0 and chunks_duration == stop_offset:
-                # No need to use the FLV concat if it's just one chunk
-                if len(chunks) == 1:
-                    url = chunks[0].get("url")
-                    stream = HTTPStream(self.session, url)
-                else:
-                    chunks = [HTTPStream(self.session, c.get("url")) for c in chunks]
-                    stream = FLVPlaylist(self.session, chunks,
-                                         duration=chunks_duration)
-            else:
-                try:
-                    stream = self._create_video_clip(chunks,
-                                                     start_offset,
-                                                     stop_offset)
-                except StreamError as err:
-                    log.error("Error while creating video '{0}': {1}".format(quality, err))
-                    continue
-
-            streams[quality] = stream
-
-        return streams
-
-    def _create_video_clip(self, chunks, start_offset, stop_offset):
-        playlist_duration = stop_offset - start_offset
-        playlist_offset = 0
-        playlist_streams = []
-        playlist_tags = []
-
-        for chunk in chunks:
-            chunk_url = chunk["url"]
-            chunk_length = chunk["length"]
-            chunk_start = playlist_offset
-            chunk_stop = chunk_start + chunk_length
-            chunk_stream = HTTPStream(self.session, chunk_url)
-
-            if chunk_start <= start_offset <= chunk_stop:
-                try:
-                    headers = extract_flv_header_tags(chunk_stream)
-                except IOError as err:
-                    raise StreamError("Error while parsing FLV: {0}", err)
-
-                if not headers.metadata:
-                    raise StreamError("Missing metadata tag in the first chunk")
-
-                metadata = headers.metadata.data.value
-                keyframes = metadata.get("keyframes")
-
-                if not keyframes:
-                    if chunk["upkeep"] == "fail":
-                        raise StreamError("Unable to seek into muted chunk, try another timestamp")
-                    else:
-                        raise StreamError("Missing keyframes info in the first chunk")
-
-                keyframe_offset = None
-                keyframe_offsets = keyframes.get("filepositions")
-                keyframe_times = [playlist_offset + t for t in keyframes.get("times")]
-                for time, offset in zip(keyframe_times, keyframe_offsets):
-                    if time > start_offset:
-                        break
-
-                    keyframe_offset = offset
-
-                if keyframe_offset is None:
-                    raise StreamError("Unable to find a keyframe to seek to "
-                                      "in the first chunk")
-
-                chunk_headers = dict(Range="bytes={0}-".format(int(keyframe_offset)))
-                chunk_stream = HTTPStream(self.session, chunk_url,
-                                          headers=chunk_headers)
-                playlist_streams.append(chunk_stream)
-                for tag in headers:
-                    playlist_tags.append(tag)
-            elif start_offset <= chunk_start < stop_offset:
-                playlist_streams.append(chunk_stream)
-
-            playlist_offset += chunk_length
-
-        return FLVPlaylist(self.session, playlist_streams,
-                           tags=playlist_tags, duration=playlist_duration)
-
-    def _get_video_streams(self):
-        log.debug("Getting video steams for {0} (type={1})".format(self.video_id, self.video_type))
-
-        if self.video_type == "b":
-            self.video_type = "a"
-
+    def _access_token(self, endpoint, asset):
         try:
-            videos = self.api.videos(self.video_type + self.video_id,
-                                     schema=_video_schema)
+            sig, token = self.api.access_token(endpoint, asset)
         except PluginError as err:
-            if "HTTP/1.1 0 ERROR" in str(err):
+            if "404 Client Error" in str(err):
                 raise NoStreamsError(self.url)
             else:
                 raise
 
-        # Parse the "t" query parameter on broadcasts and adjust
-        # start offset if needed.
-        time_offset = self.params.get("t")
-        if time_offset:
-            try:
-                time_offset = hours_minutes_seconds(time_offset)
-            except ValueError:
-                time_offset = 0
+        try:
+            restricted_bitrates = self.api.token(token)
+        except PluginError:
+            restricted_bitrates = []
 
-            videos["start_offset"] += time_offset
+        return sig, token, restricted_bitrates
 
-        return self._create_playlist_streams(videos)
+    def _switch_to_hosted_channel(self):
+        disabled = self.options.get("disable_hosting")
+        hosted_chain = [self.channel]
+        while True:
+            try:
+                host_id, target_id, login, display_name = self.api.hosted_channel(self.channel_id)
+            except PluginError:
+                return False
 
-    def _access_token(self, type="live"):
-        try:
-            if type == "live":
-                endpoint = "channels"
-                value = self.channel
-            elif type == "video":
-                endpoint = "vods"
-                value = self.video_id
-
-            sig, token = self.api.access_token(endpoint, value,
-                                               schema=_access_token_schema)
-        except PluginError as err:
-            if "404 Client Error" in str(err):
-                raise NoStreamsError(self.url)
-            else:
-                raise
+            log.info("{0} is hosting {1}".format(self.channel, login))
+            if disabled:
+                log.info("hosting was disabled by command line option")
+                return True
 
-        return sig, token
+            if login in hosted_chain:
+                loop = " -> ".join(hosted_chain + [login])
+                log.error("A loop of hosted channels has been detected, cannot find a playable stream. ({0})".format(loop))
+                return True
 
-    def _check_for_host(self):
-        host_info = self.api.hosted_channel(include_logins=1, host=self.channel_id).json()["hosts"][0]
-        if "target_login" in host_info and host_info["target_login"].lower() != self.channel.lower():
-            log.info("{0} is hosting {1}".format(self.channel, host_info["target_login"]))
-            return host_info["target_login"]
+            hosted_chain.append(login)
+            log.info("switching to {0}".format(login))
+            self._channel_id = target_id
+            self._channel = login
+            self.author = display_name
 
     def _check_for_rerun(self):
-        stream = self.api.streams(self.channel_id, schema=_stream_schema)
+        if not self.options.get("disable_reruns"):
+            return False
 
-        return stream and (
-            stream["stream_type"] != "live"
-            or stream["broadcast_platform"] == "rerun"
-            or stream["channel"] and stream["channel"]["broadcaster_software"] == "watch_party_rerun"
-        )
+        try:
+            stream_type, broadcast_platform, broadcaster_software = self.api.stream_rerun(self.channel_id)
+            if stream_type != "live" or broadcast_platform == "rerun" or broadcaster_software == "watch_party_rerun":
+                log.info("Reruns were disabled by command line option")
+                return True
+        except (PluginError, TypeError):
+            pass
 
-    def _get_hls_streams(self, stream_type="live"):
-        log.debug("Getting {0} HLS streams for {1}".format(stream_type, self.channel))
-        self._hosted_chain.append(self.channel)
+        return False
 
-        if stream_type == "live":
-            if self.options.get("disable_reruns") and self._check_for_rerun():
-                log.info("Reruns were disabled by command line option")
-                return {}
+    def _get_hls_streams_live(self):
+        if self._switch_to_hosted_channel():
+            return
+        if self._check_for_rerun():
+            return
 
-            hosted_channel = self._check_for_host()
-            if hosted_channel and self.options.get("disable_hosting"):
-                log.info("hosting was disabled by command line option")
-            elif hosted_channel:
-                log.info("switching to {0}".format(hosted_channel))
-                if hosted_channel in self._hosted_chain:
-                    log.error(
-                        u"A loop of hosted channels has been detected, "
-                        "cannot find a playable stream. ({0})".format(
-                            u" -> ".join(self._hosted_chain + [hosted_channel])))
-                    return {}
-                self.channel = hosted_channel
-                return self._get_hls_streams(stream_type)
-
-            # only get the token once the channel has been resolved
-            sig, token = self._access_token(stream_type)
-            url = self.usher.channel(self.channel, sig=sig, token=token, fast_bread=True)
-        elif stream_type == "video":
-            sig, token = self._access_token(stream_type)
-            url = self.usher.video(self.video_id, nauthsig=sig, nauth=token)
-        else:
-            log.debug("Unknown HLS stream type: {0}".format(stream_type))
-            return {}
+        # only get the token once the channel has been resolved
+        log.debug("Getting live HLS streams for {0}".format(self.channel))
+        sig, token, restricted_bitrates = self._access_token("channels", self.channel)
+        url = self.usher.channel(self.channel, sig=sig, token=token, fast_bread=True)
 
+        return self._get_hls_streams(url, restricted_bitrates)
+
+    def _get_hls_streams_video(self):
+        log.debug("Getting video HLS streams for {0}".format(self.channel))
+        sig, token, restricted_bitrates = self._access_token("vods", self.video_id)
+        url = self.usher.video(self.video_id, nauthsig=sig, nauth=token)
+
+        # If the stream is a VOD that is still being recorded, the stream should start at the beginning of the recording
+        return self._get_hls_streams(url, restricted_bitrates, force_restart=True)
+
+    def _get_hls_streams(self, url, restricted_bitrates, **extra_params):
         time_offset = self.params.get("t", 0)
         if time_offset:
             try:
@@ -755,14 +621,7 @@ class Twitch(Plugin):
                 time_offset = 0
 
         try:
-            # If the stream is a VOD that is still being recorded the stream should start at the
-            # beginning of the recording
-            streams = TwitchHLSStream.parse_variant_playlist(
-                self.session,
-                url,
-                start_offset=time_offset,
-                force_restart=not stream_type == "live"
-            )
+            streams = TwitchHLSStream.parse_variant_playlist(self.session, url, start_offset=time_offset, **extra_params)
         except IOError as err:
             err = str(err)
             if "404 Client Error" in err or "Failed to parse playlist" in err:
@@ -770,55 +629,28 @@ class Twitch(Plugin):
             else:
                 raise PluginError(err)
 
-        try:
-            token = parse_json(token, schema=_token_schema)
-            for name in token["restricted_bitrates"]:
-                if name not in streams:
-                    log.warning("The quality '{0}' is not available since it requires a subscription.".format(name))
-        except PluginError:
-            pass
+        for name in restricted_bitrates:
+            if name not in streams:
+                log.warning("The quality '{0}' is not available since it requires a subscription.".format(name))
 
         return streams
 
     def _get_clips(self):
-        data = json.dumps({'query': '''{{
-            clip(slug: "{0}") {{
-                broadcaster {{
-                    displayName
-                }}
-                title
-                videoQualities {{
-                    quality
-                    sourceURL
-                }}
-            }}
-        }}'''.format(self.clip_name)})
-        clip_data = self.session.http.post('https://gql.twitch.tv/gql',
-                                           data=data,
-                                           headers={'Client-ID': TWITCH_CLIENT_ID_PRIVATE},
-                                           ).json()['data']['clip']
-        log.trace('{0!r}'.format(clip_data))
-        if not clip_data:
+        try:
+            (self.author, self.title, self.category, streams) = self.api.clips(self.clip_name)
+        except (PluginError, TypeError):
             return
 
-        self.author = clip_data['broadcaster']['displayName']
-        self.title = clip_data['title']
-
-        streams = {}
-        for quality_option in clip_data['videoQualities']:
-            streams['{0}p'.format(quality_option['quality'])] = HTTPStream(self.session, quality_option['sourceURL'])
-        return streams
+        for quality, stream in streams:
+            yield quality, HTTPStream(self.session, stream)
 
     def _get_streams(self):
         if self.video_id:
-            if self.video_type == "v":
-                return self._get_hls_streams("video")
-            else:
-                return self._get_video_streams()
+            return self._get_hls_streams_video()
         elif self.clip_name:
             return self._get_clips()
         elif self._channel:
-            return self._get_hls_streams("live")
+            return self._get_hls_streams_live()
 
 
 __plugin__ = Twitch
diff --git a/src/streamlink/stream/segmented.py b/src/streamlink/stream/segmented.py
index d33cc9e4..dd727d8f 100644
--- a/src/streamlink/stream/segmented.py
+++ b/src/streamlink/stream/segmented.py
@@ -1,7 +1,8 @@
-import concurrent.futures.thread
-import logging
 from concurrent import futures
+from concurrent.futures.thread import ThreadPoolExecutor
+import logging
 from threading import Thread, Event
+from sys import version_info
 
 from .stream import StreamIO
 from ..buffers import RingBuffer
@@ -10,6 +11,30 @@ from ..compat import queue
 log = logging.getLogger(__name__)
 
 
+class CompatThreadPoolExecutor(ThreadPoolExecutor):
+    if version_info < (3, 9):
+        def shutdown(self, wait=True, cancel_futures=False):
+            with self._shutdown_lock:
+                self._shutdown = True
+                if cancel_futures:
+                    # Drain all work items from the queue, and then cancel their
+                    # associated futures.
+                    while True:
+                        try:
+                            work_item = self._work_queue.get_nowait()
+                        except queue.Empty:
+                            break
+                        if work_item is not None:
+                            work_item.future.cancel()
+
+                # Send a wake-up to prevent threads calling
+                # _work_queue.get(block=True) from permanently blocking.
+                self._work_queue.put(None)
+            if wait:
+                for t in self._threads:
+                    t.join()
+
+
 class SegmentedStreamWorker(Thread):
     """The general worker thread.
 
@@ -91,7 +116,7 @@ class SegmentedStreamWriter(Thread):
         self.retries = retries
         self.timeout = timeout
         self.ignore_names = ignore_names
-        self.executor = futures.ThreadPoolExecutor(max_workers=threads)
+        self.executor = CompatThreadPoolExecutor(max_workers=threads)
         self.futures = queue.Queue(size)
 
         Thread.__init__(self, name="Thread-{0}".format(self.__class__.__name__))
@@ -104,9 +129,7 @@ class SegmentedStreamWriter(Thread):
 
         self.closed = True
         self.reader.buffer.close()
-        self.executor.shutdown(wait=False)
-        if concurrent.futures.thread._threads_queues:
-            concurrent.futures.thread._threads_queues.clear()
+        self.executor.shutdown(wait=True, cancel_futures=True)
 
     def put(self, segment):
         """Adds a segment to the download pool and write queue."""


Test patch:
diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py
index ec8ace24..af9aae43 100644
--- a/tests/mixins/stream_hls.py
+++ b/tests/mixins/stream_hls.py
@@ -95,6 +95,7 @@ class HLSStreamReadThread(Thread):
         self.daemon = True
 
         self.read_wait = Event()
+        self.read_once = Event()
         self.read_done = Event()
         self.read_all = False
         self.data = []
@@ -104,11 +105,21 @@ class HLSStreamReadThread(Thread):
         self.stream = stream
         self.reader = stream.open()
 
+        # 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)
+            return self.writer_close()
+
+        self.writer_close = self.reader.writer.close
+        self.reader.writer.close = _await_read_then_close
+
     def run(self):
         while not self.reader.buffer.closed:
             # only read once per step
             self.read_wait.wait()
             self.read_wait.clear()
+            self.read_once.set()
 
             try:
                 # don't read again during teardown
diff --git a/tests/plugins/test_twitch.py b/tests/plugins/test_twitch.py
index bb90349d..d66833a9 100644
--- a/tests/plugins/test_twitch.py
+++ b/tests/plugins/test_twitch.py
@@ -291,11 +291,11 @@ class TestTwitchMetadata(unittest.TestCase):
 
     def subject_channel(self, data=True, failure=False):
         self.mock.get(
-            "https://api.twitch.tv/kraken/users.json?login=foo",
+            "https://api.twitch.tv/kraken/users?login=foo",
             json={"users": [{"_id": 1234}]}
         )
         self.mock.get(
-            "https://api.twitch.tv/kraken/streams/1234.json",
+            "https://api.twitch.tv/kraken/streams/1234",
             status_code=200 if not failure else 404,
             json={"stream": None} if not data else {"stream": {
                 "channel": {
@@ -309,7 +309,7 @@ class TestTwitchMetadata(unittest.TestCase):
 
     def subject_video(self, data=True, failure=False):
         self.mock.get(
-            "https://api.twitch.tv/kraken/videos/1337.json",
+            "https://api.twitch.tv/kraken/videos/1337",
             status_code=200 if not failure else 404,
             json={} if not data else {
                 "title": "video title",
@@ -350,23 +350,134 @@ class TestTwitchMetadata(unittest.TestCase):
             self.subject_video(failure=True)
 
 
+@patch("streamlink.plugins.twitch.log")
+class TestTwitchHosting(unittest.TestCase):
+    def subject(self, channel, hosts=None, disable=False):
+        with requests_mock.Mocker() as mock:
+            mock.get(
+                "https://api.twitch.tv/kraken/users?login=foo",
+                json={"users": [{"_id": 1}]}
+            )
+            if hosts is None:
+                mock.get("https://tmi.twitch.tv/hosts", json={})
+            else:
+                mock.get(
+                    "https://tmi.twitch.tv/hosts",
+                    [{"json": {
+                        "hosts": [dict(
+                            host_id=host_id,
+                            target_id=target_id,
+                            target_login=target_login,
+                            target_display_name=target_display_name
+                        )]}
+                      } for host_id, target_id, target_login, target_display_name in hosts]
+                )
+
+            session = Streamlink()
+            Twitch.bind(session, "tests.plugins.test_twitch")
+            plugin = Twitch("https://twitch.tv/{0}".format(channel))
+            plugin.options.set("disable-hosting", disable)
+
+            res = plugin._switch_to_hosted_channel()
+            return res, plugin.channel, plugin._channel_id, plugin.author
+
+    def test_hosting_invalid_host_data(self, mock_log):
+        res, channel, channel_id, author = self.subject("foo")
+        self.assertFalse(res, "Doesn't stop HLS resolve procedure")
+        self.assertEqual(channel, "foo", "Doesn't switch channel")
+        self.assertEqual(channel_id, 1, "Doesn't switch channel id")
+        self.assertEqual(author, None, "Doesn't override author metadata")
+        self.assertEqual(mock_log.info.mock_calls, [], "Doesn't log anything to info")
+        self.assertEqual(mock_log.error.mock_calls, [], "Doesn't log anything to error")
+
+    def test_hosting_no_host_data(self, mock_log):
+        res, channel, channel_id, author = self.subject("foo", [(1, None, None, None)])
+        self.assertFalse(res, "Doesn't stop HLS resolve procedure")
+        self.assertEqual(channel, "foo", "Doesn't switch channel")
+        self.assertEqual(channel_id, 1, "Doesn't switch channel id")
+        self.assertEqual(author, None, "Doesn't override author metadata")
+        self.assertEqual(mock_log.info.mock_calls, [], "Doesn't log anything to info")
+        self.assertEqual(mock_log.error.mock_calls, [], "Doesn't log anything to error")
+
+    def test_hosting_host_single(self, mock_log):
+        res, channel, channel_id, author = self.subject("foo", [(1, 2, "bar", "Bar"), (2, None, None, None)])
+        self.assertFalse(res, "Doesn't stop HLS resolve procedure")
+        self.assertEqual(channel, "bar", "Switches channel")
+        self.assertEqual(channel_id, 2, "Switches channel id")
+        self.assertEqual(author, "Bar", "Overrides author metadata")
+        self.assertEqual(mock_log.info.mock_calls, [
+            call("foo is hosting bar"),
+            call("switching to bar")
+        ])
+        self.assertEqual(mock_log.error.mock_calls, [], "Doesn't log anything to error")
+
+    def test_hosting_host_single_disable(self, mock_log):
+        res, channel, channel_id, author = self.subject("foo", [(1, 2, "bar", "Bar")], disable=True)
+        self.assertTrue(res, "Stops HLS resolve procedure")
+        self.assertEqual(channel, "foo", "Doesn't switch channel")
+        self.assertEqual(channel_id, 1, "Doesn't switch channel id")
+        self.assertEqual(author, None, "Doesn't override author metadata")
+        self.assertEqual(mock_log.info.mock_calls, [
+            call("foo is hosting bar"),
+            call("hosting was disabled by command line option")
+        ])
+        self.assertEqual(mock_log.error.mock_calls, [], "Doesn't log anything to error")
+
+    def test_hosting_host_multiple(self, mock_log):
+        res, channel, channel_id, author = self.subject("foo", [
+            (1, 2, "bar", "Bar"),
+            (2, 3, "baz", "Baz"),
+            (3, 4, "qux", "Qux"),
+            (4, None, None, None)
+        ])
+        self.assertFalse(res, "Doesn't stop HLS resolve procedure")
+        self.assertEqual(channel, "qux", "Switches channel")
+        self.assertEqual(channel_id, 4, "Switches channel id")
+        self.assertEqual(author, "Qux", "Overrides author metadata")
+        self.assertEqual(mock_log.info.mock_calls, [
+            call("foo is hosting bar"),
+            call("switching to bar"),
+            call("bar is hosting baz"),
+            call("switching to baz"),
+            call("baz is hosting qux"),
+            call("switching to qux")
+        ])
+        self.assertEqual(mock_log.error.mock_calls, [], "Doesn't log anything to error")
+
+    def test_hosting_host_multiple_loop(self, mock_log):
+        res, channel, channel_id, author = self.subject("foo", [
+            (1, 2, "bar", "Bar"),
+            (2, 3, "baz", "Baz"),
+            (3, 1, "foo", "Foo")
+        ])
+        self.assertTrue(res, "Stops HLS resolve procedure")
+        self.assertEqual(channel, "baz", "Has switched channel")
+        self.assertEqual(channel_id, 3, "Has switched channel id")
+        self.assertEqual(author, "Baz", "Has overridden author metadata")
+        self.assertEqual(mock_log.info.mock_calls, [
+            call("foo is hosting bar"),
+            call("switching to bar"),
+            call("bar is hosting baz"),
+            call("switching to baz"),
+            call("baz is hosting foo")
+        ])
+        self.assertEqual(mock_log.error.mock_calls, [
+            call("A loop of hosted channels has been detected, cannot find a playable stream. (foo -> bar -> baz -> foo)")
+        ])
+
+
 @patch("streamlink.plugins.twitch.log")
 class TestTwitchReruns(unittest.TestCase):
     log_call = call("Reruns were disabled by command line option")
 
-    class StopError(Exception):
-        """Stop when trying to get an access token in _get_hls_streams..."""
-
-    @patch("streamlink.plugins.twitch.Twitch._check_for_host", return_value=None)
-    @patch("streamlink.plugins.twitch.Twitch._access_token", side_effect=StopError())
-    def start(self, *mocked, **params):
+    def subject(self, **params):
         with requests_mock.Mocker() as mock:
-            mocked_users = mock.get(
-                "https://api.twitch.tv/kraken/users.json?login=foo",
+            mock.get(
+                "https://api.twitch.tv/kraken/users?login=foo",
                 json={"users": [{"_id": 1234}]}
             )
-            mocked_stream = mock.get(
-                "https://api.twitch.tv/kraken/streams/1234.json",
+            mock.get(
+                "https://api.twitch.tv/kraken/streams/1234",
                 json={"stream": None} if params.pop("offline", False) else {"stream": {
                     "stream_type": params.pop("stream_type", "live"),
                     "broadcast_platform": params.pop("broadcast_platform", "live"),
@@ -380,66 +491,33 @@ class TestTwitchReruns(unittest.TestCase):
             Twitch.bind(session, "tests.plugins.test_twitch")
             plugin = Twitch("https://www.twitch.tv/foo")
             plugin.options.set("disable-reruns", params.pop("disable", True))
-            try:
-                streams = plugin.streams()
-            except TestTwitchReruns.StopError:
-                streams = True
-                pass
 
-            return streams, mocked_users, mocked_stream, mocked[0]
+            return plugin._check_for_rerun()
 
     def test_disable_reruns_live(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start()
-        self.assertTrue(api_users.called_once)
-        self.assertTrue(api_stream.called_once)
-        self.assertTrue(access_token.called_once)
-        self.assertTrue(streams)
+        self.assertFalse(self.subject())
         self.assertNotIn(self.log_call, mock_log.info.call_args_list)
 
     def test_disable_reruns_not_live(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start(stream_type="rerun")
-        self.assertTrue(api_users.called_once)
-        self.assertTrue(api_stream.called_once)
-        self.assertFalse(access_token.called)
-        self.assertDictEqual(streams, {})
+        self.assertTrue(self.subject(stream_type="rerun"))
         self.assertIn(self.log_call, mock_log.info.call_args_list)
 
     def test_disable_reruns_mixed(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start(stream_type="rerun", broadcast_platform="live")
-        self.assertTrue(api_users.called_once)
-        self.assertTrue(api_stream.called_once)
-        self.assertFalse(access_token.called)
-        self.assertDictEqual(streams, {})
+        self.assertTrue(self.subject(stream_type="rerun", broadcast_platform="live"))
         self.assertIn(self.log_call, mock_log.info.call_args_list)
 
     def test_disable_reruns_mixed2(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start(stream_type="live", broadcast_platform="rerun")
-        self.assertTrue(api_users.called_once)
-        self.assertTrue(api_stream.called_once)
-        self.assertFalse(access_token.called)
-        self.assertDictEqual(streams, {})
+        self.assertTrue(self.subject(stream_type="live", broadcast_platform="rerun"))
         self.assertIn(self.log_call, mock_log.info.call_args_list)
 
     def test_disable_reruns_broadcaster_software(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start(broadcaster_software="watch_party_rerun")
-        self.assertTrue(api_users.called_once)
-        self.assertTrue(api_stream.called_once)
-        self.assertTrue(access_token.called_once)
-        self.assertDictEqual(streams, {})
+        self.assertTrue(self.subject(broadcaster_software="watch_party_rerun"))
         self.assertIn(self.log_call, mock_log.info.call_args_list)
 
     def test_disable_reruns_offline(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start(offline=True)
-        self.assertTrue(api_users.called_once)
-        self.assertTrue(api_stream.called_once)
-        self.assertTrue(access_token.called_once)
-        self.assertTrue(streams)
+        self.assertFalse(self.subject(offline=True))
         self.assertNotIn(self.log_call, mock_log.info.call_args_list)
 
     def test_enable_reruns(self, mock_log):
-        streams, api_users, api_stream, access_token = self.start(disable=False)
-        self.assertFalse(api_users.called)
-        self.assertFalse(api_stream.called)
-        self.assertTrue(access_token.called_once)
-        self.assertTrue(streams)
+        self.assertFalse(self.subject(stream_type="rerun", disable=False))
         self.assertNotIn(self.log_call, mock_log.info.call_args_list)
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index 37381d5c..36eeb1fb 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -164,6 +164,7 @@ class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
 
     def subject(self, *args, **kwargs):
         thread, _ = super(TestHlsPlaylistReloadTime, self).subject(*args, **kwargs)
+        self.await_read(read_all=True)
 
         return thread.reader.worker.playlist_reload_time