Repository: streamlink/streamlink

Issue:
HLS: New command to use stream=True from requests for segments
## Feature Request

<!-- Replace [ ] with [x] in order to check the box -->
- [x] This is a feature request and I have read the contribution guidelines.


### Description

Add an option to use `stream=True` for HLS Streams Stream

https://github.com/streamlink/streamlink/blob/cd6f94a5a4ee83cb5b61a295df9e13956bcfcb27/src/streamlink/stream/hls.py#L106-L110

Example `--hls-segment-stream-data` or another name
which will turn `stream=True` or `stream=False`

### **HLSStreams**

It was already added in https://github.com/streamlink/streamlink/pull/260
but removed in https://github.com/streamlink/streamlink/pull/522 because it had some issues.

it might be better with a command where you can turn it on and off

_there might be still some issues with it, but it can be ignored see below for an example_

basically required for Twitch low latency https://github.com/streamlink/streamlink/issues/1676
to get the same delay as the browser or better, you can't get the same time as the browser without _stream=True_

### Expected / Actual behavior

New command to use `stream=True` from requests for HLS segments as an optional option.

### Example Code

_ChunkedEncodingError_ might skip a segment but the latency should be more important for some livestreams than a missing frame,
also this issue is very rare.

this code is only used for normal hls streams

```diff
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index 8a67f564..fecc54f4 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -4,6 +4,7 @@ import struct
 
 from collections import defaultdict, namedtuple
 from Crypto.Cipher import AES
+from requests.exceptions import ChunkedEncodingError
 
 from streamlink.exceptions import StreamError
 from streamlink.stream import hls_playlist
@@ -45,6 +46,8 @@ class HLSStreamWriter(SegmentedStreamWriter):
         self.byterange_offsets = defaultdict(int)
         self.key_data = None
         self.key_uri = None
+        self.stream_data = options.get("hls-segment-stream-data")
+
         if self.ignore_names:
             # creates a regex from a list of segment names,
             # this will be used to ignore segments.
@@ -104,6 +107,8 @@ class HLSStreamWriter(SegmentedStreamWriter):
                 return
 
             return self.session.http.get(sequence.segment.uri,
+                                         stream=(self.stream_data
+                                                 and not sequence.segment.key),
                                          timeout=self.timeout,
                                          exception=StreamError,
                                          retries=self.retries,
@@ -134,8 +139,12 @@ class HLSStreamWriter(SegmentedStreamWriter):
 
             self.reader.buffer.write(pkcs7_decode(decrypted_chunk))
         else:
-            for chunk in res.iter_content(chunk_size):
-                self.reader.buffer.write(chunk)
+            try:
+                for chunk in res.iter_content(chunk_size):
+                    self.reader.buffer.write(chunk)
+            except ChunkedEncodingError:
+                log.error("Download of segment {0} failed", sequence.num)
+                return
 
         log.debug("Download of segment {0} complete", sequence.num)
```

---

**EDIT**: Removed DASH, HLS is enough

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/streamlink/streamlink/pull/2513?src=pr&el=h1) Report
> Merging [#2513](https://codecov.io/gh/streamlink/streamlink/pull/2513?src=pr&el=desc) into [master](https://codecov.io/gh/streamlink/streamlink/commit/209e43186b6bcf34c58f06019e8491f7ee2587d7?src=pr&el=desc) will **increase** coverage by `<.01%`.
> The diff coverage is `63.88%`.


```diff
@@            Coverage Diff             @@
##           master    #2513      +/-   ##
==========================================
+ Coverage   52.53%   52.54%   +<.01%     
==========================================
  Files         239      239              
  Lines       15070    15100      +30     
==========================================
+ Hits         7917     7934      +17     
- Misses       7153     7166      +13
```

bastimeyer: - added `hls-segment-stream-data` to the Streamlink session options
- moved the `disable_ads` and `low_latency` properties to the TwitchHLSStream class, so that the other classes can access them via the `stream` property
- passed the `stream` instance to the TwitchM3U8Parser
- added HLS tag parsing conditions (as simple decorators)
- reverted the aggressive playlist reloading behavior when `low_latency` is not set
- improved tests and added tests for low latency logic
- changed / fixed plugin parameter description
bastimeyer: Btw, while I was adding the tests, I noticed that the plugin parameters do not get initialized properly and I had to set them explicitly. Looks like they only get initialized by streamlink_cli here:
https://github.com/streamlink/streamlink/blob/1.1.1/src/streamlink_cli/main.py#L888-L914
beardypig: I think this will do for now :) I have, what _I_ think is, a better solution in the pipeline, but it is a little way off :)
beardypig: One note: why not override `HLSStreamWorker.valid_sequence` instead of introducing the complexity of `parse_condition`? :) 
bastimeyer: > HLSStreamWorker.valid_sequence

