gstreamer.h

Go to the documentation of this file.

#pragma once
 
// gstreamer
#include "gstreamer_options.h"
#include "gstreamerConventions.h"
 
// gemc
#include <gemc/gdata/event/gEventDataCollection.h>
#include <gemc/gdata/run/gRunDataCollection.h>
#include <gemc/gdata/frame/gFrameDataCollection.h>
#include <gemc/gfactory/gfactory.h>
#include <gemc/gbase/gbase.h>
 
// c++
#include <string>
#include <vector>
#include <map>


class GStreamer : public GBase<GStreamer>
{
public:
    explicit GStreamer(const std::shared_ptr<GOptions>& g) : GBase(g, GSTREAMER_LOGGER) {
    }

    virtual ~GStreamer() = default;

    [[nodiscard]] virtual bool openConnection() { return false; }

    [[nodiscard]] bool closeConnection() {
        flushEventBuffer();
        return closeConnectionImpl();
    }

    [[nodiscard]] virtual bool closeConnectionImpl() { return false; }

    void publishEventData(const std::shared_ptr<GEventDataCollection>& event_data);

    void publishRunData(const std::shared_ptr<GRunDataCollection>& run_data);

    [[nodiscard]] inline std::string getStreamType() const { return gstreamer_definitions.type; }

    inline void define_gstreamer(const GStreamerDefinition& gstreamerDefinition, int tid = -1) {
        gstreamer_definitions = GStreamerDefinition(gstreamerDefinition, tid);
    }

    static const std::vector<std::string>& supported_formats();

    static bool is_valid_format(const std::string& format);

    void set_loggers(const std::shared_ptr<GOptions>& g) {
        bufferFlushLimit = g->getRequiredScalarInt("ebuffer");
    }
 
protected:
    GStreamerDefinition gstreamer_definitions;

    [[nodiscard]] bool startEvent([[maybe_unused]] const std::shared_ptr<GEventDataCollection>& event_data) {
        if (!event_data) {
            log->error(gstreamer::ERR_PUBLISH_ERROR, "eventData is null in GStreamer::startEvent");
        }
        if (!event_data->getHeader()) {
            log->error(gstreamer::ERR_PUBLISH_ERROR, "event header is null in GStreamer::startEvent");
        }
 
        log->debug(NORMAL, "GStreamer::startEvent");
        return startEventImpl(event_data);
    }

    virtual bool startEventImpl([[maybe_unused]] const std::shared_ptr<GEventDataCollection>& event_data) {
        return false;
    }

    [[nodiscard]] bool endEvent([[maybe_unused]] const std::shared_ptr<GEventDataCollection>& event_data) {
        log->debug(NORMAL, "GStreamer::endEvent");
        return endEventImpl(event_data);
    }

    virtual bool endEventImpl([[maybe_unused]] const std::shared_ptr<GEventDataCollection>& event_data) {
        return false;
    }

    [[nodiscard]] bool publishEventHeader([[maybe_unused]] const std::unique_ptr<GEventHeader>& gevent_header) {
        if (!gevent_header) {
            log->error(gstreamer::ERR_PUBLISH_ERROR, "event header is null in GStreamer::publishEventHeader");
        }
        log->debug(NORMAL, "GStreamer::publishEventHeader");
        return publishEventHeaderImpl(gevent_header);
    }

    virtual bool publishEventHeaderImpl([[maybe_unused]] const std::unique_ptr<GEventHeader>& gevent_header) {
        return false;
    }

    [[nodiscard]] bool publishEventTrueInfoData([[maybe_unused]] const std::string& detectorName,
                                                [[maybe_unused]] const std::vector<const GTrueInfoData*>& trueInfoData) {
        log->debug(NORMAL, "GStreamer::publishEventTrueInfoData for detector %1", detectorName);
        return publishEventTrueInfoDataImpl(detectorName, trueInfoData);
    }

    virtual bool publishEventTrueInfoDataImpl([[maybe_unused]] const std::string& detectorName,
                                              [[maybe_unused]] const std::vector<const GTrueInfoData*>& trueInfoData) {
        return false;
    }

