#include <assert.h>
#include <errno.h>
+#include <inttypes.h>
#include <limits.h>
#include <math.h>
#include <netinet/in.h>
Stream::Stream(const string &url,
size_t backlog_size,
- size_t prebuffering_bytes,
+ uint64_t prebuffering_bytes,
Encoding encoding,
Encoding src_encoding,
unsigned hls_frag_duration,
Stream::Stream(const StreamProto &serialized, int data_fd)
: url(serialized.url()),
+ unavailable(serialized.unavailable()),
http_header(serialized.http_header()),
stream_header(serialized.stream_header()),
encoding(Stream::STREAM_ENCODING_RAW), // Will be changed later.
}
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() });
}
}
StreamProto Stream::serialize()
{
StreamProto serialized;
+ serialized.set_unavailable(unavailable);
serialized.set_http_header(http_header);
serialized.set_stream_header(stream_header);
serialized.add_data_fds(data_fd);
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);
remove_obsolete_starting_points();
}
+void Stream::set_header(const std::string &new_http_header, const std::string &new_stream_header)
+{
+ unavailable = false;
+ 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);
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();
}
}
{
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;
}
// (RFC 8216 4.3.3.1) and we get closer to the target by doing so.
// Note that in particular, this means we'll always extend
// as long as we don't exceed the target duration.
- double current_duration = fragments[fragments.size() - 1].pts;
+ double current_duration = pts_double - fragments[fragments.size() - 1].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;
}
}
"#EXTM3U\r\n"
"#EXT-X-VERSION:7\r\n"
"#EXT-X-TARGETDURATION:%u\r\n"
- "#EXT-X-MEDIA-SEQUENCE:%zu\r\n"
- "#EXT-X-DISCONTINUITY-SEQUENCE:%zu\r\n",
+ "#EXT-X-MEDIA-SEQUENCE:%" PRIu64 "\r\n"
+ "#EXT-X-DISCONTINUITY-SEQUENCE:%" PRIu64 "\r\n",
hls_frag_duration,
first_fragment_index,
discontinuity_counter);
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];
- snprintf(buf, sizeof(buf), "#EXTINF:%f,\r\n%s?frag=%zu-%zu\r\n",
+
+ 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(),
fragments[i].byte_position,
assert(close_after_response);
response = "HTTP/1.0 200 OK\r\n";
}
- snprintf(buf, sizeof(buf), "Content-length: %zu\r\n", playlist.size());
+ snprintf(buf, sizeof(buf), "Content-Length: %zu\r\n", playlist.size());
response.append(buf);
- response.append("Content-type: application/x-mpegURL\r\n");
+ response.append("Content-Type: application/x-mpegURL\r\n");
if (!allow_origin.empty()) {
response.append("Access-Control-Allow-Origin: ");
response.append(allow_origin);
projects / cubemap / blobdiff