Ideally, this should have been done in the disable-ads PR earlier this year, so that [`TwitchHLSStreamWriter.write`](https://github.com/streamlink/streamlink/pull/2513/files#diff-195bebeb65dac1f6a8e94f1957a95ec1L187) didn't need to be implemented. With these changes though, it doesn't make sense rewriting the entire thing again, unless you want it to be more complex.

Regarding the decorator, I've added the simple if statement to the prefetch tag method first, but it made sense to disable these advertisement tags too when the other parameter is not set, so adding a decorator made things look more simple, as all methods share this same logic. Btw, using `HLSStreamWorker.valid_sequence` to figure out whether the prefetch segments should be used or not would mean that they would always have to be created by the parser first and then discarded by the worker. A bit unnecessary.

----

I just wanted to get things finally done here with this PR. I know that most changes are not implemented the best way, but doing it right would require a much deeper refactoring of the entire thing and you already said that you had been working on something better a few months ago.

Let me fix the super class method call, though, before this gets merged.
beardypig: Fair enough @bastimeyer, I think I was out of the loop on those changes, or not paying close enough attention.
I think the parser should parse all of the segments regardless, it should only have one job - parse, now it has a dependency on the stream. If the worker wants to discard a segment that should be its responsibility, I personally think it makes it more difficult to understand  (and less testable) to delegate to the parser, it might be slightly more efficient - but at what cost? Won’t someone think of the children!? 

Hopefully the work I’m doing on the stream classes will make this kind of thing easier. 
bastimeyer: Any further concerns, comments or suggestions?

If refactoring the plugin and removing the stream dependency from the parser is that important, then tell me, and I will change it. Or just push the necessary changes to my branch yourself :)
beardypig: As it's the twitch specific parser, I think it's OK. In my WIP branch, I have not used the `stream` (https://github.com/beardypig/streamlink/blob/stream-latency/src/streamlink/plugins/twitch.py). I haven't managed to get the latency down to faster than the browser though... What `mpv` cache settings did you use? And did you notice anything weird when reloading the playlist? I seem to get one prefetch segment in the first load, then in the next reload I get that previous prefetch segment as completed one, plus another completed one, and then more prefetch segments... 
bastimeyer: > What `mpv` cache settings did you use?

This depends on the value of `hls-live-edge`. If set to `2`, then [`--cache=no`](https://mpv.io/manual/master/#options-cache) is enough to not make it stutter (MPV's demuxer will still cache one additional second). If `hls-live-edge` is set to `1`, then you need to increase MPV's cache a bit by 0.5 secs or so via [`--demuxer-readahead-secs=1.5`](https://mpv.io/manual/master/#options-demuxer-readahead-secs) or [`--cache-secs`](https://mpv.io/manual/master/#options-cache-secs).

> did you notice anything weird when reloading the playlist?

Not really. Each playlist lists two prefetch segments. One of them is the previously completed one (live edge 1), and in the case that the reload/download takes too long, then the two prefetch segments are new ones.

I think using `get_segment` in your code is not the correct choice here, because the prefetch segments don't have a state which you can pop off the stack. That's why the last regular segment gets cloned for each prefetch segment in my PR.
beardypig: Thanks for the `mpv` example. I’ll took again how `get_segment` and the state work. I need to look at more playlists :)
bastimeyer: I've been thinking a little bit about the ads filtering, and I realized that we will probably have to make both parameters mutually exclusive.

As I've said, I don't get any ads on Twitch, so I can't test and verify this, but the way the low latency stuff is currently implemented is that it clones the last regular segment data, replaces the URL with the additional prefetch segment URLs and adds these new segments to the sequence as regular ones. This means that we're also shifting the live-edge here and since we're limiting the live-edge value to 1 or 2 when the parameter gets set, we don't have proper metadata to work with when it comes to ads filtering. This means that once an ad appears in the regular segments, we've already processed it as a prefetch segment without metadata and we're shifting the start/end of an ad-block by the amount of prefetch tags, which is two, or in other words 4 seconds. Since players like VLC don't handle discontinuities well, this breaks the whole ads-filtering implementation and is a problem.

Also, another thing I've realized is that we're also allowing low latency streaming for regular non-low-latency streams without prefetch segments. This isn't wrong per se, as it enables `--hls-segment-stream-data` and thus improves the latency by at most 2 seconds, but the log output implies that it's a low latency stream on Twitch's end, which isn't true. The log output needs to be changed, so that the user knows whether it's a true low latency stream or not. There's no information available on the Twitch API, so it has to be done by parsing the HLS playlist and checking whether it includes prefetch data.
thinkpad4: Has this been added to the nightly build?  I tried a nightly from a week ago and couldn't get the --twitch-low-latency parameter to work.
bastimeyer: This is an open pull request, so of course not. If you want to try it, install it from the branch of this PR. See https://github.com/streamlink/streamlink/issues/2402#issuecomment-480455258
back-to: - should this be merged **or** do you want to make more changes?
- should we wait for the updated HLSStream version? (which will also take longer)
bastimeyer: > should this be merged **or** do you want to make more changes?

I haven't had the time and motivation to change the things mentioned in my previous comment:
https://github.com/streamlink/streamlink/pull/2513#issuecomment-513464721

I think this should be done before this PR can be merged.

> should we wait for the updated HLSStream version?

No, not unless the features of this PR are in conflict with the promised improvements (`--hls-segment-stream-data` parameter and low latency streaming behind the `--twitch-low-latency` parameter).
beardypig: My PR would make `--hls-segment-stream-data` obsolete, it would always apply, other than that there wouldn't be so much of a conflict. 
kuon: Do you think you might merge this in? I would love to have this feature built in.
bastimeyer: As you can see since my last comment, no further changes have been added yet and the PR is still in the same WIP state.
https://github.com/streamlink/streamlink/pull/2513#issuecomment-518247628
kuon: My view on both issue:

- Making this option mutually exclusive with ad filtering is fine. I see making them work together as a separate task as the problem seems non trivial.
- I think we should not over engineer the detection of low latency, adding a note in the option documentation about this option not working with non low latency twitch stream is fine.
thinkpad4: I would love to see this finally added
axm2: Would also like to see this merged
thinkpad4: Any update on this getting pushed?
bastimeyer: Locking this PR now.

I've already said multiple times that it's a WIP and not finished yet. This hasn't changed, so please stop asking, this is just unnecessary noise and not contributing anything. As you can see on my Github profile, I didn't have the time recently to work on any of my projects or pull requests.

I will see if I can add the necessary changes this weekend, but I'm not sure.

There's also @beardypig with his implementation, but I don't know anything about the status of that.

----

**What needs to be done?**

- [ ] Make ad filtering mutually exclusive with low latency streaming
- [ ] Disable low latency streaming if it's not a low latency stream
- [ ] Add proper documentation
bastimeyer: I've rebased the branch to 1.3.0. This doesn't mean though that there's anything new or that I'm confident getting this merged any time soon.

If you're testing the low latency branch, you should reinstall it. Btw, the next Twitch GUI version will require 1.3.0, so if you're using that, you will have to upgrade.
bastimeyer: I've force-pushed a rebase from Streamlink's current master branch, so that the MPV fixes are now included and merge conflicts solved. Don't merge though, nothing has changed.
back-to: > Regarding ad filtering and additional segment metadata, is it the right approach while prefetching segments to clone the last segment of the sequence and replace its URL?

if you could detect the Ad URLs they could be ignored

as you are using the prefetch URLs **there shouldn't be an issue**,
I don't think there are Ad URLs in the prefetch URLs.

but this is only testable with a larger user base

> Add proper documentation

nice to have, but not important for testing purposes

---

**i would merge this PR** and see what issues occupy on the fly.
bastimeyer: > I don't think there are Ad URLs in the prefetch URLs.

Hm, that is actually true.

> Add proper documentation
> **i would merge this PR** and see what issues occupy on the fly.

Even if the `--twitch-low-latency` parameter gets set, there is still some tweaking to do by the user on the player side, otherwise the benefits are minimal. I think that should be mentioned at least. Also the live-edge value, which is all tied to it. Not sure if an example parameter string would be a good idea, because it depends on the player that's being used. I also haven't looked into VLC parameters yet, which is problematic, because it's Streamlink's default player. Let me improve the parameter description at least a bit though.

After I've rebased the branch and fixed the merge conflicts, then I think we can merge this. In another thread somewhere here I've said that I don't really feel comfortable with the changes, because there are still a couple of issues with occasional hiccups, but to be honest, I also don't want this PR to be open and stale anymore. Merging this will probably lead to lots of further questions, but so be it then.

Patch:
diff --git a/src/streamlink/plugins/twitch.py b/src/streamlink/plugins/twitch.py
index 7b0859e1..f890b390 100644
--- a/src/streamlink/plugins/twitch.py
+++ b/src/streamlink/plugins/twitch.py
@@ -152,18 +152,50 @@ _quality_options_schema = validate.Schema(
 
 Segment = namedtuple("Segment", "uri duration title key discontinuity scte35 byterange date map")
 
+LOW_LATENCY_MAX_LIVE_EDGE = 2
+
+
+def parse_condition(attr):
+    def wrapper(func):
+        def method(self, *args, **kwargs):
+            if hasattr(self.stream, attr) and getattr(self.stream, attr, False):
+                func(self, *args, **kwargs)
+        return method
+    return wrapper
+
 
 class TwitchM3U8Parser(M3U8Parser):
+    def __init__(self, base_uri=None, stream=None, **kwargs):
+        M3U8Parser.__init__(self, base_uri, **kwargs)
+        self.stream = stream
+        self.has_prefetch_segments = False
+
+    def parse(self, *args):
+        m3u8 = super(TwitchM3U8Parser, self).parse(*args)
+        m3u8.has_prefetch_segments = self.has_prefetch_segments
+
+        return m3u8
+
+    @parse_condition("disable_ads")
     def parse_tag_ext_x_scte35_out(self, value):
         self.state["scte35"] = True
 
     # unsure if this gets used by Twitch
+    @parse_condition("disable_ads")
     def parse_tag_ext_x_scte35_out_cont(self, value):
         self.state["scte35"] = True
 
+    @parse_condition("disable_ads")
     def parse_tag_ext_x_scte35_in(self, value):
         self.state["scte35"] = False
 
+    @parse_condition("low_latency")
+    def parse_tag_ext_x_twitch_prefetch(self, value):
+        self.has_prefetch_segments = True
+        segments = self.m3u8.segments
+        if segments:
+            segments.append(segments[-1]._replace(uri=self.uri(value)))
+
     def get_segment(self, uri):
         byterange = self.state.pop("byterange", None)
         extinf = self.state.pop("extinf", (0, None))
@@ -188,19 +220,24 @@ class TwitchM3U8Parser(M3U8Parser):
 
 class TwitchHLSStreamWorker(HLSStreamWorker):
     def _reload_playlist(self, text, url):
-        return load_hls_playlist(text, url, parser=TwitchM3U8Parser)
+        return load_hls_playlist(text, url, parser=TwitchM3U8Parser, stream=self.stream)
 
+    def _set_playlist_reload_time(self, playlist, sequences):
+        if not self.stream.low_latency:
+            super(TwitchHLSStreamWorker, self)._set_playlist_reload_time(playlist, sequences)
+        else:
+            self.playlist_reload_time = sequences[-1].segment.duration
+
+    def process_sequences(self, playlist, sequences):
+        if self.playlist_sequence < 0 and self.stream.low_latency and not playlist.has_prefetch_segments:
+            log.info("This is not a low latency stream")
+
+        return super(TwitchHLSStreamWorker, self).process_sequences(playlist, sequences)
 
-class TwitchHLSStreamWriter(HLSStreamWriter):
-    def __init__(self, *args, **kwargs):
-        HLSStreamWriter.__init__(self, *args, **kwargs)
-        options = self.session.plugins.get("twitch").options
-        self.disable_ads = options.get("disable-ads")
-        if self.disable_ads:
-            log.info("Will skip ad segments")
 
+class TwitchHLSStreamWriter(HLSStreamWriter):
     def write(self, sequence, *args, **kwargs):
-        if self.disable_ads:
+        if self.stream.disable_ads:
             if sequence.segment.scte35 is not None:
                 self.reader.ads = sequence.segment.scte35
                 if self.reader.ads:
@@ -219,12 +256,38 @@ class TwitchHLSStreamReader(HLSStreamReader):
 
 
 class TwitchHLSStream(HLSStream):
+    def __init__(self, *args, **kwargs):
+        HLSStream.__init__(self, *args, **kwargs)
+
+        disable_ads = self.session.get_plugin_option("twitch", "disable-ads")
+        low_latency = self.session.get_plugin_option("twitch", "low-latency")
+
+        if low_latency and disable_ads:
+            log.info("Low latency streaming with ad filtering is currently not supported")
+            self.session.set_plugin_option("twitch", "low-latency", False)
+            low_latency = False
+
+        self.disable_ads = disable_ads
+        self.low_latency = 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))
+
         reader = TwitchHLSStreamReader(self)
         reader.open()
 
         return reader
 
