126 lines
3.7 KiB
C++
126 lines
3.7 KiB
C++
#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);
|
|
}
|