Add GStreamer input prototype

5 files changed, 315 insertions, 13 deletions

diff --git a/src/GSTInput.cpp b/src/GSTInput.cpp
new file mode 100644
index 0000000..41fbfc0
--- /dev/null
+++ b/src/GSTInput.cpp
@@ -0,0 +1,188 @@
+/* ------------------------------------------------------------------
+ * Copyright (C) 2019 Matthias P. Braendli
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied.
+ * See the License for the specific language governing permissions
+ * and limitations under the License.
+ * -------------------------------------------------------------------
+ */
+
+#include <string>
+#include <chrono>
+#include <algorithm>
+#include <functional>
+#include <cstddef>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+
+#include <gst/audio/audio.h>
+
+#include "GSTInput.h"
+
+#include "config.h"
+
+#if HAVE_GST
+
+using namespace std;
+
+GSTData::GSTData(SampleQueue<uint8_t>& samplequeue) :
+    samplequeue(samplequeue)
+{ }
+
+GSTInput::GSTInput(const std::string& uri,
+        int rate,
+        unsigned channels,
+        SampleQueue<uint8_t>& queue) :
+    m_uri(uri),
+    m_channels(channels),
+    m_rate(rate),
+    m_gst_data(queue),
+    m_samplequeue(queue)
+{ }
+
+static void error_cb(GstBus *bus, GstMessage *msg, GSTData *data)
+{
+    GError *err;
+    gchar *debug_info;
+
+    /* Print error details on the screen */
+    gst_message_parse_error(msg, &err, &debug_info);
+    g_printerr("Error received from element %s: %s\n", GST_OBJECT_NAME (msg->src), err->message);
+    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)
+{
+    /* only link once */
+    GstPad *audiopad = gst_element_get_static_pad(data->audio_convert, "sink");
+    if (GST_PAD_IS_LINKED(audiopad)) {
+        g_object_unref(audiopad);
+        return;
+    }
+
+    /* check media type */
+    GstCaps *caps = gst_pad_query_caps(pad, NULL);
+    GstStructure *str = gst_caps_get_structure(caps, 0);
+    if (!g_strrstr(gst_structure_get_name(str), "audio")) {
+        gst_caps_unref(caps);
+        gst_object_unref(audiopad);
+        return;
+    }
+    gst_caps_unref(caps);
+
+    gst_pad_link(pad, audiopad);
+
+    g_object_unref(audiopad);
+}
+
+static GstFlowReturn new_sample (GstElement *sink, GSTData *data) {
+    GstSample *sample;
+    /* Retrieve the buffer */
+    g_signal_emit_by_name(sink, "pull-sample", &sample);
+    if (sample) {
+        GstBuffer* buffer = gst_sample_get_buffer(sample);
+
+        GstMapInfo map;
+        gst_buffer_map(buffer, &map, GST_MAP_READ);
+
+        data->samplequeue.push(map.data, map.size);
+
+        gst_buffer_unmap(buffer, &map);
+        gst_sample_unref(sample);
+
+        return GST_FLOW_OK;
+    }
+    return GST_FLOW_ERROR;
+}
+
+void GSTInput::prepare()
+{
+    gst_init(nullptr, nullptr);
+
+    m_gst_data.uridecodebin = gst_element_factory_make("uridecodebin", "uridecodebin");
+    assert(m_gst_data.uridecodebin != nullptr);
+    g_object_set(m_gst_data.uridecodebin, "uri", m_uri.c_str(), nullptr);
+    g_signal_connect(m_gst_data.uridecodebin, "pad-added", G_CALLBACK(cb_newpad), &m_gst_data);
+
+    m_gst_data.audio_convert = gst_element_factory_make("audioconvert", "audio_convert");
+    assert(m_gst_data.audio_convert != nullptr);
+
+    m_gst_data.caps_filter = gst_element_factory_make("capsfilter", "caps_filter");
+    assert(m_gst_data.caps_filter != nullptr);
+
+    GstAudioInfo info;
+    gst_audio_info_set_format(&info, GST_AUDIO_FORMAT_S16, m_rate, m_channels, NULL);
+    GstCaps *audio_caps = gst_audio_info_to_caps(&info);
+    g_object_set(m_gst_data.caps_filter, "caps", audio_caps, NULL);
+
+    m_gst_data.app_sink = gst_element_factory_make("appsink", "app_sink");
+    assert(m_gst_data.app_sink != nullptr);
+
+    m_gst_data.pipeline = gst_pipeline_new("pipeline");
+    assert(m_gst_data.pipeline != nullptr);
+
+    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);
+
+    gst_bin_add_many(GST_BIN(m_gst_data.pipeline),
+            m_gst_data.uridecodebin,
+            m_gst_data.audio_convert,
+            m_gst_data.caps_filter,
+            m_gst_data.app_sink, NULL);
+
+    if (gst_element_link_many(
+                m_gst_data.audio_convert,
+                m_gst_data.caps_filter,
+                m_gst_data.app_sink, NULL) != true) {
+        throw runtime_error("Could not link GST elements");
+    }
+
+    m_gst_data.bus = gst_element_get_bus(m_gst_data.pipeline);
+    gst_bus_add_signal_watch(m_gst_data.bus);
+    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);
+}
+
+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);
+
+    if (msg) {
+        gst_message_unref(msg);
+        return false;
+    }
+    return true;
+}
+
+GSTInput::~GSTInput()
+{
+    fprintf(stderr, "<<<<<<<<<<<<<<<<<<<< DTOR\n");
+
+    if (m_gst_data.bus) {
+        gst_object_unref(m_gst_data.bus);
+    }
+
+    if (m_gst_data.pipeline) {
+        gst_element_set_state(m_gst_data.pipeline, GST_STATE_NULL);
+        gst_object_unref(m_gst_data.pipeline);
+    }
+}
+
+#endif // HAVE_GST
diff --git a/src/GSTInput.h b/src/GSTInput.h
new file mode 100644
index 0000000..07cf62e
--- /dev/null
+++ b/src/GSTInput.h
@@ -0,0 +1,89 @@
+/* ------------------------------------------------------------------
+ * Copyright (C) 2019 Matthias P. Braendli
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied.
+ * See the License for the specific language governing permissions
+ * and limitations under the License.
+ * -------------------------------------------------------------------
+ */
+/*! \file GSTInput.h
+ *
+ * This input uses GStreamer to get audio data.
+ */
+
+#pragma once
+
+#include "config.h"
+
+#if HAVE_GST
+
+#include <string>
+#include <vector>
+#include <cstddef>
+#include <cstdint>
+
+#include <gst/gst.h>
+
+#include "SampleQueue.h"
+#include "common.h"
+#include "InputInterface.h"
+
+extern "C" {
+#include "utils.h"
+}
+
+struct GSTData {
+    GSTData(SampleQueue<uint8_t>& samplequeue);
+
+    GstElement *pipeline = nullptr;
+    GstElement *uridecodebin = nullptr;
+    GstElement *audio_convert = nullptr;
+    GstElement *caps_filter = nullptr;
+    GstElement *app_sink = nullptr;
+
+    GstBus *bus = nullptr;
+    GMainLoop *main_loop = nullptr;
+
+    SampleQueue<uint8_t>& samplequeue;
+};
+
+class GSTInput : public InputInterface
+{
+    public:
+        GSTInput(const std::string& uri,
+                 int rate,
+                 unsigned channels,
+                 SampleQueue<uint8_t>& queue);
+
+        GSTInput(const GSTInput& other) = delete;
+        GSTInput& operator=(const GSTInput& other) = delete;
+        virtual ~GSTInput();
+
+        virtual void prepare() override;
+
+        virtual bool read_source(size_t num_bytes) override;
+
+        int getRate() { return m_rate; }
+
+        virtual bool fault_detected(void) const override { return false; };
+    private:
+        std::string m_uri;
+        unsigned m_channels;
+        int m_rate;
+
+        GSTData m_gst_data;
+
+        SampleQueue<uint8_t>& m_samplequeue;
+};
+
+#endif // HAVE_GST
+
diff --git a/src/odr-audioenc.cpp b/src/odr-audioenc.cpp
index f4cf01e..45aacf2 100644
--- a/src/odr-audioenc.cpp
+++ b/src/odr-audioenc.cpp
@@ -34,6 +34,7 @@
  *  Interesting starting points for the encoder
  *  - \ref odr-audioenc.cpp Main encoder file
  *  - \ref VLCInput.h VLC Input