+    @classmethod
+    def _get_variant_playlist(cls, res):
+        return load_hls_playlist(res.text, base_uri=res.url)
+
 
 class UsherService(object):
     def __init__(self, session):
@@ -353,47 +416,70 @@ class TwitchAPI(object):
 
 class Twitch(Plugin):
     arguments = PluginArguments(
-        PluginArgument("oauth-token",
-                       sensitive=True,
-                       metavar="TOKEN",
-                       help="""
-        An OAuth token to use for Twitch authentication.
-        Use --twitch-oauth-authenticate to create a token.
-        """),
-        PluginArgument("cookie",
-                       sensitive=True,
-                       metavar="COOKIES",
-                       help="""
-        Twitch cookies to authenticate to allow access to subscription channels.
-
-        Example:
-
-          "_twitch_session_id=xxxxxx; persistent=xxxxx"
-
-        Note: This method is the old and clunky way of authenticating with
-        Twitch, using --twitch-oauth-authenticate is the recommended and
-        simpler way of doing it now.
-        """
-                       ),
-        PluginArgument("disable-hosting",
-                       action="store_true",
-                       help="""
-        Do not open the stream if the target channel is hosting another channel.
-        """
-                       ),
-        PluginArgument("disable-ads",
-                       action="store_true",
-                       help="""
-        Skip embedded advertisement segments at the beginning or during a stream.
-        Will cause these segments to be missing from the stream.
-        """
-                       ),
-        PluginArgument("disable-reruns",
-                       action="store_true",
-                       help="""
-        Do not open the stream if the target channel is currently broadcasting a rerun.
-        """
-                       ))
+        PluginArgument(
+            "oauth-token",
+            sensitive=True,
+            metavar="TOKEN",
+            help="""
+            An OAuth token to use for Twitch authentication.
+            Use --twitch-oauth-authenticate to create a token.
+            """
+        ),
+        PluginArgument(
+            "cookie",
+            sensitive=True,
+            metavar="COOKIES",
+            help="""
+            Twitch cookies to authenticate to allow access to subscription channels.
+
+            Example:
+
+              "_twitch_session_id=xxxxxx; persistent=xxxxx"
+
+            Note: This method is the old and clunky way of authenticating with
+            Twitch, using --twitch-oauth-authenticate is the recommended and
+            simpler way of doing it now.
+            """
+        ),
+        PluginArgument(
+            "disable-hosting",
+            action="store_true",
+            help="""
+            Do not open the stream if the target channel is hosting another channel.
+            """
+        ),
+        PluginArgument(
+            "disable-ads",
+            action="store_true",
+            help="""
+            Skip embedded advertisement segments at the beginning or during a stream.
+            Will cause these segments to be missing from the stream.
+            """
+        ),
+        PluginArgument(
+            "disable-reruns",
+            action="store_true",
+            help="""
+            Do not open the stream if the target channel is currently broadcasting a rerun.
+            """
+        ),
+        PluginArgument(
+            "low-latency",
+            action="store_true",
+            help="""
+            Enables low latency streaming by prefetching HLS segments.
+            Sets --hls-segment-stream-data to true and --hls-live-edge to {live_edge}, if it is higher.
+            Reducing --hls-live-edge to 1 will result in the lowest latency possible.
+
+            Low latency streams have to be enabled by the broadcasters on Twitch themselves.
+            Regular streams can cause buffering issues with this option enabled.
+
+            Note: The caching/buffering settings of the chosen player may need to be adjusted as well.
+            Please refer to the player's own documentation for the required parameters and its configuration.
+            Player parameters can be set via Streamlink's --player or --player-args parameters.
+            """.format(live_edge=LOW_LATENCY_MAX_LIVE_EDGE)
+        )
+    )
 
     @classmethod
     def stream_weight(cls, key):
