Improved GenericStreamSource, with stream discoverer

Also timeout to fail if open does not works + new GST icon.

Source.h

@@ -23,7 +23,8 @@
 #define ICON_SOURCE_GROUP 10, 6
 #define ICON_SOURCE_RENDER 0, 2
 #define ICON_SOURCE_CLONE 9, 2
-#define ICON_SOURCE 12, 11
+#define ICON_SOURCE_GSTREAMER 16, 16
+#define ICON_SOURCE 13, 11
 
 class SourceCallback;
 class ImageShader;

Stream.cpp

@@ -17,6 +17,9 @@
  * along with this program. If not, see <https://www.gnu.org/licenses/>.
 **/
 
+#include <thread>
+#include <chrono>
+
 //  Desktop OpenGL function loader
 #include <glad/glad.h>
 
@@ -124,15 +127,10 @@ GstFlowReturn callback_stream_discoverer (GstAppSink *sink, gpointer p)
 StreamInfo StreamDiscoverer(const std::string &description, guint w, guint h)
 {
     // the stream info to return
-    StreamInfo info;
+    StreamInfo info(w, h);
 
-    // obvious fast answer: valid values are provided in argument
+    // no valid info, run a test pipeline to discover the size of the stream
-    if (w > 0 && h > 0 ) {
+    if ( !info.valid() ) {
-        info.width = w;
-        info.height = h;
-    }
-    // otherwise, run a test pipeline to discover the size of the stream
-    else {
         // complete the pipeline description with an appsink (to add a callback)
         std::string _description = description;
         _description += " ! appsink name=sink";
@@ -157,19 +155,15 @@ StreamInfo StreamDiscoverer(const std::string &description, guint w, guint h)
                 // start to play the pipeline
                 gst_element_set_state (_pipeline, GST_STATE_PLAYING);
 
-                // wait for the callback_stream_discoverer to return
+                // wait for the callback_stream_discoverer to return, no more than 4 sec
                 std::mutex mtx;
                 std::unique_lock<std::mutex> lck(mtx);
-                // if waited more than 2 seconds, its dead :(
+                info.discovered.wait_for(lck,std::chrono::seconds(TIMEOUT));
-                if ( info.discovered.wait_for(lck,std::chrono::seconds(2)) == std::cv_status::timeout)
-                    Log::Warning("Failed to discover stream size.");
 
                 // stop and delete pipeline
                 GstStateChangeReturn ret = gst_element_set_state (_pipeline, GST_STATE_NULL);
-                if (ret == GST_STATE_CHANGE_ASYNC) {
+                if (ret == GST_STATE_CHANGE_ASYNC)
-                    GstState state;
+                    gst_element_get_state (_pipeline, NULL, NULL, 1000000);
-                    gst_element_get_state (_pipeline, &state, NULL, GST_CLOCK_TIME_NONE);
-                }
                 gst_object_unref (_pipeline);
             }
         }
@@ -214,9 +208,8 @@ void Stream::execute_open()
     GError *error = NULL;
     pipeline_ = gst_parse_launch (description.c_str(), &error);
     if (error != NULL) {
-        Log::Warning("Stream %s Could not construct pipeline %s:\n%s", std::to_string(id_).c_str(), description.c_str(), error->message);
+        fail(std::string("Could not construct pipeline: ") + error->message + "\n" + description);
         g_clear_error (&error);
-        failed_ = true;
         return;
     }
     g_object_set(G_OBJECT(pipeline_), "name", std::to_string(id_).c_str(), NULL);
@@ -227,16 +220,14 @@ void Stream::execute_open()
             ",height=" + std::to_string(height_);
     GstCaps *caps = gst_caps_from_string(capstring.c_str());
     if (!caps || !gst_video_info_from_caps (&v_frame_video_info_, caps)) {
-        Log::Warning("Stream %d Could not configure video frame info", id_);
+        fail("Could not configure video frame info");
-        failed_ = true;
         return;
     }
 
     // setup appsink
     GstElement *sink = gst_bin_get_by_name (GST_BIN (pipeline_), "sink");
     if (!sink) {
-        Log::Warning("Stream %s Could not configure  sink", std::to_string(id_).c_str());
+        fail("Could not configure pipeline sink.");
-        failed_ = true;
         return;
     }
 
@@ -276,8 +267,7 @@ void Stream::execute_open()
     live_ = false;
     GstStateChangeReturn ret = gst_element_set_state (pipeline_, desired_state_);
     if (ret == GST_STATE_CHANGE_FAILURE) {
-        Log::Warning("Stream %s Could not open '%s'", std::to_string(id_).c_str(), description_.c_str());
+        fail(std::string("Could not open ") + description_);
-        failed_ = true;
         return;
     }
     else if (ret == GST_STATE_CHANGE_NO_PREROLL) {
@@ -295,6 +285,24 @@ void Stream::execute_open()
     // all good
     Log::Info("Stream %s Opened '%s' (%d x %d)", std::to_string(id_).c_str(), description.c_str(), width_, height_);
     opened_ = true;
+
+    // launch a timeout to check on open status
+    std::thread( timeout_open, this ).detach();
+}
+
+void Stream::fail(const std::string &message)
+{
+    Log::Warning("Stream %s %s.", std::to_string(id_).c_str(), message.c_str() );
+    failed_ = true;
+}
+
+void Stream::timeout_open(Stream *str)
+{
+    // vait for timeout
+    std::this_thread::sleep_for(std::chrono::seconds(TIMEOUT));
+
+    if (!str->textureinitialized_)
+        str->fail("Failed to initialize");
 }
 
 bool Stream::isOpen() const
@@ -332,15 +340,13 @@ void Stream::close()
     // clean up GST
     if (pipeline_ != nullptr) {
         // force flush
-        GstState state;
         gst_element_send_event(pipeline_, gst_event_new_seek (1.0, GST_FORMAT_TIME, GST_SEEK_FLAG_FLUSH,
                     GST_SEEK_TYPE_SET, 0, GST_SEEK_TYPE_END, 0) );
-        gst_element_get_state (pipeline_, &state, NULL, GST_CLOCK_TIME_NONE);
+        gst_element_get_state (pipeline_, NULL, NULL, 1000000);
-
+        // force end
         GstStateChangeReturn ret = gst_element_set_state (pipeline_, GST_STATE_NULL);
-        if (ret == GST_STATE_CHANGE_ASYNC) {
+        if (ret == GST_STATE_CHANGE_ASYNC)
-            gst_element_get_state (pipeline_, &state, NULL, GST_CLOCK_TIME_NONE);
+            gst_element_get_state (pipeline_, NULL, NULL, 1000000);
-        }
         gst_object_unref (pipeline_);
         pipeline_ = nullptr;
     }
@@ -391,10 +397,8 @@ void Stream::enable(bool on)
 
         //  apply state change
         GstStateChangeReturn ret = gst_element_set_state (pipeline_, requested_state);
-        if (ret == GST_STATE_CHANGE_FAILURE) {
+        if (ret == GST_STATE_CHANGE_FAILURE)
-            Log::Warning("Stream %s Failed to enable", std::to_string(id_).c_str());
+            fail("Failed to enable");
-            failed_ = true;
-        }
 
     }
 }