+ *  - \ref GSTInput.h GST Input
  *  - \ref AlsaInput.h Alsa Input
  *  - \ref JackInput.h JACK Input
  *  - \ref Outputs.h ZeroMQ, file and EDI outputs
@@ -53,6 +54,7 @@
 #include "FileInput.h"
 #include "JackInput.h"
 #include "VLCInput.h"
+#include "GSTInput.h"
 #include "SampleQueue.h"
 #include "AACDecoder.h"
 #include "StatsPublish.h"
@@ -65,6 +67,7 @@ extern "C" {
 #include "utils.h"
 }
 
+#include <algorithm>
 #include <vector>
 #include <deque>
 #include <chrono>
@@ -103,7 +106,7 @@ void usage(const char* name)
     "ODR-AudioEnc %s is an audio encoder for both DAB and DAB+.\n"
     "The encoder can read from JACK, ALSA or\n"
     "a file source and encode to a ZeroMQ output for ODR-DabMux.\n"
-    "It can also use libvlc as an input.\n"
+    "It can also use libvlc and GStreamer as input.\n"
     "\n"
     "The -D option enables sound card clock drift compensation.\n"
     "A consumer sound card has a clock that is always a bit imprecise, and\n"
@@ -162,6 +165,12 @@ void usage(const char* name)
 #else
     "     The VLC input was disabled at compile-time\n"
 #endif
