Add class for transcoding a Song into an OGG Vorbis stream.

CMakeLists.txt

@@ -52,6 +52,7 @@ find_package(Protobuf)
 pkg_check_modules(TAGLIB REQUIRED taglib>=1.6)
 pkg_check_modules(GSTREAMER gstreamer-0.10)
 pkg_check_modules(GSTREAMER_BASE gstreamer-base-0.10)
+pkg_check_modules(GSTREAMER_APP gstreamer-app-0.10)
 pkg_check_modules(GLIB glib-2.0)
 pkg_check_modules(LIBXML libxml-2.0)
 pkg_check_modules(GOBJECT gobject-2.0)

src/CMakeLists.txt

@@ -847,12 +847,14 @@ if(HAVE_REMOTE)
     remote/portforwarder.h
     remote/remote.h
     remote/remoteconfig.h
+    remote/streampipeline.h
   )
   list(APPEND SOURCES
     remote/icesession.cpp
     remote/portforwarder.h
     remote/remote.cpp
     remote/remoteconfig.cpp
+    remote/streampipeline.cpp
   )
 endif(HAVE_REMOTE)
 
@@ -973,6 +975,7 @@ target_link_libraries(clementine_lib
   ${QT_LIBRARIES}
   ${GSTREAMER_BASE_LIBRARIES}
   ${GSTREAMER_LIBRARIES}
+  ${GSTREAMER_APP_LIBRARIES}
   ${QTSINGLEAPPLICATION_LIBRARIES}
   ${QTSINGLECOREAPPLICATION_LIBRARIES}
   ${QTIOCOMPRESSOR_LIBRARIES}

src/engines/gstengine.cpp

@@ -136,7 +136,7 @@ bool GstEngine::Init() {
     SetEnv("GST_REGISTRY", registry_filename);
   }
 
-  initialising_ = QtConcurrent::run(this, &GstEngine::InitialiseGstreamer);
+  initialising_ = QtConcurrent::run(&GstEngine::InitialiseGstreamer);
 
   return true;
 }

src/engines/gstengine.h