    [[nodiscard]] bool publishEventDigitizedData([[maybe_unused]] const std::string& detectorName,
                                                 [[maybe_unused]] const std::vector<const GDigitizedData*>& digitizedData) {
        log->debug(NORMAL, "GStreamer::publishEventDigitizedData for detector %1", detectorName);
        return publishEventDigitizedDataImpl(detectorName, digitizedData);
    }
 
    [[nodiscard]] bool publishEventGeneratedParticles([[maybe_unused]] const std::string& bankName,
                                                      [[maybe_unused]] const GGeneratedParticleBank& particles) {
        log->debug(NORMAL, "GStreamer::publishEventGeneratedParticles for bank %1", bankName);
        return publishEventGeneratedParticlesImpl(bankName, particles);
    }

    [[nodiscard]] bool publishEventAncestors(const GAncestorBank& ancestors) {
        log->debug(NORMAL, "GStreamer::publishEventAncestors");
        return publishEventAncestorsImpl(ancestors);
    }

    virtual bool publishEventGeneratedParticlesImpl([[maybe_unused]] const std::string& bankName,
                                                    [[maybe_unused]] const GGeneratedParticleBank& particles) {
        return true;
    }

    virtual bool publishEventAncestorsImpl([[maybe_unused]] const GAncestorBank& ancestors) { return true; }

    virtual bool publishEventDigitizedDataImpl([[maybe_unused]] const std::string& detectorName,
                                               [[maybe_unused]] const std::vector<const GDigitizedData*>& digitizedData) {
        return false;
    }

    [[nodiscard]] bool startRun([[maybe_unused]] const std::shared_ptr<GRunDataCollection>& run_data) {
        if (!run_data) { log->error(gstreamer::ERR_PUBLISH_ERROR, "run_data is null in GStreamer::startRun"); }
        if (!run_data->getHeader()) {
            log->error(gstreamer::ERR_PUBLISH_ERROR, "run header is null in GStreamer::startRun");
        }
 
        log->debug(NORMAL, "GStreamer::startRun");
        return startRunImpl(run_data);
    }

    virtual bool startRunImpl([[maybe_unused]] const std::shared_ptr<GRunDataCollection>& run_data) {
        return false;
    }

    [[nodiscard]] bool endRun([[maybe_unused]] const std::shared_ptr<GRunDataCollection>& run_data) {
        log->debug(NORMAL, "GStreamer::endRun");
        return endRunImpl(run_data);
    }

    virtual bool endRunImpl([[maybe_unused]] const std::shared_ptr<GRunDataCollection>& run_data) {
        return false;
    }

    bool publishRunHeader([[maybe_unused]] const std::unique_ptr<GRunHeader>& run_header) {
        log->debug(NORMAL, "GStreamer::publishRunHeader");
        return publishRunHeaderImpl(run_header);
    }

    virtual bool publishRunHeaderImpl([[maybe_unused]] const std::unique_ptr<GRunHeader>& run_header) {
        return false;
    }

    bool publishRunDigitizedData([[maybe_unused]] const std::string& detectorName,
                                 [[maybe_unused]] const std::vector<const GDigitizedData*>& digitizedData) {
        log->debug(NORMAL, "GStreamer::publishRunDigitizedData for detector %1", detectorName);
        return publishRunDigitizedDataImpl(detectorName, digitizedData);
    }

    virtual bool publishRunDigitizedDataImpl([[maybe_unused]] const std::string& detectorName,
                                             [[maybe_unused]] const std::vector<const GDigitizedData*>& digitizedData) {
        return false;
    }

    [[nodiscard]] bool startStream([[maybe_unused]] const GFrameDataCollection* frameRunData) {
        flushEventBuffer();
        log->debug(NORMAL, "GStreamer::startStream");
        return startStreamImpl(frameRunData);
    }

    virtual bool startStreamImpl([[maybe_unused]] const GFrameDataCollection* frameRunData) { return false; }