+    "   For the GStreamer input:\n"
+#if HAVE_GST
+    "     -G, --gst-uri=uri                    Enable GStreamer input and use the URI given as source\n"
+#else
+    "     The GStreamer input was disabled at compile-time\n"
+#endif
     "   Drift compensation\n"
     "     -D, --drift-comp                     Enable ALSA/VLC sound card drift compensation.\n"
     "   Encoder parameters:\n"
@@ -414,6 +423,8 @@ public:
     vector<string> vlc_additional_opts;
     unsigned verbosity = 0;
 
+    string gst_uri;
+
     string jack_name;
 
     bool drift_compensation = false;
@@ -449,8 +460,8 @@ public:
     string dab_channel_mode;
 
     /* Keep track of peaks */
-    int peak_left  = 0;
-    int peak_right = 0;
+    int16_t peak_left  = 0;
+    int16_t peak_right = 0;
 
     /* On silence, die after the silence_timeout expires */
     bool die_on_silence = false;
@@ -487,7 +498,7 @@ public:
     ~AudioEnc();
 
     int run();
-    bool send_frame(const uint8_t *buf, size_t len, int peak_left, int peak_right);
+    bool send_frame(const uint8_t *buf, size_t len, int16_t peak_left, int16_t peak_right);
     shared_ptr<InputInterface> initialise_input();
 };
 
@@ -504,6 +515,9 @@ int AudioEnc::run()
 #if HAVE_VLC
     if (not vlc_uri.empty()) num_inputs++;
 #endif