diff --git a/src/streamlink/session.py b/src/streamlink/session.py
index 93e89b5f..ea554f6c 100644
--- a/src/streamlink/session.py
+++ b/src/streamlink/session.py
@@ -58,6 +58,7 @@ class Streamlink(object):
             "hls-segment-attempts": 3,
             "hls-segment-threads": 1,
             "hls-segment-timeout": 10.0,
+            "hls-segment-stream-data": False,
             "hls-timeout": 60.0,
             "hls-playlist-reload-attempts": 3,
             "hls-start-offset": 0,
@@ -131,6 +132,9 @@ class Streamlink(object):
         hls-segment-threads      (int) The size of the thread pool used
                                  to download segments, default: ``1``
 
+        hls-segment-stream-data  (bool) Stream HLS segment downloads,
+                                 default: ``False``
+
         hls-segment-timeout      (float) HLS segment connect and read
                                  timeout, default: ``10.0``
 
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index 721f0139..ef9d0312 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -4,6 +4,7 @@ import struct
 
 from collections import defaultdict, namedtuple
 from Crypto.Cipher import AES
+from requests.exceptions import ChunkedEncodingError
 
 from streamlink.exceptions import StreamError
 from streamlink.stream import hls_playlist
@@ -46,6 +47,7 @@ class HLSStreamWriter(SegmentedStreamWriter):
         self.key_data = None
         self.key_uri = None
         self.key_uri_override = options.get("hls-segment-key-uri")