    [[nodiscard]] bool publishFrameHeader([[maybe_unused]] const GFrameHeader* gframeHeader) {
        log->debug(NORMAL, "GStreamer::publishFrameHeader");
        return publishFrameHeaderImpl(gframeHeader);
    }

    virtual bool publishFrameHeaderImpl([[maybe_unused]] const GFrameHeader* gframeHeader) { return false; }

    [[nodiscard]] bool publishPayload([[maybe_unused]] const std::vector<GIntegralPayload*>* payload) {
        log->debug(NORMAL, "GStreamer::publishPayload");
        return publishPayloadImpl(payload);
    }

    virtual bool publishPayloadImpl([[maybe_unused]] const std::vector<GIntegralPayload*>* payload) { return false; }

    [[nodiscard]] bool endStream([[maybe_unused]] const GFrameDataCollection* frameRunData) {
        log->debug(NORMAL, "GStreamer::endStream");
        return endStreamImpl(frameRunData);
    }

    virtual bool endStreamImpl([[maybe_unused]] const GFrameDataCollection* frameRunData) { return false; }

    void flushEventBuffer();
 
private:
    [[nodiscard]] virtual std::string filename() const = 0;

    std::vector<std::shared_ptr<GEventDataCollection>> eventBuffer;

    size_t bufferFlushLimit = 10;
 
public:
    static GStreamer* instantiate(const dlhandle h, std::shared_ptr<GOptions> g) {
        if (!h) return nullptr;
        using fptr = GStreamer* (*)(std::shared_ptr<GOptions>);
 
        auto sym = dlsym(h, "GStreamerFactory");
        if (!sym) return nullptr;
 
        auto func = reinterpret_cast<fptr>(sym);
        return func(g);
    }
};
 
 
namespace gstreamer {
 
    using gstreamersMap = std::unordered_map<std::string, std::shared_ptr<GStreamer>>;

    inline std::shared_ptr<const gstreamersMap> gstreamersMapPtr(const std::shared_ptr<GOptions>& gopts,
                                                                 int                              thread_id = -1) {
        auto log = std::make_shared<GLogger>(gopts, "gstreamersMap worker for thread id" + std::to_string(thread_id),
                                             GSTREAMER_LOGGER);
 
        GManager manager(gopts);
 
        auto gstreamers = std::make_shared<gstreamersMap>();
 
        for (const auto& gstreamer_def : gstreamer::getGStreamerDefinition(gopts)) {
            if (gstreamer_def.format == "sro") { continue; } // Shared GSROFactory owns crate output.
            auto        gstreamer_def_thread = GStreamerDefinition(gstreamer_def, thread_id);
            std::string gstreamer_plugin     = gstreamer_def_thread.gstreamerPluginName();
            // Key the map by the format-derived plugin name combined with the per-output
            // rootname so that neither multiple same-format outputs (e.g. two csv files)
            // nor multiple same-rootname outputs of different formats (e.g. dc.csv and
            // dc.hipo) collide; the plugin library is still loaded by the plugin name.
            const std::string output_key    = gstreamer_plugin + ":" + gstreamer_def_thread.rootname;
 
            // Load the plugin object for this configured output. Each call returns a
            // fresh GStreamer instance, so same-format outputs stay independent.
            auto streamer = manager.LoadAndRegisterObjectFromLibrary<GStreamer>(gstreamer_plugin, gopts);
            auto [it, inserted] = gstreamers->emplace(output_key, streamer);
            if (!inserted) {
                log->warning("duplicate gstreamer output name '%1' - ignoring later definition", output_key);
                continue;
            }
 
            // Bind the thread-specialized definition to this (freshly created) plugin instance.
            it->second->define_gstreamer(gstreamer_def_thread);
        }
 
        return gstreamers;
    }

    inline std::shared_ptr<const gstreamersMap> preloadGStreamerPlugins(const std::shared_ptr<GOptions>& gopts) {
        if (gstreamer::getGStreamerDefinition(gopts).empty()) { return std::make_shared<gstreamersMap>(); }
        return gstreamersMapPtr(gopts);
    }
 
} // namespace gstreamer