@@ -436,10 +440,9 @@ void Stream::play(bool on)
 
     // all ready, apply state change immediately
     GstStateChangeReturn ret = gst_element_set_state (pipeline_, desired_state_);
-    if (ret == GST_STATE_CHANGE_FAILURE) {
+    if (ret == GST_STATE_CHANGE_FAILURE)
-        Log::Warning("Stream %s Failed to play", std::to_string(id_).c_str());
+        fail("Failed to play");
-        failed_ = true;
+
-    }
 #ifdef STREAM_DEBUG
     else if (on)
         Log::Info("Stream %s Start", std::to_string(id_).c_str());
@@ -616,11 +619,17 @@ void Stream::update()
             // try to get info from discoverer
             if (discoverer_.wait_for( std::chrono::milliseconds(4) ) == std::future_status::ready )
             {
-                // got all info needed for openning !
+                // get info
                 StreamInfo i(discoverer_.get());
-                width_ = i.width;
+                if (i.valid()) {
-                height_ = i.height;
+                    // got all info needed for openning !
-                execute_open();
+                    width_ = i.width;
+                    height_ = i.height;
+                    execute_open();
+                }
+                // invalid info; fail
+                else
+                    fail("Failed to determine resolution");
             }
         }
         // wait next frame to display

Stream.h

