+}
+
+void Mux::thread_func()
+{
+ unique_lock<mutex> lock(mu);
+ for ( ;; ) {
+ packet_queue_ready.wait(lock, [this]() {
+ return writer_thread_should_quit || (!packet_queue.empty() && plug_count == 0);
+ });
+ if (writer_thread_should_quit && packet_queue.empty()) {
+ // All done.
+ break;
+ }
+
+ assert(!packet_queue.empty() && plug_count == 0);
+ vector<QueuedPacket> packets;
+ swap(packets, packet_queue);
+
+ lock.unlock();
+ for (QueuedPacket &qp : packets) {
+ write_packet_or_die(*qp.pkt, qp.unscaled_pts);
+ av_packet_free(&qp.pkt);
+ }
+ lock.lock();
+ }
+}
+
+void MuxMetrics::init(const vector<pair<string, string>> &labels)
+{
+ vector<pair<string, string>> labels_video = labels;
+ labels_video.emplace_back("stream", "video");
+ global_metrics.add("mux_stream_bytes", labels_video, &metric_video_bytes);
+
+ vector<pair<string, string>> labels_audio = labels;
+ labels_audio.emplace_back("stream", "audio");
+ global_metrics.add("mux_stream_bytes", labels_audio, &metric_audio_bytes);
+
+ global_metrics.add("mux_written_bytes", labels, &metric_written_bytes);