GStreamer 中具有定时录制管道的多线程应用程序

Question

我正在尝试构建一个可以记录特定持续时间（例如 5 秒）的管道，然后停止进程并优雅地进行清理。为此，我在一个单独的线程中设置一个计时器，该计时器在结束时将 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;
  }
}

我遇到的问题是我似乎从来没有在公交车上收到过 EOS 消息:

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;
}

具体来说，我从未收到消息“EOS 已收到 - 停止录制”。

我尝试过的解决方案： 我尝试在管道上设置消息转发属性，以便将事件传播到每个元素，但这没有帮助。

我还尝试在 g_idle_add 函数中运行 stop 函数，以便它在主线程中运行，但这没有帮助。这是 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 似乎以某种方式注册，因为我得到的记录似乎是我作为参数发送的持续时间。然而由于某种原因它永远不会到达公共汽车。 我正在使用 GStreamer 1.16

Answer 1

通过注册巴士解决了我的问题

bus_watch_id = gst_bus_add_watch(bus, (GstBusFunc)message_cb, this); 

而不是：

 g_signal_connect(G_OBJECT(bus), "message", G_CALLBACK(message_cb), this);