@@ -67,6 +67,7 @@ class GstEngine : public Engine::Base, public BufferConsumer {
 
   bool Init();
   void EnsureInitialised() { initialising_.waitForFinished(); }
+  static void InitialiseGstreamer();
 
   int AddBackgroundStream(const QUrl& url);
   void StopBackgroundStream(int id);
@@ -130,7 +131,6 @@ class GstEngine : public Engine::Base, public BufferConsumer {
 
   static void SetEnv(const char* key, const QString& value);
   PluginDetailsList GetPluginList(const QString& classname) const;
-  void InitialiseGstreamer();
 
   void StartFadeout();
 
@@ -196,4 +196,3 @@ class GstEngine : public Engine::Base, public BufferConsumer {
 
 
 #endif /*AMAROK_GSTENGINE_H*/
-

src/remote/icesession.cpp

@@ -188,6 +188,7 @@ void ICESession::OnReceiveData(pj_ice_strans* ice_st,
   qDebug() << "Received data" << data;
 
   me->receive_buffer_.append(data);
+  emit me->readyRead();
 }
 
 void ICESession::OnICEComplete(pj_ice_strans* ice_st,
@@ -206,6 +207,7 @@ void ICESession::OnICEComplete(pj_ice_strans* ice_st,
       pj_sockaddr addr;
       pj_getdefaultipinterface(pj_AF_INET(), &addr);
       pj_ice_strans_sendto(ice_st, me->component_id_, data, strlen(data), &addr, sizeof(addr));
+      emit me->Connected();
       break;
     }
     default:

src/remote/icesession.h

@@ -37,6 +37,7 @@ class ICESession : public QIODevice {
 
  signals:
   void CandidatesAvailable(const xrme::SIPInfo& candidates);
+  void Connected();
 
  private:
   pj_ice_strans* ice_instance_;

src/remote/streampipeline.cpp

@@ -0,0 +1,125 @@
+#include "streampipeline.h"
+
+#include <gst/app/gstappsink.h>
+
+#include "core/logging.h"
+
+const char* StreamPipeline::kPipeline =
+    "audioconvert ! vorbisenc ! oggmux";
+
+StreamPipeline::StreamPipeline(QObject* parent)
+  : QIODevice(parent),
+    pipeline_(NULL) {
+}
+
+StreamPipeline::~StreamPipeline() {
+  gst_object_unref(pipeline_);
+}
+
+void StreamPipeline::Init(const Song& song) {
+  pipeline_ = gst_pipeline_new("pipeline");
+  GstElement* uridecodebin = CreateElement("uridecodebin", pipeline_);
+  qLog(Debug) << "Streaming:" << song.url();
+  g_object_set(
+      G_OBJECT(uridecodebin), "uri", song.url().toString().toUtf8().constData(), NULL);
+  g_signal_connect(
+      G_OBJECT(uridecodebin), "pad-added", G_CALLBACK(NewPadCallback), this);
+
+  GError* error = NULL;
+  convert_bin_ = gst_parse_bin_from_description(kPipeline, TRUE, &error);
+  gst_bin_add(GST_BIN(pipeline_), convert_bin_);
+
+  gst_element_set_state(uridecodebin, GST_STATE_PLAYING);
+
+  app_sink_ = CreateElement("appsink", pipeline_);
+  g_object_set(G_OBJECT(app_sink_), "emit-signals", TRUE, NULL);
+  g_signal_connect(
+      G_OBJECT(app_sink_), "new-buffer", G_CALLBACK(NewBufferCallback), this);
+
+  qLog(Debug) << "Linking appsink:" << gst_element_link(convert_bin_, app_sink_);
+
+  gst_bus_set_sync_handler(
+      gst_pipeline_get_bus(GST_PIPELINE(pipeline_)), BusCallbackSync, this);
+  bus_cb_id_ = gst_bus_add_watch(
+      gst_pipeline_get_bus(GST_PIPELINE(pipeline_)), BusCallback, this);
+}
+
+qint64 StreamPipeline::readData(char* data, qint64 max_size) {
+  QByteArray ret = buffer_.left(max_size);
+  buffer_ = buffer_.mid(ret.size());
+  memcpy(data, ret.constData(), ret.size());
+  return ret.size();
+}
+
+qint64 StreamPipeline::writeData(const char* data, qint64 max_size) {
+  return -1;
+}
+
+GstElement* StreamPipeline::CreateElement(
+    const QString& factory_name, GstElement* bin) {
+  GstElement* ret = gst_element_factory_make(
+      factory_name.toAscii().constData(),
+      factory_name.toAscii().constData());
+  qLog(Debug) << "Created element:" << factory_name << ret;
+  if (ret && bin) {
+    gst_bin_add(GST_BIN(bin), ret);
+  }
+  return ret;
+}
+
+GstBusSyncReply StreamPipeline::BusCallbackSync(
+    GstBus*, GstMessage* msg, gpointer self) {
+  reinterpret_cast<StreamPipeline*>(self)->HandleMessage(msg);
+  return GST_BUS_PASS;
+}
+
+gboolean StreamPipeline::BusCallback(GstBus*, GstMessage* msg, gpointer self) {
+  reinterpret_cast<StreamPipeline*>(self)->HandleMessage(msg);
+  return FALSE;
+}
+
+void StreamPipeline::HandleMessage(GstMessage* msg) {
+  switch (GST_MESSAGE_TYPE(msg)) {
+    case GST_MESSAGE_EOS:
+      qLog(Debug) << "Finished streaming";
+      break;
+    case GST_MESSAGE_ERROR: {
+      GError* error;
+      gchar* debugs;
+      gst_message_parse_error(msg, &error, &debugs);
+      QString message = QString::fromLocal8Bit(error->message);
+      g_error_free(error);
+      free(debugs);
+      qLog(Debug) << message;
+      break;
+    }
+    default:
+      break;
+  }
+}
+
+void StreamPipeline::NewBufferCallback(GstElement* app_sink, gpointer self) {
+  StreamPipeline* me = reinterpret_cast<StreamPipeline*>(self);
+
+  GstBuffer* buffer = gst_app_sink_pull_buffer((GstAppSink*)app_sink);
+  me->buffer_.append((const char*)buffer->data, buffer->size);
+  gst_buffer_unref(buffer);
+
+  emit me->readyRead();
+}
+
+void StreamPipeline::NewPadCallback(
+    GstElement* decodebin, GstPad* pad, gpointer self) {
+  qLog(Debug) << "Linking pads";
+  StreamPipeline* me = reinterpret_cast<StreamPipeline*>(self);
+  GstPad* const audiopad = gst_element_get_pad(me->convert_bin_, "sink");
+
+  gst_pad_link(pad, audiopad);
+  gst_object_unref(audiopad);
+
+  QMetaObject::invokeMethod(me, "StartPipeline", Qt::QueuedConnection);
+}
+
+void StreamPipeline::StartPipeline() {
+  gst_element_set_state(pipeline_, GST_STATE_PLAYING);
+}

src/remote/streampipeline.h

@@ -0,0 +1,47 @@
+#ifndef STREAMPIPELINE_H
+#define STREAMPIPELINE_H
+
+#include <QIODevice>
+
+#include <gst/gst.h>
+
+#include "core/song.h"
+
+// Transcodes a song into an OGG Vorbis stream.
+class StreamPipeline : public QIODevice {
+  Q_OBJECT
+ public:
+  StreamPipeline(QObject* parent = 0);
+  virtual ~StreamPipeline();
+
+  void Init(const Song& song);
+
+ private slots:
+  void StartPipeline();
+
+ protected:
+  // QIODevice
+  virtual qint64 readData(char* data, qint64 max_size);
+  virtual qint64 writeData(const char* data, qint64 max_size);
+
+ private:
+  GstElement* CreateElement(
+      const QString& factory_name, GstElement* bin = NULL);
+  static void HandleMessage(GstMessage* msg);
+
+  GstElement* pipeline_;
+  GstElement* app_sink_;
+  GstElement* convert_bin_;
+  int bus_cb_id_;
+
+  QByteArray buffer_;
+
+  static gboolean BusCallback(GstBus*, GstMessage*, gpointer);
+  static GstBusSyncReply BusCallbackSync(GstBus*, GstMessage*, gpointer);
+  static void NewBufferCallback(GstElement*, gpointer);
+  static void NewPadCallback(GstElement*, GstPad*, gpointer);
+
+  static const char* kPipeline;
+};
+
+#endif