+#if HAVE_GST
+    if (not gst_uri.empty()) num_inputs++;
+#endif
 
     if (num_inputs == 0) {
         fprintf(stderr, "No input defined!\n");
@@ -977,8 +991,8 @@ int AudioEnc::run()
         for (int i = 0; i < read_bytes; i+=4) {
             int16_t l = input_buf[i] | (input_buf[i+1] << 8);
             int16_t r = input_buf[i+2] | (input_buf[i+3] << 8);
-            peak_left  = MAX(peak_left,  l);
-            peak_right = MAX(peak_right, r);
+            peak_left  = std::max(peak_left,  l);
+            peak_right = std::max(peak_right, r);
         }
 
         if (stats_publisher) {
@@ -992,7 +1006,7 @@ int AudioEnc::run()
          * threshold is 0, and not configurable. The rationale is that we want to
          * guard against connection issues, not source level issues.
          */
-        if (die_on_silence && MAX(peak_left, peak_right) == 0) {
+        if (die_on_silence && std::max(peak_left, peak_right) == 0) {
             const unsigned int frame_time_msec = 1000ul *
                 read_bytes / (BYTES_PER_SAMPLE * channels * sample_rate);
 
@@ -1174,7 +1188,7 @@ int AudioEnc::run()
             if (show_level) {
                 if (channels == 1) {
                     fprintf(stderr, "\rIn: [%-6s] %1s %1s %1s",
-                            level(1, MAX(peak_right, peak_left)),
+                            level(1, std::max(peak_right, peak_left)),
                             status & STATUS_PAD_INSERTED ? "P" : " ",
                             status & STATUS_UNDERRUN ? "U" : " ",
                             status & STATUS_OVERRUN ? "O" : " ");
@@ -1215,7 +1229,7 @@ int AudioEnc::run()
     return retval;
 }
 
-bool AudioEnc::send_frame(const uint8_t *buf, size_t len, int peak_left, int peak_right)
+bool AudioEnc::send_frame(const uint8_t *buf, size_t len, int16_t peak_left, int16_t peak_right)
 {
     if (file_output) {
         file_output->update_audio_levels(peak_left, peak_right);
@@ -1287,6 +1301,11 @@ shared_ptr<InputInterface> AudioEnc::initialise_input()
                 queue);
     }
 #endif
+#if HAVE_GST
+    else if (not gst_uri.empty()) {
+        input = make_shared<GSTInput>(gst_uri, sample_rate, channels, queue);
+    }
+#endif
 #if HAVE_ALSA
     else if (drift_compensation) {
         input = make_shared<AlsaInputThreaded>(alsa_device, channels,
@@ -1322,6 +1341,7 @@ int main(int argc, char *argv[])
         {"timestamp-delay",        required_argument,  0, 'T'},
         {"decode",                 required_argument,  0,  6 },
         {"format",                 required_argument,  0, 'f'},
+        {"gst-uri",                required_argument,  0, 'G'},
         {"input",                  required_argument,  0, 'i'},
         {"jack",                   required_argument,  0, 'j'},
         {"output",                 required_argument,  0, 'o'},
@@ -1372,7 +1392,7 @@ int main(int argc, char *argv[])
     int ch=0;
     int index;
     while(ch != -1) {
-        ch = getopt_long(argc, argv, "aAhDlRVb:B:c:e:f:i:j:k:L:o:r:d:p:P:s:S:T:v:w:Wg:C:", longopts, &index);
+        ch = getopt_long(argc, argv, "aAhDlRVb:B:c:e:f:G:i:j:k:L:o:r:d:p:P:s:S:T:v:w:Wg:C:", longopts, &index);
         switch (ch) {
         case 0: // AAC-LC
             audio_enc.aot = AOT_DABPLUS_AAC_LC;
@@ -1442,6 +1462,11 @@ int main(int argc, char *argv[])
                 return 1;
             }
             break;
+#ifdef HAVE_GST
+        case 'G':
+            audio_enc.gst_uri = optarg;
+            break;
+#endif
         case 'i':
             audio_enc.infile = optarg;
             break;
diff --git a/src/utils.c b/src/utils.c
index 24da427..928304e 100644
--- a/src/utils.c
+++ b/src/utils.c
@@ -3,6 +3,9 @@
 #include <stdint.h>
 #include <math.h>
 
+#define MIN(a,b) (((a)<(b))?(a):(b))
+#define MAX(a,b) (((a)>(b))?(a):(b))
+
 /* Taken from sox */
 const char* level(int channel, int peak)
 {
diff --git a/src/utils.h b/src/utils.h
index 2cb06c3..ca77a53 100644
--- a/src/utils.h
+++ b/src/utils.h
@@ -4,9 +4,6 @@
 #include <stdint.h>
 #include <stddef.h>
 
-#define MIN(a,b) (((a)<(b))?(a):(b))
-#define MAX(a,b) (((a)>(b))?(a):(b))
-
 #define NUMOF(l) (sizeof(l) / sizeof(*l))
 
 #define linear_to_dB(x) (log10(x) * 20)