14 #ifndef NVDS3D_GST_NVDS3D_PIPELINE_CONTEXT_H
15 #define NVDS3D_GST_NVDS3D_PIPELINE_CONTEXT_H
31 namespace ds3d {
namespace gst {
42 GstPad* peer = gst_pad_get_peer(pad);
45 "send EOS downstream [elem:%s] skipped; not linked\n",
46 GST_ELEMENT_NAME(GST_ELEMENT(gst_pad_get_parent(pad))));
50 "sending EOS downstream [elem:%s->%s]\n", GST_ELEMENT_NAME(GST_ELEMENT(gst_pad_get_parent(pad))),
51 GST_ELEMENT_NAME(GST_ELEMENT(gst_pad_get_parent(peer))));
52 if (gst_pad_send_event(peer, gst_event_new_eos()) == FALSE) {
54 "send EOS downstream [elem:%s->%s] failed\n", GST_ELEMENT_NAME(GST_ELEMENT(gst_pad_get_parent(pad))),
55 GST_ELEMENT_NAME(GST_ELEMENT(gst_pad_get_parent(peer))));
57 gst_object_unref(peer);
97 runMainLoop(std::move(loopQuitCb)),
"failed to run main loop on the pipeline");
107 std::unique_lock<std::mutex> lock(
mutex());
108 if (!
_StatusCond.wait_for(lock, std::chrono::seconds(3), [
this]() {
109 return _mainStopped || _eosReceived;
111 LOG_DEBUG(
"waiting for EOS timed out, force to stop");
124 gst_bus_remove_watch(
bus());
136 std::unique_lock<std::mutex> locker(
mutex());
140 if (!g_main_loop_is_running(
mainLoop())) {
145 GstState state = GST_STATE_NULL;
146 GstState pending = GST_STATE_NULL;
147 GstStateChangeReturn ret = gst_element_get_state(
148 GST_ELEMENT(
pipeline()), &state, &pending,
149 (timeout ? timeout * 1000000 : GST_CLOCK_TIME_NONE));
153 while (ret == GST_STATE_CHANGE_FAILURE && times++ < 3) {
154 ret = gst_element_get_state(GST_ELEMENT(
pipeline()), &state, &pending, 0);
156 if (ret == GST_STATE_CHANGE_FAILURE) {
159 if (state == GST_STATE_PLAYING || pending == GST_STATE_PLAYING) {
167 std::unique_lock<std::mutex> locker(
mutex());
176 std::unique_lock<std::mutex> locker(
mutex());
178 if (
_StatusCond.wait_for(locker, std::chrono::milliseconds(3000)) ==
179 std::cv_status::timeout) {
180 LOG_DEBUG(
"waiting loop timed out, force loop to stop");
196 std::unique_lock<std::mutex> locker(
mutex());
210 LOG_WARNING(
"set pipeline state to GST_STATE_NULL failed");
212 GstState end = GST_STATE_NULL;
214 if (!
isGood(c) || end != GST_STATE_NULL) {
215 LOG_WARNING(
"waiting for pipeline state to null failed, force to quit");
219 c =
setState(each.get(), GST_STATE_NULL);
227 std::function<bool()>* f = (std::function<bool()>*)(user_data);
234 std::unique_lock<std::mutex> locker(
mutex());
237 "failed to run main loop due to loop might not set or thread already running.");
240 auto loopThread = std::make_unique<std::thread>([
this, quitCb = std::move(loopQuitCb)]() {
243 std::unique_lock<std::mutex> locker(
mutex());
250 std::atomic_bool loopStarted{
false};
251 std::function<bool()> loopCheck = [&loopStarted,
this]() ->
bool {
252 std::unique_lock<std::mutex> locker(
mutex());
261 if (!
_StatusCond.wait_for(locker, std::chrono::milliseconds(20000), [
this, &loopStarted]() {
262 return loopStarted || _mainStopped;
273 LOG_ERROR(
"Run main loop stopped too fast, please check.");
286 GstIterator* itr =
nullptr;
290 for (itr = gst_bin_iterate_sources(GST_BIN(
pipeline())); gst_iterator_next(itr, &data) == GST_ITERATOR_OK;) {
291 GstElement* elem = GST_ELEMENT_CAST(g_value_get_object(&data));
292 LOG_DEBUG(
"sending EOS downstream from src element %s\n", GST_ELEMENT_NAME(elem));
296 g_value_reset(&data);
298 g_value_unset(&data);
299 gst_iterator_free(itr);
310 virtual bool busCall(GstMessage* msg)
313 switch (GST_MESSAGE_TYPE(msg)) {
314 case GST_MESSAGE_EOS:
317 std::unique_lock<std::mutex> locker(
mutex());
325 case GST_MESSAGE_ERROR: {
326 gchar* debug =
nullptr;
327 GError* error =
nullptr;
328 gst_message_parse_error(msg, &error, &debug);
330 "ERROR from element %s: %s, details: %s", GST_OBJECT_NAME(msg->src), error->message,
331 (debug ? debug :
""));
338 case GST_MESSAGE_STATE_CHANGED: {
339 GstState oldState, newState, pendingState;
341 gst_message_parse_state_changed(msg, &oldState, &newState, &pendingState);
343 "Element %s changed state from %s to %s, pending: %s.", GST_OBJECT_NAME(msg->src),
344 gst_element_state_get_name(oldState), gst_element_state_get_name(newState),
345 gst_element_state_get_name(pendingState));
364 GstStateChangeReturn ret = gst_element_set_state(ele, state);
366 ret != GST_STATE_CHANGE_FAILURE,
ErrCode::kGst,
"element set state: %d failed", state);
373 GstElement* ele, GstState* state, GstState* pending =
nullptr,
size_t timeout = 0)
376 GstStateChangeReturn ret = gst_element_get_state(
377 ele, state, pending, (timeout ? timeout * 1000000 : GST_CLOCK_TIME_NONE));
379 case GST_STATE_CHANGE_FAILURE:
381 case GST_STATE_CHANGE_SUCCESS:
382 case GST_STATE_CHANGE_NO_PREROLL:
390 static gboolean
sBusCall(GstBus*
bus, GstMessage* msg, gpointer data)
394 return ctx->busCall(msg);
416 #endif // NVDS3D_GST_NVDS3D_PIPELINE_CONTEXT_H