Keep the HLS backlog even if the stream header changes.
[cubemap] / stream.cpp
index 2c7c42ed75154fbff911fcdcf53421c7dd7500e3..6a5ef99f78fa80a8294b6bf9d3bf8ee52d0bb1c0 100644 (file)
@@ -77,7 +77,7 @@ Stream::Stream(const StreamProto &serialized, int data_fd)
        }
 
        for (const FragmentStartProto &fragment : serialized.fragment()) {
-               fragments.push_back(FragmentStart { size_t(fragment.byte_position()), fragment.pts() });
+               fragments.push_back(FragmentStart { size_t(fragment.byte_position()), fragment.pts(), fragment.begins_header() });
        }
 }
 
@@ -96,6 +96,7 @@ StreamProto Stream::serialize()
                FragmentStartProto *proto = serialized.add_fragment();
                proto->set_byte_position(fragment.byte_position);
                proto->set_pts(fragment.pts);
+               proto->set_begins_header(fragment.begins_header);
        }
        serialized.set_first_fragment_index(first_fragment_index);
        serialized.set_discontinuity_counter(discontinuity_counter);
@@ -151,6 +152,45 @@ void Stream::set_backlog_size(size_t new_size)
        remove_obsolete_starting_points();
 }
 
+void Stream::set_header(const std::string &new_http_header, const std::string &new_stream_header)
+{
+       http_header = new_http_header;
+       if (new_stream_header == stream_header) {
+               return;
+       }
+
+       // We cannot start at any of the older starting points anymore,
+       // since they'd get the wrong header for the stream (not to mention
+       // that a changed header probably means the stream restarted,
+       // which means any client starting on the old one would probably
+       // stop playing properly at the change point). Next block
+       // should be a suitable starting point (if not, something is
+       // pretty strange), so it will fill up again soon enough.
+       suitable_starting_points.clear();
+
+       // HLS, on the other hand, can deal with discontinuities and multiple
+       // headers. At least in theory (client support varies wildly).
+       if (!fragments.empty()) {
+               // Commit the old header to the backlog, so that we can serve it
+               // for all the old fragments for as long as they exist.
+               if (!stream_header.empty()) {
+                       // End the current fragment and make a new one for the header.
+                       fragments.push_back(Stream::FragmentStart { bytes_received, 0.0, true });
+                       process_queued_data();
+                       Stream::DataElement elem;
+                       elem.data.iov_base = (char *)stream_header.data();
+                       elem.data.iov_len = stream_header.size();
+                       add_data_raw({ elem });
+                       remove_obsolete_starting_points();
+
+                       // The discontinuity counter will be increased when
+                       // this header goes out of the backlog.
+               }
+               clear_hls_playlist_cache();
+       }
+       stream_header = new_stream_header;
+}
+
 void Stream::put_client_to_sleep(Client *client)
 {
        sleeping_clients.push_back(client);
@@ -241,8 +281,12 @@ void Stream::remove_obsolete_starting_points()
        assert(backlog_size >= hls_backlog_margin);
        while (!fragments.empty() &&
               bytes_received - fragments[0].byte_position > (backlog_size - hls_backlog_margin)) {
+               if (fragments[0].begins_header) {
+                       ++discontinuity_counter;
+               } else {
+                       ++first_fragment_index;
+               }
                fragments.pop_front();
-               ++first_fragment_index;
                clear_hls_playlist_cache();
        }
 }
@@ -383,9 +427,11 @@ bool Stream::add_fragment_boundary(size_t byte_position, const RationalPTS &pts)
 {
        double pts_double = double(pts.pts) * pts.timebase_den / pts.timebase_num;
 
-       if (fragments.size() <= 1) {
+       if (fragments.size() <= 1 ||
+           fragments[fragments.size() - 1].begins_header ||
+           fragments[fragments.size() - 2].begins_header) {
                // Just starting up, so try to establish the first in-progress fragment.
-               fragments.push_back(FragmentStart{ byte_position, pts_double });
+               fragments.push_back(FragmentStart{ byte_position, pts_double, false });
                return false;
        }
 
@@ -398,12 +444,12 @@ bool Stream::add_fragment_boundary(size_t byte_position, const RationalPTS &pts)
        double candidate_duration = pts_double - fragments[fragments.size() - 2].pts;
        if (lrintf(candidate_duration) <= hls_frag_duration &&
            fabs(candidate_duration - hls_frag_duration) < fabs(current_duration - hls_frag_duration)) {
-               fragments.back() = FragmentStart{ byte_position, pts_double };
+               fragments.back() = FragmentStart{ byte_position, pts_double, false };
                return false;
        } else {
                // Extending the in-progress fragment would make it too long,
                // so finalize it and start a new in-progress fragment.
-               fragments.push_back(FragmentStart{ byte_position, pts_double });
+               fragments.push_back(FragmentStart{ byte_position, pts_double, false });
                return true;
        }
 }
@@ -430,15 +476,57 @@ shared_ptr<const string> Stream::generate_hls_playlist(bool http_11, bool close_
 
        string playlist = buf;
 
-       if (!stream_header.empty()) {
-               snprintf(buf, sizeof(buf), "#EXT-X-MAP:URI=\"%s?frag=header\"\r\n", url.c_str());
-               playlist += buf;
-       }
-
-       playlist += "\r\n";
        if (fragments.size() >= 3) {
+               bool printed_header_for_this_group = false;
+               bool printed_first_header = false;
                for (size_t i = 0; i < fragments.size() - 2; ++i) {
                        char buf[256];
+
+                       if (fragments[i].begins_header) {
+                               // End of this group. (We've already printed the header
+                               // as part of the previous group.)
+                               printed_header_for_this_group = false;
+                               continue;
+                       }
+                       if (!printed_header_for_this_group) {
+                               // Look forward until we find the header for this group (if any).
+                               for (size_t j = i + 1; j < fragments.size() - 1; ++j) {
+                                       if (fragments[j].begins_header) {
+                                               if (printed_first_header) {
+                                                       playlist += "#EXT-X-DISCONTINUITY\r\n";
+                                               }
+                                               snprintf(buf, sizeof(buf),
+                                                       "#EXT-X-MAP:URI=\"%s?frag=%" PRIu64 "-%" PRIu64 "\"\r\n",
+                                                       url.c_str(), fragments[j].byte_position,
+                                                       fragments[j + 1].byte_position);
+                                               playlist += buf;
+                                               printed_first_header = true;
+                                               printed_header_for_this_group = true;
+                                               break;
+                                       }
+                               }
+
+                               if (!printed_header_for_this_group && !stream_header.empty()) {
+                                       if (printed_first_header) {
+                                               playlist += "#EXT-X-DISCONTINUITY\r\n";
+                                       }
+                                       snprintf(buf, sizeof(buf), "#EXT-X-MAP:URI=\"%s?frag=header\"\r\n", url.c_str());
+                                       playlist += buf;
+                               }
+
+                               // Even if we didn't find anything, we don't want to search again for each fragment.
+                               printed_first_header = true;
+                               printed_header_for_this_group = true;
+                       }
+
+                       if (fragments[i + 1].begins_header) {
+                               // Since we only have start pts for each block and not duration,
+                               // we have no idea how long this fragment is; the encoder restarted
+                               // before it got to output the next pts. However, it's likely
+                               // to be very short, so instead of trying to guess, we just skip it.
+                               continue;
+                       }
+
                        snprintf(buf, sizeof(buf), "#EXTINF:%f,\r\n%s?frag=%" PRIu64 "-%" PRIu64 "\r\n",
                                fragments[i + 1].pts - fragments[i].pts,
                                url.c_str(),