@@ -15,6 +15,7 @@
 class Visitor;
 
 #define N_FRAME 3
+#define TIMEOUT 4
 
 struct StreamInfo {
 
@@ -22,15 +23,17 @@ struct StreamInfo {
     guint height;
     std::condition_variable discovered;
 
-    StreamInfo() {
+    StreamInfo(guint w=0, guint h=0) {
-        width = 640;
+        width = w;
-        height = 480;
+        height = h;
     }
 
     StreamInfo(const StreamInfo& b) {
         width = b.width;
         height = b.height;
     }
+
+    inline bool valid() { return width > 0 && height > 0; }
 };
 
 class Stream {
@@ -207,9 +210,11 @@ protected:
 
     // gst pipeline control
     virtual void execute_open();
+    virtual void fail(const std::string &message);
+    static void timeout_open(Stream *str);
 
     // gst frame filling
-    bool textureinitialized_;
+    std::atomic<bool> textureinitialized_;
     void init_texture(guint index);
     void fill_texture(guint index);
     bool fill_frame(GstBuffer *buf, FrameStatus status);

StreamSource.cpp

@@ -28,6 +28,7 @@
 #include "Stream.h"
 #include "Visitor.h"
 #include "Log.h"
+#include "BaseToolkit.h"
 
 #include "StreamSource.h"
 
@@ -44,19 +45,29 @@ GenericStreamSource::GenericStreamSource() : StreamSource()
 
 void GenericStreamSource::setDescription(const std::string &desc)
 {
-    Log::Notify("Creating Source with Stream description '%s'", desc.c_str());
+    gst_description_ = desc;
-
+    gst_elements_ = BaseToolkit::splitted(desc, '!');
-    std::string pipeline = desc;
+    Log::Notify("Creating Source with Stream description '%s'", gst_description_.c_str());
-    pipeline.append(" ! queue max-size-buffers=10 ! videoconvert");
 
     // open gstreamer
-    stream_->open(pipeline);
+    stream_->open(gst_description_ + " ! queue max-size-buffers=10 ! videoconvert" );
     stream_->play(true);
 
     // will be ready after init and one frame rendered
     ready_ = false;
 }
 
+
+std::string GenericStreamSource::description() const
+{
+    return gst_description_;
+}
+
+std::list<std::string> GenericStreamSource::gstElements() const
+{
+    return gst_elements_;
+}
+
 void GenericStreamSource::accept(Visitor& v)
 {
     Source::accept(v);
@@ -64,6 +75,18 @@ void GenericStreamSource::accept(Visitor& v)
         v.visit(*this);
 }
 
+glm::ivec2 GenericStreamSource::icon() const
+{
+    return glm::ivec2(ICON_SOURCE_GSTREAMER);
+}
+
+std::string GenericStreamSource::info() const
+{
+    std::string src_element = gst_elements_.front();
+    src_element = src_element.substr(0, src_element.find(" "));
+    return std::string("Gstreamer custom pipeline with source '")+src_element+"'";
+}
+
 StreamSource::StreamSource(uint64_t id) : Source(id), stream_(nullptr)
 {
 }

StreamSource.h

@@ -59,6 +59,9 @@ protected:
  */
 class GenericStreamSource : public StreamSource
 {
+    std::string gst_description_;
+    std::list<std::string> gst_elements_;
+
 public:
     GenericStreamSource();
 
@@ -70,6 +73,11 @@ public:
 
     // specific interface
     void setDescription(const std::string &desc);
+    std::string description() const;
+    std::list<std::string> gstElements() const;
+
+    glm::ivec2 icon() const override;
+    std::string info() const override;
 };
 
 #endif // STREAMSOURCE_H