+        self.stream_data = options.get("hls-segment-stream-data")
 
         if self.ignore_names:
             # creates a regex from a list of segment names,
@@ -109,6 +111,8 @@ class HLSStreamWriter(SegmentedStreamWriter):
                 return
 
             return self.session.http.get(sequence.segment.uri,
+                                         stream=(self.stream_data
+                                                 and not sequence.segment.key),
                                          timeout=self.timeout,
                                          exception=StreamError,
                                          retries=self.retries,
@@ -139,8 +143,13 @@ class HLSStreamWriter(SegmentedStreamWriter):
 
             self.reader.buffer.write(pkcs7_decode(decrypted_chunk))
         else:
-            for chunk in res.iter_content(chunk_size):
-                self.reader.buffer.write(chunk)
+            try:
+                for chunk in res.iter_content(chunk_size):
+                    self.reader.buffer.write(chunk)
+            except ChunkedEncodingError:
+                log.error("Download of segment {0} failed", sequence.num)
+
+                return
 
         log.debug("Download of segment {0} complete", sequence.num)
 
@@ -213,6 +222,10 @@ class HLSStreamWorker(SegmentedStreamWorker):
         if sequences:
             self.process_sequences(playlist, sequences)
 
+    def _set_playlist_reload_time(self, playlist, sequences):
+        self.playlist_reload_time = (playlist.target_duration
+                                     or sequences[-1].segment.duration)
+
     def process_sequences(self, playlist, sequences):
         first_sequence, last_sequence = sequences[0], sequences[-1]
 
@@ -220,7 +233,7 @@ class HLSStreamWorker(SegmentedStreamWorker):
             log.debug("Segments in this playlist are encrypted")
 
         self.playlist_changed = ([s.num for s in self.playlist_sequences] != [s.num for s in sequences])
-        self.playlist_reload_time = (playlist.target_duration or last_sequence.segment.duration)
+        self._set_playlist_reload_time(playlist, sequences)
         self.playlist_sequences = sequences
 
         if not self.playlist_changed:
@@ -352,6 +365,10 @@ class HLSStream(HTTPStream):
 
         return reader
 
+    @classmethod
+    def _get_variant_playlist(cls, res):
+        return hls_playlist.load(res.text, base_uri=res.url)
+
     @classmethod
     def parse_variant_playlist(cls, session_, url, name_key="name",
                                name_prefix="", check_streams=False,
@@ -378,7 +395,7 @@ class HLSStream(HTTPStream):
         res = session_.http.get(url, exception=IOError, **request_params)
 
         try:
-            parser = hls_playlist.load(res.text, base_uri=res.url)
+            parser = cls._get_variant_playlist(res)
         except ValueError as err:
             raise IOError("Failed to parse playlist: {0}".format(err))
 
diff --git a/src/streamlink_cli/argparser.py b/src/streamlink_cli/argparser.py
index 4071a1c9..40ea303b 100644
--- a/src/streamlink_cli/argparser.py
+++ b/src/streamlink_cli/argparser.py
@@ -743,6 +743,13 @@ def build_parser():
         Default is 3.
         """
     )
+    transport.add_argument(
+        "--hls-segment-stream-data",
+        action="store_true",
+        help="""
+        Immediately write segment data into output buffer while downloading.
+        """
+    )
     transport.add_argument(
         "--hls-segment-attempts",
         type=num(int, min=0),
diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py
index a8586e24..d7d93e0f 100644
--- a/src/streamlink_cli/main.py
+++ b/src/streamlink_cli/main.py
@@ -791,6 +791,9 @@ def setup_options():
     if args.hls_live_edge:
         streamlink.set_option("hls-live-edge", args.hls_live_edge)
 
+    if args.hls_segment_stream_data:
+        streamlink.set_option("hls-segment-stream-data", args.hls_segment_stream_data)
+
     if args.hls_segment_attempts:
         streamlink.set_option("hls-segment-attempts", args.hls_segment_attempts)
 


Test patch:
diff --git a/tests/plugins/test_twitch.py b/tests/plugins/test_twitch.py
index 3c74d2db..9fb9b9ae 100644
--- a/tests/plugins/test_twitch.py
+++ b/tests/plugins/test_twitch.py
@@ -35,16 +35,21 @@ class TestPluginTwitch(unittest.TestCase):
 
 
 class TestTwitchHLSStream(unittest.TestCase):
+    url_master = "http://mocked/path/master.m3u8"
+    url_playlist = "http://mocked/path/playlist.m3u8"
+    url_segment = "http://mocked/path/stream{0}.ts"
+
     scte35_out = "#EXT-X-DISCONTINUITY\n#EXT-X-SCTE35-OUT\n"
     scte35_out_cont = "#EXT-X-SCTE35-OUT-CONT\n"
     scte35_in = "#EXT-X-DISCONTINUITY\n#EXT-X-SCTE35-IN\n"
     segment = "#EXTINF:1.000,\nstream{0}.ts\n"
+    prefetch = "#EXT-X-TWITCH-PREFETCH:{0}\n"
 
     def getMasterPlaylist(self):
         with text("hls/test_master.m3u8") as pl:
             return pl.read()
 
-    def getPlaylist(self, media_sequence, items):
+    def getPlaylist(self, media_sequence, items, prefetch=None):
         playlist = """
 #EXTM3U
 #EXT-X-VERSION:5
@@ -57,39 +62,39 @@ class TestTwitchHLSStream(unittest.TestCase):
                 playlist += item
             else:
                 playlist += self.segment.format(item)
+        for item in prefetch or []:
+            playlist += self.prefetch.format(self.url_segment.format(item))
 
         return playlist
 
-    def start_streamlink(self, kwargs=None):
+    def start_streamlink(self, disable_ads=False, low_latency=False, kwargs=None):
         kwargs = kwargs or {}
         log.info("Executing streamlink")
         streamlink = Streamlink()
 
         streamlink.set_option("hls-live-edge", 4)
-        streamlink.plugins.get("twitch").options.set("disable-ads", True)
+        streamlink.set_plugin_option("twitch", "disable-ads", disable_ads)
+        streamlink.set_plugin_option("twitch", "low-latency", low_latency)
 
-        masterStream = TwitchHLSStream.parse_variant_playlist(
-            streamlink,
-            "http://mocked/path/master.m3u8",
-            **kwargs
-        )
+        masterStream = TwitchHLSStream.parse_variant_playlist(streamlink, self.url_master, **kwargs)
         stream = masterStream["1080p (source)"].open()
         data = b"".join(iter(partial(stream.read, 8192), b""))
         stream.close()
         log.info("End of streamlink execution")
-        return data
+        return streamlink, data
 
     def mock(self, mocked, method, url, *args, **kwargs):
         mocked[url] = method(url, *args, **kwargs)
 
-    def get_result(self, streams, playlists):
+    def get_result(self, streams, playlists, **kwargs):
         mocked = {}
         with requests_mock.Mocker() as mock:
-            self.mock(mocked, mock.get, "http://mocked/path/master.m3u8", text=self.getMasterPlaylist())
-            self.mock(mocked, mock.get, "http://mocked/path/playlist.m3u8", [{"text": p} for p in playlists])
+            self.mock(mocked, mock.get, self.url_master, text=self.getMasterPlaylist())
+            self.mock(mocked, mock.get, self.url_playlist, [{"text": p} for p in playlists])
             for i, stream in enumerate(streams):
-                self.mock(mocked, mock.get, "http://mocked/path/stream{0}.ts".format(i), content=stream)
-            return self.start_streamlink(), mocked
+                self.mock(mocked, mock.get, self.url_segment.format(i), content=stream)
+            streamlink, data = self.start_streamlink(**kwargs)
+            return streamlink, data, mocked
 
     @patch("streamlink.plugins.twitch.log")
     def test_hls_scte35_start_with_end(self, mock_logging):
@@ -99,12 +104,12 @@ class TestTwitchHLSStream(unittest.TestCase):
             self.getPlaylist(4, [self.scte35_in, 4, 5, 6, 7]),
             self.getPlaylist(8, [8, 9, 10, 11]) + "#EXT-X-ENDLIST\n"
         ]
-        result, mocked = self.get_result(streams, playlists)
+        streamlink, result, mocked = self.get_result(streams, playlists, disable_ads=True)
 
         expected = b''.join(streams[4:12])
         self.assertEqual(expected, result)
-        for i, _ in enumerate(streams):
-            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        for i in range(0, 12):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
         mock_logging.info.assert_has_calls([
             call("Will skip ad segments"),
             call("Will skip ads beginning with segment 0"),
@@ -118,12 +123,12 @@ class TestTwitchHLSStream(unittest.TestCase):
             self.getPlaylist(0, [0, 1, 2, 3]),
             self.getPlaylist(4, [self.scte35_in, 4, 5, 6, 7]) + "#EXT-X-ENDLIST\n"
         ]
-        result, mocked = self.get_result(streams, playlists)
+        streamlink, result, mocked = self.get_result(streams, playlists, disable_ads=True)
 
         expected = b''.join(streams[0:8])
         self.assertEqual(expected, result)
-        for i, _ in enumerate(streams):
-            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        for i in range(0, 8):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
         mock_logging.info.assert_has_calls([
             call("Will skip ad segments")
         ])
@@ -135,12 +140,12 @@ class TestTwitchHLSStream(unittest.TestCase):
             self.getPlaylist(0, [self.scte35_out_cont, 0, 1, 2, 3]),
             self.getPlaylist(4, [self.scte35_in, 4, 5, 6, 7]) + "#EXT-X-ENDLIST\n"
         ]
-        result, mocked = self.get_result(streams, playlists)
+        streamlink, result, mocked = self.get_result(streams, playlists, disable_ads=True)
 
         expected = b''.join(streams[4:8])
         self.assertEqual(expected, result)
-        for i, _ in enumerate(streams):
-            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        for i in range(0, 8):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
         mock_logging.info.assert_has_calls([
             call("Will skip ad segments"),
             call("Will skip ads beginning with segment 0"),
@@ -155,12 +160,12 @@ class TestTwitchHLSStream(unittest.TestCase):
             self.getPlaylist(4, [self.scte35_out, 4, 5, 6, 7]),
             self.getPlaylist(8, [8, 9, 10, 11]) + "#EXT-X-ENDLIST\n"
         ]
-        result, mocked = self.get_result(streams, playlists)
+        streamlink, result, mocked = self.get_result(streams, playlists, disable_ads=True)
 
         expected = b''.join(streams[0:4])
         self.assertEqual(expected, result)
-        for i, _ in enumerate(streams):
-            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        for i in range(0, 12):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
         mock_logging.info.assert_has_calls([
             call("Will skip ad segments"),
             call("Will skip ads beginning with segment 4")
@@ -176,18 +181,111 @@ class TestTwitchHLSStream(unittest.TestCase):
             self.getPlaylist(12, [12, 13, self.scte35_in, 14, 15]),
             self.getPlaylist(16, [16, 17, 18, 19]) + "#EXT-X-ENDLIST\n"
         ]
-        result, mocked = self.get_result(streams, playlists)
+        streamlink, result, mocked = self.get_result(streams, playlists, disable_ads=True)
 
         expected = b''.join(streams[0:6]) + b''.join(streams[14:20])
         self.assertEqual(expected, result)
-        for i, _ in enumerate(streams):
-            self.assertTrue(mocked["http://mocked/path/stream{0}.ts".format(i)].called)
+        for i in range(0, 20):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
         mock_logging.info.assert_has_calls([
             call("Will skip ad segments"),
             call("Will skip ads beginning with segment 6"),
             call("Will stop skipping ads beginning with segment 14")
         ])
 
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_scte35_no_disable_ads(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(20)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3]),
+            self.getPlaylist(4, [4, 5, self.scte35_out, 6, 7]),
+            self.getPlaylist(8, [8, 9, 10, 11]),
+            self.getPlaylist(12, [12, 13, self.scte35_in, 14, 15]),
+            self.getPlaylist(16, [16, 17, 18, 19]) + "#EXT-X-ENDLIST\n"
+        ]
+        streamlink, result, mocked = self.get_result(streams, playlists)
+
+        expected = b''.join(streams[0:20])
+        self.assertEqual(expected, result)
+        for i in range(0, 20):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
+        mock_logging.info.assert_has_calls([])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_prefetch(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(10)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3], [4, 5]),
+            self.getPlaylist(4, [4, 5, 6, 7], [8, 9]) + "#EXT-X-ENDLIST\n"
+        ]
+        streamlink, result, mocked = self.get_result(streams, playlists, low_latency=True)
+
+        self.assertEqual(2, streamlink.options.get("hls-live-edge"))
+        self.assertEqual(True, streamlink.options.get("hls-segment-stream-data"))
+
+        expected = b''.join(streams[4:10])
+        self.assertEqual(expected, result)
+        for i in range(0, 3):
+            self.assertFalse(mocked[self.url_segment.format(i)].called, i)
+        for i in range(4, 9):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
+        mock_logging.info.assert_has_calls([
+            call("Low latency streaming (HLS live edge: 2)")
+        ])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_prefetch_no_low_latency(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(10)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3], [4, 5]),
+            self.getPlaylist(4, [4, 5, 6, 7], [8, 9]) + "#EXT-X-ENDLIST\n"
+        ]
+        streamlink, result, mocked = self.get_result(streams, playlists)
+
+        self.assertEqual(4, streamlink.options.get("hls-live-edge"))
+        self.assertEqual(False, streamlink.options.get("hls-segment-stream-data"))
+
+        expected = b''.join(streams[0:8])
+        self.assertEqual(expected, result)
+        for i in range(0, 7):
+            self.assertTrue(mocked[self.url_segment.format(i)].called, i)
+        for i in range(8, 9):
+            self.assertFalse(mocked[self.url_segment.format(i)].called, i)
+        mock_logging.info.assert_has_calls([])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_no_low_latency_with_disable_ads(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(10)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3], [4, 5]),
+            self.getPlaylist(4, [4, 5, 6, 7], [8, 9]) + "#EXT-X-ENDLIST\n"
+        ]
+        streamlink, result, mocked = self.get_result(streams, playlists, low_latency=True, disable_ads=True)
+
+        self.assertFalse(streamlink.get_plugin_option("twitch", "low-latency"))
+        self.assertTrue(streamlink.get_plugin_option("twitch", "disable-ads"))
+
+        mock_logging.info.assert_has_calls([
+            call("Low latency streaming with ad filtering is currently not supported")
+        ])
+
+    @patch("streamlink.plugins.twitch.log")
+    def test_hls_no_low_latency_no_prefetch(self, mock_logging):
+        streams = ["[{0}]".format(i).encode("ascii") for i in range(10)]
+        playlists = [
+            self.getPlaylist(0, [0, 1, 2, 3], []),
+            self.getPlaylist(4, [4, 5, 6, 7], []) + "#EXT-X-ENDLIST\n"
+        ]
+        streamlink, result, mocked = self.get_result(streams, playlists, low_latency=True)
+
+        self.assertTrue(streamlink.get_plugin_option("twitch", "low-latency"))
+        self.assertFalse(streamlink.get_plugin_option("twitch", "disable-ads"))
+
+        mock_logging.info.assert_has_calls([
+            call("Low latency streaming (HLS live edge: 2)"),
+            call("This is not a low latency stream")
+        ])
+
 
 @patch("streamlink.plugins.twitch.log")
 class TestTwitchReruns(unittest.TestCase):