GStreamer多线程定时录制管道：EOS消息未触发总线回调问题

场景与问题 #

我正在构建一个GStreamer管道，可按指定时长（如5秒）进行录制，录制完成后优雅停止进程并清理资源。实现逻辑是在单独线程中设置定时器，定时器结束时向管道发送EOS事件，但总线始终无法接收到EOS消息——从未输出“EOS Received - Stopping Recording”这条日志。不过录制文件时长符合设定参数，说明EOS似乎已被管道识别。使用GStreamer 1.16版本。

发送EOS的代码实现 #

GstState current, pending;
gst_element_get_state(pipeline, &current, &pending, GST_CLOCK_TIME_NONE);
if (current != GST_STATE_PLAYING) {
  std::cerr << "Pipeline is not in PLAYING state, current state: "
            << gst_element_state_get_name(current) << std::endl;
}
gst_element_send_event(pipeline, gst_event_new_eos());
std::cout << "EOS sent" << std::endl;
// Wait for the EOS message to be received
{
  std::unique_lock<std::mutex> lock(stop_mutex);
  stop_cv.wait(lock, [this] { return eos_received; });
  std::cout << "eos received" << std::endl;
}
// we can quit the main loop
g_main_loop_quit(loop);
g_main_loop_unref(loop);
running = false;
}
}

总线消息回调代码 #

gboolean AutoRecorderSink::message_cb(GstBus *bus, GstMessage *message,
                                      gpointer user_data) {
  AutoRecorderSink *self = static_cast<AutoRecorderSink *>(user_data);
  switch (GST_MESSAGE_TYPE(message)) {
  case GST_MESSAGE_ERROR:
    GError *err;
    gchar *debug_info;
    gst_message_parse_error(message, &err, &debug_info);
    std::cerr << "Error received from element " << GST_OBJECT_NAME(message->src)
              << ": " << err->message << std::endl;
    std::cerr << "Debugging information: " << (debug_info ? debug_info : "none")
              << std::endl;
    g_clear_error(&err);
    g_free(debug_info);
    self->stop();
    break;
  case GST_MESSAGE_EOS:
    std::cout << "EOS Received - Stopping Recording.\n";
    {
      std::lock_guard<std::mutex> lock(self->stop_mutex);
      self->eos_received = true;
    }
    self->stop_cv.notify_one(); // continue with the stop() function
    break;
  default:
    break;
  }
  return TRUE;
}

已尝试的解决方案 #

• 设置管道的message-forward属性，让事件传播到每个元素，未解决问题。
• 通过g_idle_add让停止函数在主线程运行，无效果，相关代码如下：

gboolean AutoRecorderSink::stop_pipeline_idle(gpointer user_data) {
    AutoRecorderSink *recorder = static_cast<AutoRecorderSink *>(user_data);
    recorder->perform_stop();
    return G_SOURCE_REMOVE; // Remove the idle function after execution
}

void AutoRecorderSink::stop() {
    std::cout << "Stop called, checking running variable" << std::endl;
    if (running) {
        // Queue the stop logic to run in the main loop thread
        g_idle_add(stop_pipeline_idle, this);
    }
}

void AutoRecorderSink::perform_stop() {
    std::cout << "Stopping the pipeline" << std::endl;

    GstState current, pending;
    gst_element_get_state(pipeline, &current, &pending, GST_CLOCK_TIME_NONE);
    if (current != GST_STATE_PLAYING && current != GST_STATE_PAUSED) {
        std::cerr << "Pipeline is not in PLAYING or PAUSED state, current state: "
                  << gst_element_state_get_name(current) << std::endl;
    }

    // Send EOS to the pipeline
    if (!gst_element_send_event(pipeline, gst_event_new_eos())) {
        std::cerr << "Failed to send EOS event" << std::endl;
        return;
    }
    std::cout << "EOS sent" << std::endl;

    // Wait for the EOS message to be received on the bus
    {
        std::unique_lock<std::mutex> lock(stop_mutex);
        stop_cv.wait(lock, [this] { return eos_received; });
        std::cout << "EOS received" << std::endl;
    }

    // Clean up GMainLoop
    if (loop) {
        g_main_loop_quit(loop);
        g_main_loop_unref(loop);
        loop = nullptr;
    }
    running = false;
}

备注：录制文件时长符合设定参数，但EOS消息从未到达总线。

内容的提问来源于stack exchange，提问作者Karim Dhrif