Fix GStreamer input, rework ICY-Text write

1 files changed, 99 insertions, 13 deletions

diff --git a/src/GSTInput.cpp b/src/GSTInput.cpp
index 41fbfc0..bc7d44b 100644
--- a/src/GSTInput.cpp
+++ b/src/GSTInput.cpp
@@ -26,6 +26,7 @@
 #include <cstring>
 
 #include <gst/audio/audio.h>
+#include <gst/app/gstappsink.h>
 
 #include "GSTInput.h"
 
@@ -46,8 +47,7 @@ GSTInput::GSTInput(const std::string& uri,
     m_uri(uri),
     m_channels(channels),
     m_rate(rate),
-    m_gst_data(queue),
-    m_samplequeue(queue)
+    m_gst_data(queue)
 { }
 
 static void error_cb(GstBus *bus, GstMessage *msg, GSTData *data)
@@ -61,8 +61,6 @@ static void error_cb(GstBus *bus, GstMessage *msg, GSTData *data)
     g_printerr("Debugging information: %s\n", debug_info ? debug_info : "none");
     g_clear_error(&err);
     g_free(debug_info);
-
-    g_main_loop_quit(data->main_loop);
 }
 
 static void cb_newpad(GstElement *decodebin, GstPad *pad, GSTData *data)
@@ -89,10 +87,9 @@ static void cb_newpad(GstElement *decodebin, GstPad *pad, GSTData *data)
     g_object_unref(audiopad);
 }
 
-static GstFlowReturn new_sample (GstElement *sink, GSTData *data) {
-    GstSample *sample;
+static GstFlowReturn new_sample(GstElement *sink, GSTData *data) {
     /* Retrieve the buffer */
-    g_signal_emit_by_name(sink, "pull-sample", &sample);
+    GstSample* sample = gst_app_sink_pull_sample(GST_APP_SINK(sink));
     if (sample) {
         GstBuffer* buffer = gst_sample_get_buffer(sample);
 
@@ -121,6 +118,14 @@ void GSTInput::prepare()
     m_gst_data.audio_convert = gst_element_factory_make("audioconvert", "audio_convert");
     assert(m_gst_data.audio_convert != nullptr);
 
+    m_gst_data.audio_resample = gst_element_factory_make("audioresample", "audio_resample");
+    assert(m_gst_data.audio_resample != nullptr);
+    g_object_set(m_gst_data.audio_resample,
+            "sinc-filter-mode", GST_AUDIO_RESAMPLER_FILTER_MODE_FULL,
+            "quality", 6, // between 0 and 10, 10 being best
+            /* default audio-resampler-method: GST_AUDIO_RESAMPLER_METHOD_KAISER */
+            NULL);
+
     m_gst_data.caps_filter = gst_element_factory_make("capsfilter", "caps_filter");
     assert(m_gst_data.caps_filter != nullptr);
 
@@ -135,6 +140,7 @@ void GSTInput::prepare()
     m_gst_data.pipeline = gst_pipeline_new("pipeline");
     assert(m_gst_data.pipeline != nullptr);
 
+    // TODO also set max-buffers
     g_object_set(m_gst_data.app_sink, "emit-signals", TRUE, "caps", audio_caps, NULL);
     g_signal_connect(m_gst_data.app_sink, "new-sample", G_CALLBACK(new_sample), &m_gst_data);
     gst_caps_unref(audio_caps);
@@ -142,11 +148,13 @@ void GSTInput::prepare()
     gst_bin_add_many(GST_BIN(m_gst_data.pipeline),
             m_gst_data.uridecodebin,
             m_gst_data.audio_convert,
+            m_gst_data.audio_resample,
             m_gst_data.caps_filter,
             m_gst_data.app_sink, NULL);
 
     if (gst_element_link_many(
                 m_gst_data.audio_convert,
+                m_gst_data.audio_resample,
                 m_gst_data.caps_filter,
                 m_gst_data.app_sink, NULL) != true) {
         throw runtime_error("Could not link GST elements");
@@ -157,23 +165,101 @@ void GSTInput::prepare()
     g_signal_connect(G_OBJECT(m_gst_data.bus), "message::error", (GCallback)error_cb, &m_gst_data);
 
     gst_element_set_state(m_gst_data.pipeline, GST_STATE_PLAYING);
+
+    m_running = true;
+    m_thread = std::thread(&GSTInput::process, this);
 }
 
 bool GSTInput::read_source(size_t num_bytes)
 {
-    // Reading done in glib main loop
-    GstMessage *msg = gst_bus_pop_filtered(m_gst_data.bus, GST_MESSAGE_EOS);
+    return m_running;
+}
 
-    if (msg) {
+ICY_TEXT_t GSTInput::get_icy_text() const
+{
+    ICY_TEXT_t now_playing;
+    {
+        std::lock_guard<std::mutex> lock(m_nowplaying_mutex);
+        now_playing = m_nowplaying;
+    }
+
+    return now_playing;
+}
+
+void GSTInput::process()
+{
+    while (m_running) {
+        GstMessage *msg = gst_bus_timed_pop(m_gst_data.bus, 100000);
+
+        if (not msg) {
+            continue;
+        }
+
+        switch (GST_MESSAGE_TYPE(msg)) {
+            case GST_MESSAGE_BUFFERING:
+                {
+                    gint percent = 0;
+                    gst_message_parse_buffering(msg, &percent);
+                    //fprintf(stderr, "GST buffering %d\n", percent);
+                    break;
+                }
+            case GST_MESSAGE_TAG:
+                {
+                    GstTagList *tags = nullptr;
+                    gst_message_parse_tag(msg, &tags);
+                    //fprintf(stderr, "Got tags from element %s\n", GST_OBJECT_NAME(msg->src));
+
+                    string new_title;
+
+                    auto extract_title = [](const GstTagList *list, const gchar *tag, void *user_data) {
+                        GValue val = { 0, };
+
+                        auto new_title = (string*)user_data;
+
+                        gst_tag_list_copy_value(&val, list, tag);
+
+                        if (strcmp(tag, "title") == 0 and G_VALUE_HOLDS_STRING(&val)) {
+                            *new_title = g_value_dup_string(&val);
+                        }
+
+                        g_value_unset(&val);
+                    };
+
+                    gst_tag_list_foreach(tags, extract_title, &new_title);
+
+                    gst_tag_list_unref(tags);
+                    {
+                        std::lock_guard<std::mutex> lock(m_nowplaying_mutex);
+                        m_nowplaying.useNowPlaying(new_title);
+                    }
+                    break;
+                }
+            case GST_MESSAGE_ERROR:
+                {
+                    GError *err = nullptr;
+                    gst_message_parse_error(msg, &err, nullptr);
+                    fprintf(stderr, "GST error: %s\n", err->message);
+                    g_error_free(err);
+                    m_fault = true;
+                    break;
+                }
+            case GST_MESSAGE_EOS:
+                m_fault = true;
+                break;
+            default:
+                //fprintf(stderr, "GST message %s\n", gst_message_type_get_name(GST_MESSAGE_TYPE(msg)));
+                break;
+        }
         gst_message_unref(msg);
-        return false;
     }
-    return true;
 }
 
 GSTInput::~GSTInput()
 {
-    fprintf(stderr, "<<<<<<<<<<<<<<<<<<<< DTOR\n");
+    m_running = false;
+    if (m_thread.joinable()) {
+        m_thread.join();
+    }
 
     if (m_gst_data.bus) {
         gst_object_unref(m_gst_data.bus);