pipeline.hpp

Go to the documentation of this file.

/*
 * SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: LicenseRef-NvidiaProprietary
 *
 * NVIDIA CORPORATION, its affiliates and licensors retain all intellectual
 * property and proprietary rights in and to this material, related
 * documentation and any modifications thereto. Any use, reproduction,
 * disclosure or distribution of this material and related documentation
 * without an express license agreement from NVIDIA CORPORATION or
 * its affiliates is strictly prohibited.
 */
 
#ifndef PIPELINE_HPP
#define PIPELINE_HPP
 
#include <string>
#include <memory>
#include <thread>
 
#include "object.hpp"
#include "element.hpp"
#include "buffer_probe.hpp"
#include "signal_handler.hpp"
 
namespace deepstream
{
 
class Pipeline : public Object {
 public:
 
  typedef enum {
    INVALID,
    EMPTY,
    READY,
    PAUSED,
    PLAYING
  } State;
 
  class Message {
    public:
      virtual ~Message() {}
      uint32_t type() { return type_; }
    protected:
      uint32_t type_;
 
      Message(uint32_t type): type_(type) {}
  };
 
  class EOSMessage : public Message {
    public:
      EOSMessage(void*);
  };
 
  class DynamicSourceMessage : public Message {
    public:
      DynamicSourceMessage(void*);
 
      inline bool isSourceAdded() const { return source_added_; }
      inline uint32_t getSourceId() const { return source_id_; }
      inline const std::string& getSensorId() const { return sensor_id_; }
      inline const std::string& getSensorName() const { return sensor_name_; }
      inline const std::string& getUri() const { return uri_; }
 
    protected:
      bool source_added_ = false;
      uint32_t source_id_ = 0;
      std::string sensor_id_;
      std::string sensor_name_;
      std::string uri_;
  };
 
  class StateTransitionMessage : public Message {
    public:
      StateTransitionMessage(void*);
      inline const std::string& getName() const { return name_; };
      inline void getState(State& old_state, State& new_state) const {
        old_state = old_state_;
        new_state = new_state_;
      }
    protected:
      State old_state_;
      State new_state_;
      std::string name_;
  };
 
  Pipeline(const char* name);
 
  Pipeline(const char* name, const std::string& config_file);
 
  virtual ~Pipeline();
 
  template<typename... Args>
  Pipeline& add(const std::string& element_type, const std::string& element_name, const Args&... args) {
    Element element = Element(element_type, element_name);
    if constexpr (sizeof...(args) > 0) {
      element.set(args...);
    }
    return this->add(element);
  }
 
  Pipeline& add(Element element);
 
  Element* find(const::std::string& name);
 
  Element& operator[](const std::string& name) {
    Element* e = this->find(name);
    if (e == nullptr) {
      throw std::runtime_error(name + " Not found");
    }
    return *e;
  }
 
  Pipeline& link(
    std::pair<std::string, std::string> route,
    std::pair<std::string, std::string> tips
  );
 
  template <typename... Args>
  Pipeline &link(const std::string &arg1, const std::string arg2, const Args &...args)
  {
    link(std::make_pair(arg1, arg2), std::make_pair("", ""));
    if constexpr (sizeof...(args) > 0) {
        return this->link(arg2, args...);
    } else {
        return *this;
    }
  }
 
  Pipeline& attach(const std::string& elmenent_name, CustomObject* object, const std::string tip="");
 
  Pipeline& attach(
    const std::string& element_name,
    const std::string& plugin_name,
    const std::string& object_name,
    const std::string tip="");
 
  template<typename... Args>
  Pipeline& attach(
    const std::string& element_name,
    const std::string& plugin_name,
    const std::string& object_name,
    const std::string tip,
    const Args&... args) {
    attach(element_name, plugin_name, object_name, tip);
    if (sizeof...(args) > 0) {
      auto& element = (*this)[element_name];
      if (auto handler = element.getSignalHandler(object_name)) {
        handler->set(args...);
      } else if (auto probe = element.getProbe(object_name)) {
        probe->set(args...);
      }
    }
    return *this;
  }
 
  Pipeline& install(std::function<void(Pipeline&, int key)> keyboard_listener) {
    keyboard_listener_ = keyboard_listener;
    return *this;
  }
 
  int prepare();
  int prepare(std::function<void(Pipeline &, const Message &)> listener);
 
  Pipeline& start();
  Pipeline& start(std::function<void(Pipeline&, const Message&)> listener);
  Pipeline &activate();
  Pipeline& wait();
 
  Pipeline& stop();
 
  bool isRunning() const;
 
  Pipeline& pause();
  Pipeline& resume();
  Pipeline& seek(uint64_t timestamp);
 
  uint32_t startRTSP(uint16_t rtsp_port, uint16_t udp_port, uint32_t buffer_size=0);
 
  void handleKey(int key);
 
protected:
  int run();
  int run_after_prepare();
 
  // This is GMainLoop*
  void *loop_ = NULL;
  // gstreamer bus watch id
  uint bus_watch_id_ = 0;
  // gstreamer bus data
  void *bus_data_ = NULL;
  std::function<void(Pipeline&, int key)> keyboard_listener_;
 
  // Thread
  std::thread thread_;
  // Hold all the element references here
  std::map<std::string, Element> elements_;
  // Hold the signal emitters for action control
  std::map<std::string, std::unique_ptr<SignalEmitter>> action_owners_;
  // start timestamp for the whole pipeline
  uint64_t start_pts_ = 0;
};
 
} // namespace deepstream
 
#endif