X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=stream.cpp;h=6a5ef99f78fa80a8294b6bf9d3bf8ee52d0bb1c0;hp=2c7c42ed75154fbff911fcdcf53421c7dd7500e3;hb=980ac162414c9fce62af4fdb9cfc282865b31572;hpb=cb042ee88b5f4cd6e5f55bada4d9d86807f1640b

diff --git a/stream.cpp b/stream.cpp
index 2c7c42e..6a5ef99 100644
--- a/stream.cpp
+++ b/stream.cpp
@@ -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(),