vizionStreamer/GStreamerPipeline.cpp

#include "GStreamerPipeline.h"
#include <iostream>
#include <cstring>

GStreamerPipeline::GStreamerPipeline(const std::string& pipelineDescription)
    : pipeline_(nullptr), appsrc_(nullptr), bus_(nullptr), running_(false),
      pipelineDescription_(pipelineDescription), width_(0), height_(0) {
    gst_init(nullptr, nullptr);
}

GStreamerPipeline::~GStreamerPipeline() {
    stop();
}

void GStreamerPipeline::setPipelineDescription(const std::string& description) {
    if (!running_) {
        pipelineDescription_ = description;
    }
}

bool GStreamerPipeline::start() {
    if (running_) {
        std::cerr << "GStreamer pipeline already running" << std::endl;
        return false;
    }

    if (pipelineDescription_.empty()) {
        std::cerr << "Pipeline description is empty" << std::endl;
        return false;
    }

    GError* error = nullptr;
    std::string fullPipeline = "appsrc name=source ! " + pipelineDescription_;

    pipeline_ = gst_parse_launch(fullPipeline.c_str(), &error);
    if (error) {
        std::cerr << "Failed to create pipeline: " << error->message << std::endl;
        g_error_free(error);
        return false;
    }

    appsrc_ = gst_bin_get_by_name(GST_BIN(pipeline_), "source");
    if (!appsrc_) {
        std::cerr << "Failed to get appsrc element" << std::endl;
        gst_object_unref(pipeline_);
        return false;
    }

    // Configure appsrc
    g_object_set(G_OBJECT(appsrc_),
                 "stream-type", GST_APP_STREAM_TYPE_STREAM,
                 "format", GST_FORMAT_TIME,
                 "is-live", TRUE,
                 nullptr);

    // Set callbacks
    GstAppSrcCallbacks callbacks;
    callbacks.need_data = onNeedData;
    callbacks.enough_data = onEnoughData;
    callbacks.seek_data = nullptr;
    gst_app_src_set_callbacks(GST_APP_SRC(appsrc_), &callbacks, this, nullptr);

    // Start the pipeline
    GstStateChangeReturn ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
    if (ret == GST_STATE_CHANGE_FAILURE) {
        std::cerr << "Failed to start pipeline" << std::endl;
        gst_object_unref(appsrc_);
        gst_object_unref(pipeline_);
        return false;
    }

    bus_ = gst_element_get_bus(pipeline_);
    running_ = true;

    std::cout << "GStreamer pipeline started: " << fullPipeline << std::endl;
    return true;
}

void GStreamerPipeline::stop() {
    if (!running_) {
        return;
    }

    running_ = false;

    if (appsrc_) {
        gst_app_src_end_of_stream(GST_APP_SRC(appsrc_));
    }

    if (pipeline_) {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
    }

    if (appsrc_) {
        gst_object_unref(appsrc_);
        appsrc_ = nullptr;
    }

    if (bus_) {
        gst_object_unref(bus_);
        bus_ = nullptr;
    }

    std::cout << "GStreamer pipeline stopped" << std::endl;
}

bool GStreamerPipeline::pushBuffer(uint8_t* data, size_t size, int width, int height, const std::string& format) {
    if (!running_ || !appsrc_) {
        return false;
    }

    // Update format if changed
    if (width != width_ || height != height_ || format != format_) {
        width_ = width;
        height_ = height;
        format_ = format;

        // Set caps based on format
        std::string capsStr;
        if (format == "YUY2" || format == "UYVY") {
            capsStr = "video/x-raw,format=" + format + ",width=" + std::to_string(width) +
                     ",height=" + std::to_string(height) + ",framerate=30/1";
        } else if (format == "MJPG") {
            capsStr = "image/jpeg,width=" + std::to_string(width) +
                     ",height=" + std::to_string(height) + ",framerate=30/1";
        } else if (format == "BGR" || format == "RGB") {
            capsStr = "video/x-raw,format=" + format + ",width=" + std::to_string(width) +
                     ",height=" + std::to_string(height) + ",framerate=30/1";
        } else {
            capsStr = "video/x-raw,width=" + std::to_string(width) +
                     ",height=" + std::to_string(height) + ",framerate=30/1";
        }

        GstCaps* caps = gst_caps_from_string(capsStr.c_str());
        gst_app_src_set_caps(GST_APP_SRC(appsrc_), caps);
        gst_caps_unref(caps);
    }

    // Create buffer and copy data
    GstBuffer* buffer = gst_buffer_new_allocate(nullptr, size, nullptr);
    GstMapInfo map;
    gst_buffer_map(buffer, &map, GST_MAP_WRITE);
    memcpy(map.data, data, size);
    gst_buffer_unmap(buffer, &map);

    // Push buffer to pipeline
    GstFlowReturn ret = gst_app_src_push_buffer(GST_APP_SRC(appsrc_), buffer);
    if (ret != GST_FLOW_OK) {
        std::cerr << "Failed to push buffer to pipeline: " << ret << std::endl;
        return false;
    }

    return true;
}

void GStreamerPipeline::onNeedData(GstAppSrc* appsrc, guint unused, gpointer user_data) {
    // Called when pipeline needs more data
}

void GStreamerPipeline::onEnoughData(GstAppSrc* appsrc, gpointer user_data) {
    // Called when pipeline has enough data buffered
}