How to integrate splitmux into deepstream-test5-app

pandian · June 7, 2023, 8:46am

Please provide complete information as applicable to your setup.

• Hardware Platform (Jetson / GPU) - jetson nano
• DeepStream Version - 6.0.1
• JetPack Version (valid for Jetson only) - 4.6.3
• TensorRT Version - 8.4
• NVIDIA GPU Driver Version (valid for GPU only) - 10.2
• Issue Type( questions, new requirements, bugs) - the question is about how to integrate splitmux into deepstream-test5-app
• How to reproduce the issue ? (This is for bugs. Including which sample app is using, the configuration files content, the command line used and other details for reproducing) - i would like do split video in deepstream test -5 smart record how to integrate splitmux into deep stream test5-app
• Requirement details( This is for new requirement. Including the module name-for which plugin or for which sample application, the function description)

jetson@ubuntu:/opt/nvidia/deepstream/deepstream-6.0/sources/apps/sample_apps/deepstream-test51-app$ sudo make
cc -c -o src/deepstream_sink_bin.o -DPLATFORM_TEGRA -I…/…/apps-common/includes -I…/…/…/includes -I…/deepstream-app/ -DDS_VERSION_MINOR=0 -DDS_VERSION_MAJOR=5 -I /usr/local/cuda-10.2/include -I include -I includes -pthread -I/usr/include/gstreamer-1.0 -I/usr/include/orc-0.4 -I/usr/include/gstreamer-1.0 -I/usr/include/json-glib-1.0 -I/usr/include/glib-2.0 -I/usr/lib/aarch64-linux-gnu/glib-2.0/include src/deepstream_sink_bin.c
src/deepstream_sink_bin.c: In function ‘create_encode_file_bin’:
src/deepstream_sink_bin.c:427:10: error: ‘NV_DS_CONTAINER_HLS’ undeclared (first use in this function); did you mean ‘NV_DS_CONTAINER_MKV’?
case NV_DS_CONTAINER_HLS:
^~~~~~~~~~~~~~~~~~~
NV_DS_CONTAINER_MKV
src/deepstream_sink_bin.c:427:10: note: each undeclared identifier is reported only once for each function it appears in
src/deepstream_sink_bin.c:428:44: error: ‘NVDS_ELEM_HILSINK’ undeclared (first use in this function); did you mean ‘NVDS_ELEM_TILER’?
bin->mux = gst_element_factory_make (NVDS_ELEM_HILSINK, “SINK_SUB_BIN_HLS_MUX”);
^~~~~~~~~~~~~~~~~
NVDS_ELEM_TILER
src/deepstream_sink_bin.c:429:10: error: ‘NV_DS_CONTAINER_SM’ undeclared (first use in this function); did you mean ‘NV_DS_CONTAINER_HLS’?
case NV_DS_CONTAINER_SM:
^~~~~~~~~~~~~~~~~~
NV_DS_CONTAINER_HLS
src/deepstream_sink_bin.c:430:44: error: ‘NVDS_ELEM_SPLITMUX’ undeclared (first use in this function); did you mean ‘NVDS_ELEM_PGIE’?
bin->mux = gst_element_factory_make (NVDS_ELEM_SPLITMUX, “SPLIT_MUX”);
^~~~~~~~~~~~~~~~~~
NVDS_ELEM_PGIE
Makefile:70: recipe for target ‘src/deepstream_sink_bin.o’ failed
make: *** [src/deepstream_sink_bin.o] Error 1

yuweiw · June 7, 2023, 10:48am

Did you add the NV_DS_CONTAINER_HLS by yourself? You need to modify more source code.
1.add splitmux tag in the config file: sources\apps\sample_apps\deepstream-test5\configs\test5_config_file_src_infer.txt
2.add parsing code for splitmux tag: sources\apps\apps-common\src\deepstream_config_file_parser.c
3.create splixmux plugin based on the parsed tag: sources\apps\apps-common\src\deepstream_sink_bin.c

pandian · June 7, 2023, 10:49am

ok

pandian · June 7, 2023, 11:30am

where can i refer it for doing this

pandian · June 7, 2023, 1:39pm

1.now i am only use this configuration file -test5_dec_infer-resnet_tracker_sgie_tiled_display_int8_1.txt why should i update in sources\apps\sample_apps\deepstream-test5\configs\test5_config_file_src_infer.txt

2.what should i define in sources\apps\apps-common\src\deepstream_config_file_parser.c instead of this i have defined in deepstream_config.h

3.I have created plugin for splitmux plugin deepstream_sink_bin.c file

yuweiw · June 8, 2023, 2:02am

There is currently no similar reference as the deepstream app demo is open source.
What I posted earlier was just a simple process, you need to add code based on your actual scenario.
Since you have created this plugin, have you had any issues during the runtime?

pandian · June 8, 2023, 7:30am

jetson@ubuntu:/opt/nvidia/deepstream/deepstream-6.0/sources/apps/sample_apps/deepstream-test51-app$ sudo make
cc -c -o src/deepstream_sink_bin.o -DPLATFORM_TEGRA -I…/…/apps-common/includes -I…/…/…/includes -I…/deepstream-app/ -DDS_VERSION_MINOR=0 -DDS_VERSION_MAJOR=5 -I /usr/local/cuda-10.2/include -I include -I includes -pthread -I/usr/include/gstreamer-1.0 -I/usr/include/orc-0.4 -I/usr/include/gstreamer-1.0 -I/usr/include/json-glib-1.0 -I/usr/include/glib-2.0 -I/usr/lib/aarch64-linux-gnu/glib-2.0/include src/deepstream_sink_bin.c
src/deepstream_sink_bin.c: In function ‘create_encode_file_bin’:
src/deepstream_sink_bin.c:434:6: error: ‘NvDsSinkBinSubBin {aka struct }’ has no member named ‘splitmuxsink’
bin->splitmuxsink = gst_element_factory_make(“splitmuxsink”, “splitmuxsink”);
^~
src/deepstream_sink_bin.c:435:11: error: ‘NvDsSinkBinSubBin {aka struct }’ has no member named ‘splitmuxsink’
if (!bin->splitmuxsink) {
^~
In file included from /usr/include/glib-2.0/gobject/gobject.h:24:0,
from /usr/include/glib-2.0/gobject/gbinding.h:29,
from /usr/include/glib-2.0/glib-object.h:23,
from /usr/include/gstreamer-1.0/gst/gstenumtypes.h:7,
from /usr/include/gstreamer-1.0/gst/gst.h:31,
from …/…/apps-common/includes/deepstream_common.h:31,
from src/deepstream_sink_bin.c:30:
src/deepstream_sink_bin.c:438:28: error: ‘NvDsSinkBinSubBin {aka struct }’ has no member named ‘splitmuxsink’
g_object_set(G_OBJECT(bin->splitmuxsink), “location”, config->output_file_path,
^
/usr/include/glib-2.0/gobject/gtype.h:2224:57: note: in definition of macro ‘_G_TYPE_CIC’
((ct*) g_type_check_instance_cast ((GTypeInstance*) ip, gt))
^~
/usr/include/glib-2.0/gobject/gobject.h:51:38: note: in expansion of macro ‘G_TYPE_CHECK_INSTANCE_CAST’
#define G_OBJECT(object) (G_TYPE_CHECK_INSTANCE_CAST ((object), G_TYPE_OBJECT, GObject))
^~~~~~~~~~~~~~~~~~~~~~~~~~
src/deepstream_sink_bin.c:438:16: note: in expansion of macro ‘G_OBJECT’
g_object_set(G_OBJECT(bin->splitmuxsink), “location”, config->output_file_path,
^~~~~~~~
Makefile:70: recipe for target ‘src/deepstream_sink_bin.o’ failed
make: *** [src/deepstream_sink_bin.o] Error 1
jetson@ubuntu:/opt/nvidia/deepstream/deepstream-6.0/sources/apps/sample_apps/deepstream-test51-app$ ^C

after adding splitmux into deepstream_sink_bin.c getting this error

this ais a deepstream_sinks.h

#ifndef NVGSTDS_SINKS_H
#define NVGSTDS_SINKS_H

#ifdef __cplusplus
extern “C”
{
#endif

#include <gst/gst.h>

typedef enum
{
NV_DS_SINK_FAKE = 1,
NV_DS_SINK_RENDER_EGL,
NV_DS_SINK_ENCODE_FILE,
NV_DS_SINK_UDPSINK,
NV_DS_SINK_RENDER_OVERLAY,
NV_DS_SINK_MSG_CONV_BROKER,
NV_DS_SINK_SPLITMUX,
} NvDsSinkType;

typedef enum
{
NV_DS_CONTAINER_MP4 = 1,
NV_DS_CONTAINER_MKV,
NV_DS_CONTAINER_SPLITMUX
} NvDsContainerType;

typedef enum
{
NV_DS_ENCODER_H264 = 1,
NV_DS_ENCODER_H265,
NV_DS_ENCODER_MPEG4,
NV_DS_ENCODER_SPLITMUX
} NvDsEncoderType;

typedef enum
{
NV_DS_ENCODER_TYPE_HW,
NV_DS_ENCODER_TYPE_SW
}NvDsEncHwSwType;

typedef struct
{
NvDsSinkType type;
NvDsContainerType container;
NvDsEncoderType codec;
NvDsEncHwSwType enc_type;
gint bitrate;
guint profile;
gint sync;
gchar *output_file_path;
guint gpu_id;
guint rtsp_port;
guint udp_port;
guint64 udp_buffer_size;
guint iframeinterval;
guint splitmux_interval;
gchar *splitmux_output_path;
} NvDsSinkEncoderConfig;

typedef struct
{
NvDsSinkType type;
gint width;
gint height;
gint sync;
gboolean qos;
gboolean qos_value_specified;
guint gpu_id;
guint nvbuf_memory_type;
guint display_id;
guint overlay_id;
guint offset_x;
guint offset_y;
} NvDsSinkRenderConfig;

typedef struct
{
gboolean enable;
/** MsgConv settings /
gchar config_file_path;
guint conv_payload_type;
gchar* conv_msg2p_lib;
guint conv_comp_id;
gchar* debug_payload_dir;
gboolean multiple_payloads;
gboolean conv_msg2p_new_api;
guint conv_frame_interval;
/** Broker settings /
gchar proto_lib;
gchar* conn_str;
gchar* topic;
gchar* broker_config_file_path;
guint broker_comp_id;
gboolean disable_msgconv;
gint sync;
gboolean new_api;
} NvDsSinkMsgConvBrokerConfig;

typedef struct
{
gboolean enable;
guint source_id;
gboolean link_to_demux;
NvDsSinkType type;
gint sync;
NvDsSinkEncoderConfig encoder_config;
NvDsSinkRenderConfig render_config;
NvDsSinkMsgConvBrokerConfig msg_conv_broker_config;
NvDsSinkMsgConvBrokerConfig splitmux_config
} NvDsSinkSubBinConfig;

typedef struct
{
GstElement *bin;
GstElement *queue;
GstElement *transform;
GstElement *cap_filter;
GstElement *enc_caps_filter;
GstElement *encoder;
GstElement *codecparse;
GstElement *mux;
GstElement *sink;
GstElement *splitmuxsink;
GstElement *rtppay;
gulong sink_buffer_probe;
} NvDsSinkBinSubBin;

typedef struct
{
GstElement *bin;
GstElement *queue;
GstElement *tee;

gint num_bins;
NvDsSinkBinSubBin sub_bins[MAX_SINK_BINS];
} NvDsSinkBin;

/**

Initialize @ref NvDsSinkBin. It creates and adds sink and
other elements needed for processing to the bin.
It also sets properties mentioned in the configuration file under
group @ref CONFIG_GROUP_SINK
@param[in] num_sub_bins number of sink elements.
@param[in] config_array array of pointers of type @ref NvDsSinkSubBinConfig
```
       parsed from configuration file.
```
@param[in] bin pointer to @ref NvDsSinkBin to be filled.
@param[in] index id of source element.
@return true if bin created successfully.
*/
gboolean create_sink_bin (guint num_sub_bins,
NvDsSinkSubBinConfig *config_array, NvDsSinkBin *bin, guint index);

void destroy_sink_bin (void);
gboolean create_demux_sink_bin (guint num_sub_bins,
NvDsSinkSubBinConfig *config_array, NvDsSinkBin *bin, guint index);

void set_rtsp_udp_port_num (guint rtsp_port_num, guint udp_port_num);

#ifdef __cplusplus
}
#endif

#endif

this a deepstream_sink_bin.c

#include <sys/types.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/wait.h>

#include “deepstream_common.h”
#include “deepstream_sinks.h”
#include <gst/rtsp-server/rtsp-server.h>
#include <cuda_runtime_api.h>

static guint uid = 0;
static GstRTSPServer *server [MAX_SINK_BINS];
static guint server_count = 0;
static GMutex server_cnt_lock;

GST_DEBUG_CATEGORY_EXTERN (NVDS_APP);

/**

Function to create sink bin for Display / Fakesink.
*/
static gboolean
create_render_bin (NvDsSinkRenderConfig * config, NvDsSinkBinSubBin * bin)
{
gboolean ret = FALSE;
gchar elem_name[50];
GstElement *connect_to;
GstCaps *caps = NULL;

uid++;

struct cudaDeviceProp prop;
cudaGetDeviceProperties(&prop, config->gpu_id);

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin%d”, uid);
bin->bin = gst_bin_new (elem_name);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_sink%d”, uid);
switch (config->type) {
case NV_DS_SINK_RENDER_EGL:
GST_CAT_INFO (NVDS_APP, “NVvideo renderer\n”);
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_EGL, elem_name);
g_object_set (G_OBJECT (bin->sink), “window-x”, config->offset_x,
“window-y”, config->offset_y, “window-width”, config->width,
“window-height”, config->height, NULL);
g_object_set (G_OBJECT (bin->sink), “enable-last-sample”, FALSE, NULL);
break;
case NV_DS_SINK_RENDER_OVERLAY:
#ifndef IS_TEGRA
NVGSTDS_ERR_MSG_V (“Overlay is only supported for Jetson”);
return FALSE;
#endif
GST_CAT_INFO (NVDS_APP, “NVvideo renderer\n”);
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_OVERLAY, elem_name);
g_object_set (G_OBJECT (bin->sink), “display-id”, config->display_id,
NULL);
g_object_set (G_OBJECT (bin->sink), “overlay”, config->overlay_id, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-x”, config->offset_x, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-y”, config->offset_y, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-w”, config->width, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-h”, config->height, NULL);
break;
case NV_DS_SINK_FAKE:
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_FAKESINK, elem_name);
g_object_set (G_OBJECT (bin->sink), “enable-last-sample”, FALSE, NULL);
break;
default:
return FALSE;
}

if (!bin->sink) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_object_set (G_OBJECT (bin->sink), “sync”, config->sync, “max-lateness”, -1,
“async”, FALSE, “qos”, config->qos, NULL);

if(!prop.integrated) {
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_cap_filter%d”, uid);
bin->cap_filter = gst_element_factory_make (NVDS_ELEM_CAPS_FILTER, elem_name);
if (!bin->cap_filter) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
gst_bin_add (GST_BIN (bin->bin), bin->cap_filter);
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_transform%d”, uid);
if (config->type == NV_DS_SINK_RENDER_EGL) {
if(prop.integrated) {
bin->transform =
gst_element_factory_make (NVDS_ELEM_EGLTRANSFORM, elem_name);
} else {
bin->transform = gst_element_factory_make (NVDS_ELEM_VIDEO_CONV, elem_name);
}
if (!bin->transform) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
gst_bin_add (GST_BIN (bin->bin), bin->transform);

if(!prop.integrated) {
  caps = gst_caps_new_empty_simple ("video/x-raw");

  GstCapsFeatures *feature = NULL;
  feature = gst_caps_features_new (MEMORY_FEATURES, NULL);
  gst_caps_set_features (caps, 0, feature);
  g_object_set (G_OBJECT (bin->cap_filter), "caps", caps, NULL);

  g_object_set (G_OBJECT (bin->transform), "gpu-id", config->gpu_id, NULL);
  g_object_set (G_OBJECT (bin->transform), "nvbuf-memory-type",
      config->nvbuf_memory_type, NULL);
}

}

g_snprintf (elem_name, sizeof (elem_name), “render_queue%d”, uid);
bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, elem_name);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

gst_bin_add_many (GST_BIN (bin->bin), bin->queue, bin->sink, NULL);

connect_to = bin->sink;

if (bin->cap_filter) {
NVGSTDS_LINK_ELEMENT (bin->cap_filter, connect_to);
connect_to = bin->cap_filter;
}

if (bin->transform) {
NVGSTDS_LINK_ELEMENT (bin->transform, connect_to);
connect_to = bin->transform;
}

NVGSTDS_LINK_ELEMENT (bin->queue, connect_to);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

ret = TRUE;

done:
if (caps) {
gst_caps_unref (caps);
}
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

static void broker_queue_overrun (GstElement* sink_queue, gpointer user_data)
{
(void)sink_queue;
(void)user_data;
NVGSTDS_WARN_MSG_V("nvmsgbroker queue overrun; Older Message Buffer "
“Dropped; Network bandwidth might be insufficient\n”);
}

/**

Function to create sink bin to generate meta-msg, convert to json based on
a schema and send over msgbroker.
*/
static gboolean
create_msg_conv_broker_bin (NvDsSinkMsgConvBrokerConfig config,
NvDsSinkBinSubBin bin)
{
/ Create the subbin: → q → msgconv → msgbroker bin */
gboolean ret = FALSE;
gchar elem_name[50];

uid++;

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin%d”, uid);
bin->bin = gst_bin_new (elem_name);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_queue%d”, uid);
bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, elem_name);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

/** set threshold on queue to avoid pipeline choke when broker is stuck on network

leaky=2 (2): downstream - Leaky on downstream (old buffers) */
g_object_set(G_OBJECT(bin->queue), “leaky”, 2, NULL);
g_object_set(G_OBJECT(bin->queue), “max-size-buffers”, 20, NULL);
g_signal_connect (G_OBJECT (bin->queue), “overrun”,
G_CALLBACK (broker_queue_overrun), bin);

/* Create msg converter to generate payload from buffer metadata */
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_transform%d”, uid);
if (config->disable_msgconv)
{
bin->transform = gst_element_factory_make (“queue”, elem_name);
}
else
{
bin->transform = gst_element_factory_make (NVDS_ELEM_MSG_CONV, elem_name);
}
if (!bin->transform) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

if (!config->disable_msgconv)
g_object_set (G_OBJECT(bin->transform), “config”, config->config_file_path,
“msg2p-lib”, (config->conv_msg2p_lib ? config->conv_msg2p_lib : NULL),
“payload-type”, config->conv_payload_type,
“comp-id”, config->conv_comp_id,
“debug-payload-dir”, config->debug_payload_dir,
“multiple-payloads”, config->multiple_payloads,
“msg2p-newapi”, config->conv_msg2p_new_api,
“frame-interval”, config->conv_frame_interval,
NULL);

/* Create msg broker to send payload to server */
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_sink%d”, uid);
bin->sink = gst_element_factory_make (NVDS_ELEM_MSG_BROKER, elem_name);
if (!bin->sink) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
g_object_set (G_OBJECT(bin->sink), “proto-lib”, config->proto_lib,
“conn-str”, config->conn_str,
“topic”, config->topic,
“sync”, config->sync, “async”, FALSE,
“config”, config->broker_config_file_path,
“comp-id”, config->broker_comp_id,
“new-api”, config->new_api,
NULL);

gst_bin_add_many (GST_BIN (bin->bin),
bin->queue, bin->transform, bin->sink,
NULL);

NVGSTDS_LINK_ELEMENT (bin->queue, bin->transform);
NVGSTDS_LINK_ELEMENT (bin->transform, bin->sink);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

ret = TRUE;

done:
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

/**

Probe function to drop upstream “GST_QUERY_SEEKING” query from h264parse element.
This is a WAR to avoid memory leaks from h264parse element
*/
static GstPadProbeReturn
seek_query_drop_prob (GstPad * pad, GstPadProbeInfo * info,
gpointer u_data)
{
if (GST_PAD_PROBE_INFO_TYPE(info) &
GST_PAD_PROBE_TYPE_QUERY_UPSTREAM)
{
GstQuery *query = GST_PAD_PROBE_INFO_QUERY(info);
if(GST_QUERY_TYPE (query) == GST_QUERY_SEEKING)
{
return GST_PAD_PROBE_DROP;
}
}
return GST_PAD_PROBE_OK;
}

/**

Function to create sink bin to generate encoded output.
*/
static gboolean
create_encode_file_bin (NvDsSinkEncoderConfig * config, NvDsSinkBinSubBin * bin)
{
GstCaps *caps = NULL;
gboolean ret = FALSE;
gchar elem_name[50];
int probe_id = 0;
gulong bitrate = config->bitrate;
guint profile = config->profile;

uid++;

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin%d”, uid);
bin->bin = gst_bin_new (elem_name);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_queue%d”, uid);
bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, elem_name);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_transform%d”, uid);
bin->transform = gst_element_factory_make (NVDS_ELEM_VIDEO_CONV, elem_name);
if (!bin->transform) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_cap_filter%d”, uid);
bin->cap_filter = gst_element_factory_make (NVDS_ELEM_CAPS_FILTER, elem_name);
if (!bin->cap_filter) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
if (config->codec == NV_DS_ENCODER_MPEG4 || config->enc_type == NV_DS_ENCODER_TYPE_SW)
caps = gst_caps_from_string (“video/x-raw, format=I420”);
else
caps = gst_caps_from_string (“video/x-raw(memory:NVMM), format=I420”);
g_object_set (G_OBJECT (bin->cap_filter), “caps”, caps, NULL);

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_encoder%d”, uid);
switch (config->codec) {
case NV_DS_ENCODER_H264:
if (config->enc_type == NV_DS_ENCODER_TYPE_SW)
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H264_SW, elem_name);
else
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H264_HW, elem_name);
break;
case NV_DS_ENCODER_H265:
if (config->enc_type == NV_DS_ENCODER_TYPE_SW)
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H265_SW, elem_name);
else
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H265_HW, elem_name);
break;
case NV_DS_ENCODER_MPEG4:
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_MPEG4, elem_name);
break;
default:
goto done;
}
if (!bin->encoder) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

NVGSTDS_ELEM_ADD_PROBE(probe_id,
bin->encoder, “sink”,
seek_query_drop_prob, GST_PAD_PROBE_TYPE_QUERY_UPSTREAM, bin);

probe_id = probe_id;

if (config->codec == NV_DS_ENCODER_MPEG4)
config->enc_type = NV_DS_ENCODER_TYPE_SW;

struct cudaDeviceProp prop;
cudaGetDeviceProperties(&prop, config->gpu_id);

if(prop.integrated && config->enc_type == NV_DS_ENCODER_TYPE_HW) {
g_object_set (G_OBJECT (bin->encoder), “bufapi-version”, 1, NULL);
}

if (config->enc_type == NV_DS_ENCODER_TYPE_HW) {
g_object_set (G_OBJECT (bin->encoder), “profile”, profile, NULL);
g_object_set (G_OBJECT (bin->encoder), “iframeinterval”, config->iframeinterval, NULL);
g_object_set (G_OBJECT (bin->encoder), “bitrate”, bitrate, NULL);
} else {
if (config->codec == NV_DS_ENCODER_MPEG4)
g_object_set (G_OBJECT (bin->encoder), “bitrate”, bitrate, NULL);
else {
//bitrate is in kbits/sec for software encoder x264enc and x265enc
g_object_set (G_OBJECT (bin->encoder), “bitrate”, bitrate / 1000, NULL);
}
}

switch (config->codec) {
case NV_DS_ENCODER_H264:
bin->codecparse = gst_element_factory_make (“h264parse”, “h264-parser”);
break;
case NV_DS_ENCODER_H265:
bin->codecparse = gst_element_factory_make (“h265parse”, “h265-parser”);
break;
case NV_DS_ENCODER_MPEG4:
bin->codecparse = gst_element_factory_make (“mpeg4videoparse”, “mpeg4-parser”);
break;
default:
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_mux%d”, uid);
switch (config->container) {
case NV_DS_CONTAINER_MP4:
bin->mux = gst_element_factory_make (NVDS_ELEM_MUX_MP4, elem_name);
break;
case NV_DS_CONTAINER_MKV:
bin->mux = gst_element_factory_make (NVDS_ELEM_MKV, elem_name);
break;
default:
goto done;
}
if (!bin->mux) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
bin->splitmuxsink = gst_element_factory_make(“splitmuxsink”, “splitmuxsink”);
if (!bin->splitmuxsink) {
NVGSTDS_ERR_MSG_V(“Failed to create ‘splitmuxsink’”);
goto done;
g_object_set(G_OBJECT(bin->splitmuxsink), “location”, config->output_file_path,
“muxer”, bin->mux, NULL);
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_sink%d”, uid);
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_FILE, elem_name);
if (!bin->sink) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_object_set (G_OBJECT (bin->sink), “location”, config->output_file_path,
“sync”, config->sync, “async”, FALSE, NULL);
g_object_set (G_OBJECT (bin->transform), “gpu-id”, config->gpu_id, NULL);
gst_bin_add_many (GST_BIN (bin->bin), bin->queue,
bin->transform, bin->codecparse, bin->cap_filter,
bin->encoder, bin->mux, bin->sink, NULL);

NVGSTDS_LINK_ELEMENT (bin->queue, bin->transform);

NVGSTDS_LINK_ELEMENT (bin->transform, bin->cap_filter);
NVGSTDS_LINK_ELEMENT (bin->cap_filter, bin->encoder);

NVGSTDS_LINK_ELEMENT (bin->encoder, bin->codecparse);
NVGSTDS_LINK_ELEMENT (bin->codecparse, bin->mux);
NVGSTDS_LINK_ELEMENT (bin->mux, bin->sink);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

ret = TRUE;

done:
if (caps) {
gst_caps_unref (caps);
}
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

static gboolean
start_rtsp_streaming (guint rtsp_port_num, guint updsink_port_num,
NvDsEncoderType enctype, guint64 udp_buffer_size)
{
GstRTSPMountPoints *mounts;
GstRTSPMediaFactory *factory;
char udpsrc_pipeline[512];

char port_num_Str[64] = { 0 };
char *encoder_name;

if (enctype == NV_DS_ENCODER_H264) {
encoder_name = “H264”;
} else if (enctype == NV_DS_ENCODER_H265) {
encoder_name = “H265”;
} else {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
return FALSE;
}

if (udp_buffer_size == 0)
udp_buffer_size = 512 * 1024;

sprintf (udpsrc_pipeline,
"( udpsrc name=pay0 port=%d buffer-size=%lu caps="application/x-rtp, media=video, "
“clock-rate=90000, encoding-name=%s, payload=96 " )”,
updsink_port_num, udp_buffer_size, encoder_name);

sprintf (port_num_Str, “%d”, rtsp_port_num);

g_mutex_lock (&server_cnt_lock);

server [server_count] = gst_rtsp_server_new ();
g_object_set (server [server_count], “service”, port_num_Str, NULL);

mounts = gst_rtsp_server_get_mount_points (server [server_count]);

factory = gst_rtsp_media_factory_new ();
gst_rtsp_media_factory_set_launch (factory, udpsrc_pipeline);

gst_rtsp_mount_points_add_factory (mounts, “/ds-test”, factory);

g_object_unref (mounts);

gst_rtsp_server_attach (server [server_count], NULL);

server_count++;

g_mutex_unlock (&server_cnt_lock);

g_print
(“\n *** DeepStream: Launched RTSP Streaming at rtsp://localhost:%d/ds-test ***\n\n”,
rtsp_port_num);

return TRUE;
}

static gboolean
create_udpsink_bin (NvDsSinkEncoderConfig * config, NvDsSinkBinSubBin * bin)
{
GstCaps *caps = NULL;
gboolean ret = FALSE;
gchar elem_name[50];
gchar encode_name[50];
gchar rtppay_name[50];
int probe_id = 0;

//guint rtsp_port_num = g_rtsp_port_num++;
uid++;

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin%d”, uid);
bin->bin = gst_bin_new (elem_name);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_queue%d”, uid);
bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, elem_name);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_transform%d”, uid);
bin->transform = gst_element_factory_make (NVDS_ELEM_VIDEO_CONV, elem_name);
if (!bin->transform) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_cap_filter%d”, uid);
bin->cap_filter = gst_element_factory_make (NVDS_ELEM_CAPS_FILTER, elem_name);
if (!bin->cap_filter) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

if (config->enc_type == NV_DS_ENCODER_TYPE_SW)
caps = gst_caps_from_string (“video/x-raw, format=I420”);
else
caps = gst_caps_from_string (“video/x-raw(memory:NVMM), format=I420”);

g_object_set (G_OBJECT (bin->cap_filter), “caps”, caps, NULL);

g_snprintf (encode_name, sizeof (encode_name), “sink_sub_bin_encoder%d”, uid);
g_snprintf (rtppay_name, sizeof (rtppay_name), “sink_sub_bin_rtppay%d”, uid);

switch (config->codec) {
case NV_DS_ENCODER_H264:
bin->codecparse = gst_element_factory_make (“h264parse”, “h264-parser”);
bin->rtppay = gst_element_factory_make (“rtph264pay”, rtppay_name);
if (config->enc_type == NV_DS_ENCODER_TYPE_SW)
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H264_SW, encode_name);
else
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H264_HW, encode_name);
break;
case NV_DS_ENCODER_H265:
bin->codecparse = gst_element_factory_make (“h265parse”, “h265-parser”);
g_object_set (G_OBJECT (bin->codecparse), “config-interval”, -1, NULL);
bin->rtppay = gst_element_factory_make (“rtph265pay”, rtppay_name);
if (config->enc_type == NV_DS_ENCODER_TYPE_SW)
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H265_SW, encode_name);
else
bin->encoder = gst_element_factory_make (NVDS_ELEM_ENC_H265_HW, encode_name);
break;
default:
goto done;
}

if (!bin->encoder) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, encode_name);
goto done;
}

NVGSTDS_ELEM_ADD_PROBE(probe_id,
bin->encoder, “sink”,
seek_query_drop_prob, GST_PAD_PROBE_TYPE_QUERY_UPSTREAM, bin);

probe_id = probe_id;

if (!bin->rtppay) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, rtppay_name);
goto done;
}

if (config->enc_type == NV_DS_ENCODER_TYPE_SW) {
//bitrate is in kbits/sec for software encoder x264enc and x265enc
g_object_set (G_OBJECT (bin->encoder), “bitrate”, config->bitrate/1000, NULL);
} else {
g_object_set (G_OBJECT (bin->encoder), “bitrate”, config->bitrate, NULL);
g_object_set (G_OBJECT (bin->encoder), “profile”, config->profile, NULL);
g_object_set (G_OBJECT (bin->encoder), “iframeinterval”, config->iframeinterval, NULL);
}

struct cudaDeviceProp prop;
cudaGetDeviceProperties(&prop, config->gpu_id);

if(prop.integrated) {
if (config->enc_type == NV_DS_ENCODER_TYPE_HW) {
g_object_set (G_OBJECT (bin->encoder), “preset-level”, 1, NULL);
g_object_set (G_OBJECT (bin->encoder), “insert-sps-pps”, 1, NULL);
g_object_set (G_OBJECT (bin->encoder), “bufapi-version”, 1, NULL);
}
} else {
g_object_set (G_OBJECT (bin->transform), “gpu-id”, config->gpu_id, NULL);
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_udpsink%d”, uid);
bin->sink = gst_element_factory_make (“udpsink”, elem_name);
if (!bin->sink) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_object_set (G_OBJECT (bin->sink), “host”, “224.224.255.255”, “port”,
config->udp_port, “async”, FALSE, “sync”, 0, NULL);

gst_bin_add_many (GST_BIN (bin->bin),
bin->queue, bin->cap_filter, bin->transform,
bin->encoder, bin->codecparse, bin->rtppay, bin->sink, NULL);

NVGSTDS_LINK_ELEMENT (bin->queue, bin->transform);
NVGSTDS_LINK_ELEMENT (bin->transform, bin->cap_filter);
NVGSTDS_LINK_ELEMENT (bin->cap_filter, bin->encoder);
NVGSTDS_LINK_ELEMENT (bin->encoder, bin->codecparse);
NVGSTDS_LINK_ELEMENT (bin->codecparse, bin->rtppay);
NVGSTDS_LINK_ELEMENT (bin->rtppay, bin->sink);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

ret = TRUE;

ret = start_rtsp_streaming (config->rtsp_port, config->udp_port, config->codec,
config->udp_buffer_size);
if (ret != TRUE) {
g_print (“%s: start_rtsp_straming function failed\n”, func);
}

done:
if (caps) {
gst_caps_unref (caps);
}
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

gboolean
create_sink_bin (guint num_sub_bins, NvDsSinkSubBinConfig * config_array,
NvDsSinkBin * bin, guint index)
{
gboolean ret = FALSE;
guint i;

bin->bin = gst_bin_new (“sink_bin”);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create element ‘sink_bin’”);
goto done;
}

bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, “sink_bin_queue”);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create element ‘sink_bin_queue’”);
goto done;
}

gst_bin_add (GST_BIN (bin->bin), bin->queue);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

bin->tee = gst_element_factory_make (NVDS_ELEM_TEE, “sink_bin_tee”);
if (!bin->tee) {
NVGSTDS_ERR_MSG_V (“Failed to create element ‘sink_bin_tee’”);
goto done;
}

gst_bin_add (GST_BIN (bin->bin), bin->tee);

NVGSTDS_LINK_ELEMENT (bin->queue, bin->tee);

g_object_set (G_OBJECT (bin->tee), “allow-not-linked”, TRUE, NULL);

for (i = 0; i < num_sub_bins; i++) {
if (!config_array[i].enable) {
continue;
}
if (config_array[i].source_id != index) {
continue;
}
if (config_array[i].link_to_demux) {
continue;
}
switch (config_array[i].type) {
case NV_DS_SINK_RENDER_EGL:
case NV_DS_SINK_RENDER_OVERLAY:
case NV_DS_SINK_FAKE:
config_array[i].render_config.type = config_array[i].type;
config_array[i].render_config.sync = config_array[i].sync;
if (!create_render_bin (&config_array[i].render_config,
&bin->sub_bins[i]))
goto done;
break;
case NV_DS_SINK_ENCODE_FILE:
config_array[i].encoder_config.sync = config_array[i].sync;
if (!create_encode_file_bin (&config_array[i].encoder_config,
&bin->sub_bins[i]))
goto done;
break;
case NV_DS_SINK_UDPSINK:
if (!create_udpsink_bin (&config_array[i].encoder_config,
&bin->sub_bins[i]))
goto done;
break;
case NV_DS_SINK_MSG_CONV_BROKER:
config_array[i].msg_conv_broker_config.sync = config_array[i].sync;
if (!create_msg_conv_broker_bin (&config_array[i].msg_conv_broker_config,
&bin->sub_bins[i]))
goto done;
break;
default:
goto done;
}

if(config_array[i].type != NV_DS_SINK_MSG_CONV_BROKER) {
  gst_bin_add (GST_BIN (bin->bin), bin->sub_bins[i].bin);
  if (!link_element_to_tee_src_pad (bin->tee, bin->sub_bins[i].bin)) {
    goto done;
  }
}
bin->num_bins++;

}

if (bin->num_bins == 0) {
NvDsSinkRenderConfig config;
config.type = NV_DS_SINK_FAKE;
if (!create_render_bin (&config, &bin->sub_bins[0]))
goto done;
gst_bin_add (GST_BIN (bin->bin), bin->sub_bins[0].bin);
if (!link_element_to_tee_src_pad (bin->tee, bin->sub_bins[0].bin)) {
goto done;
}
bin->num_bins = 1;
}

ret = TRUE;
done:
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

gboolean
create_demux_sink_bin (guint num_sub_bins, NvDsSinkSubBinConfig * config_array,
NvDsSinkBin * bin, guint index)
{
gboolean ret = FALSE;
guint i;

bin->bin = gst_bin_new (“sink_bin”);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create element ‘sink_bin’”);
goto done;
}

bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, “sink_bin_queue”);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create element ‘sink_bin_queue’”);
goto done;
}

gst_bin_add (GST_BIN (bin->bin), bin->queue);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

bin->tee = gst_element_factory_make (NVDS_ELEM_TEE, “sink_bin_tee”);
if (!bin->tee) {
NVGSTDS_ERR_MSG_V (“Failed to create element ‘sink_bin_tee’”);
goto done;
}

gst_bin_add (GST_BIN (bin->bin), bin->tee);

NVGSTDS_LINK_ELEMENT (bin->queue, bin->tee);

for (i = 0; i < num_sub_bins; i++) {
if (!config_array[i].enable) {
continue;
}
if (!config_array[i].link_to_demux) {
continue;
}
switch (config_array[i].type) {
case NV_DS_SINK_RENDER_EGL:
case NV_DS_SINK_RENDER_OVERLAY:
case NV_DS_SINK_FAKE:
config_array[i].render_config.type = config_array[i].type;
config_array[i].render_config.sync = config_array[i].sync;
if (!create_render_bin (&config_array[i].render_config,
&bin->sub_bins[i]))
goto done;
break;
case NV_DS_SINK_ENCODE_FILE:
config_array[i].encoder_config.sync = config_array[i].sync;
if (!create_encode_file_bin (&config_array[i].encoder_config,
&bin->sub_bins[i]))
goto done;
break;
case NV_DS_SINK_UDPSINK:
if (!create_udpsink_bin (&config_array[i].encoder_config,
&bin->sub_bins[i]))
goto done;
break;
case NV_DS_SINK_MSG_CONV_BROKER:
config_array[i].msg_conv_broker_config.sync = config_array[i].sync;
if (!create_msg_conv_broker_bin (&config_array[i].msg_conv_broker_config,
&bin->sub_bins[i]))
goto done;
break;
default:
goto done;
}

if(config_array[i].type != NV_DS_SINK_MSG_CONV_BROKER) {
  gst_bin_add (GST_BIN (bin->bin), bin->sub_bins[i].bin);
  if (!link_element_to_tee_src_pad (bin->tee, bin->sub_bins[i].bin)) {
    goto done;
  }
}
bin->num_bins++;

}

if (bin->num_bins == 0) {
NvDsSinkRenderConfig config;
config.type = NV_DS_SINK_FAKE;
if (!create_render_bin (&config, &bin->sub_bins[0]))
goto done;
gst_bin_add (GST_BIN (bin->bin), bin->sub_bins[0].bin);
if (!link_element_to_tee_src_pad (bin->tee, bin->sub_bins[0].bin)) {
goto done;
}
bin->num_bins = 1;
}

ret = TRUE;
done:
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

static GstRTSPFilterResult
client_filter (GstRTSPServer * server, GstRTSPClient * client,
gpointer user_data)
{
return GST_RTSP_FILTER_REMOVE;
}

void
destroy_sink_bin ()
{
GstRTSPMountPoints *mounts;
GstRTSPSessionPool pool;
guint i = 0;
for (i = 0; i < server_count; i++) {
mounts = gst_rtsp_server_get_mount_points (server [i]);
gst_rtsp_mount_points_remove_factory (mounts, “/ds-test”);
g_object_unref (mounts);
gst_rtsp_server_client_filter (server [i], client_filter, NULL);
pool = gst_rtsp_server_get_session_pool (server [i]);
gst_rtsp_session_pool_cleanup (pool);
g_object_unref (pool);
}
}
/

Copyright (c) 2018-2021, NVIDIA CORPORATION. All rights reserved.
Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the “Software”),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
*/

#include <sys/types.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/wait.h>

#include “deepstream_common.h”
#include “deepstream_sinks.h”
#include <gst/rtsp-server/rtsp-server.h>
#include <cuda_runtime_api.h>

static guint uid = 0;
static GstRTSPServer *server [MAX_SINK_BINS];
static guint server_count = 0;
static GMutex server_cnt_lock;

GST_DEBUG_CATEGORY_EXTERN (NVDS_APP);

/**

Function to create sink bin for Display / Fakesink.
*/
static gboolean
create_render_bin (NvDsSinkRenderConfig * config, NvDsSinkBinSubBin * bin)
{
gboolean ret = FALSE;
gchar elem_name[50];
GstElement *connect_to;
GstCaps *caps = NULL;

uid++;

struct cudaDeviceProp prop;
cudaGetDeviceProperties(&prop, config->gpu_id);

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin%d”, uid);
bin->bin = gst_bin_new (elem_name);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_sink%d”, uid);
switch (config->type) {
case NV_DS_SINK_RENDER_EGL:
GST_CAT_INFO (NVDS_APP, “NVvideo renderer\n”);
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_EGL, elem_name);
g_object_set (G_OBJECT (bin->sink), “window-x”, config->offset_x,
“window-y”, config->offset_y, “window-width”, config->width,
“window-height”, config->height, NULL);
g_object_set (G_OBJECT (bin->sink), “enable-last-sample”, FALSE, NULL);
break;
case NV_DS_SINK_RENDER_OVERLAY:
#ifndef IS_TEGRA
NVGSTDS_ERR_MSG_V (“Overlay is only supported for Jetson”);
return FALSE;
#endif
GST_CAT_INFO (NVDS_APP, “NVvideo renderer\n”);
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_OVERLAY, elem_name);
g_object_set (G_OBJECT (bin->sink), “display-id”, config->display_id,
NULL);
g_object_set (G_OBJECT (bin->sink), “overlay”, config->overlay_id, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-x”, config->offset_x, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-y”, config->offset_y, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-w”, config->width, NULL);
g_object_set (G_OBJECT (bin->sink), “overlay-h”, config->height, NULL);
break;
case NV_DS_SINK_FAKE:
bin->sink = gst_element_factory_make (NVDS_ELEM_SINK_FAKESINK, elem_name);
g_object_set (G_OBJECT (bin->sink), “enable-last-sample”, FALSE, NULL);
break;
default:
return FALSE;
}

if (!bin->sink) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

g_object_set (G_OBJECT (bin->sink), “sync”, config->sync, “max-lateness”, -1,
“async”, FALSE, “qos”, config->qos, NULL);

if(!prop.integrated) {
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_cap_filter%d”, uid);
bin->cap_filter = gst_element_factory_make (NVDS_ELEM_CAPS_FILTER, elem_name);
if (!bin->cap_filter) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
gst_bin_add (GST_BIN (bin->bin), bin->cap_filter);
}

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_transform%d”, uid);
if (config->type == NV_DS_SINK_RENDER_EGL) {
if(prop.integrated) {
bin->transform =
gst_element_factory_make (NVDS_ELEM_EGLTRANSFORM, elem_name);
} else {
bin->transform = gst_element_factory_make (NVDS_ELEM_VIDEO_CONV, elem_name);
}
if (!bin->transform) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
gst_bin_add (GST_BIN (bin->bin), bin->transform);

if(!prop.integrated) {
  caps = gst_caps_new_empty_simple ("video/x-raw");

  GstCapsFeatures *feature = NULL;
  feature = gst_caps_features_new (MEMORY_FEATURES, NULL);
  gst_caps_set_features (caps, 0, feature);
  g_object_set (G_OBJECT (bin->cap_filter), "caps", caps, NULL);

  g_object_set (G_OBJECT (bin->transform), "gpu-id", config->gpu_id, NULL);
  g_object_set (G_OBJECT (bin->transform), "nvbuf-memory-type",
      config->nvbuf_memory_type, NULL);
}

}

g_snprintf (elem_name, sizeof (elem_name), “render_queue%d”, uid);
bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, elem_name);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

gst_bin_add_many (GST_BIN (bin->bin), bin->queue, bin->sink, NULL);

connect_to = bin->sink;

if (bin->cap_filter) {
NVGSTDS_LINK_ELEMENT (bin->cap_filter, connect_to);
connect_to = bin->cap_filter;
}

if (bin->transform) {
NVGSTDS_LINK_ELEMENT (bin->transform, connect_to);
connect_to = bin->transform;
}

NVGSTDS_LINK_ELEMENT (bin->queue, connect_to);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

ret = TRUE;

done:
if (caps) {
gst_caps_unref (caps);
}
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

static void broker_queue_overrun (GstElement* sink_queue, gpointer user_data)
{
(void)sink_queue;
(void)user_data;
NVGSTDS_WARN_MSG_V("nvmsgbroker queue overrun; Older Message Buffer "
“Dropped; Network bandwidth might be insufficient\n”);
}

/**

Function to create sink bin to generate meta-msg, convert to json based on
a schema and send over msgbroker.
*/
static gboolean
create_msg_conv_broker_bin (NvDsSinkMsgConvBrokerConfig config,
NvDsSinkBinSubBin bin)
{
/ Create the subbin: → q → msgconv → msgbroker bin */
gboolean ret = FALSE;
gchar elem_name[50];

uid++;

g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin%d”, uid);
bin->bin = gst_bin_new (elem_name);
if (!bin->bin) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_queue%d”, uid);
bin->queue = gst_element_factory_make (NVDS_ELEM_QUEUE, elem_name);
if (!bin->queue) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

/** set threshold on queue to avoid pipeline choke when broker is stuck on network

leaky=2 (2): downstream - Leaky on downstream (old buffers) */
g_object_set(G_OBJECT(bin->queue), “leaky”, 2, NULL);
g_object_set(G_OBJECT(bin->queue), “max-size-buffers”, 20, NULL);
g_signal_connect (G_OBJECT (bin->queue), “overrun”,
G_CALLBACK (broker_queue_overrun), bin);

/* Create msg converter to generate payload from buffer metadata */
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_transform%d”, uid);
if (config->disable_msgconv)
{
bin->transform = gst_element_factory_make (“queue”, elem_name);
}
else
{
bin->transform = gst_element_factory_make (NVDS_ELEM_MSG_CONV, elem_name);
}
if (!bin->transform) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}

if (!config->disable_msgconv)
g_object_set (G_OBJECT(bin->transform), “config”, config->config_file_path,
“msg2p-lib”, (config->conv_msg2p_lib ? config->conv_msg2p_lib : NULL),
“payload-type”, config->conv_payload_type,
“comp-id”, config->conv_comp_id,
“debug-payload-dir”, config->debug_payload_dir,
“multiple-payloads”, config->multiple_payloads,
“msg2p-newapi”, config->conv_msg2p_new_api,
“frame-interval”, config->conv_frame_interval,
NULL);

/* Create msg broker to send payload to server */
g_snprintf (elem_name, sizeof (elem_name), “sink_sub_bin_sink%d”, uid);
bin->sink = gst_element_factory_make (NVDS_ELEM_MSG_BROKER, elem_name);
if (!bin->sink) {
NVGSTDS_ERR_MSG_V (“Failed to create ‘%s’”, elem_name);
goto done;
}
g_object_set (G_OBJECT(bin->sink), “proto-lib”, config->proto_lib,
“conn-str”, config->conn_str,
“topic”, config->topic,
“sync”, config->sync, “async”, FALSE,
“config”, config->broker_config_file_path,
“comp-id”, config->broker_comp_id,
“new-api”, config->new_api,
NULL);

gst_bin_add_many (GST_BIN (bin->bin),
bin->queue, bin->transform, bin->sink,
NULL);

NVGSTDS_LINK_ELEMENT (bin->queue, bin->transform);
NVGSTDS_LINK_ELEMENT (bin->transform, bin->sink);

NVGSTDS_BIN_ADD_GHOST_PAD (bin->bin, bin->queue, “sink”);

ret = TRUE;

done:
if (!ret) {
NVGSTDS_ERR_MSG_V (“%s failed”, func);
}
return ret;
}

/**

Probe function to drop upstream “GST_QUERY_SEEKING” query from h264parse element.
This is a WAR to avoid memory leaks from h264parse element
*/
static GstPadProbeReturn
seek_query_drop_prob (GstPad * pad, GstPadProbeInfo * info,
gpointer u_data)
{
if (GST_PAD_PROBE_INFO_TYPE(info) &
GST_PAD_PROBE_TYPE_QUERY_UPSTREAM)
{
GstQuery *query = GST_PAD_PROBE_INFO_QUERY(info);
if(GST_QUERY_TYPE (query) == GST_QUERY_SEEKING)
{
return GST_PAD_PROBE_DROP;
}
}
return GST_PAD_PROBE_OK;
}

/**

Function to create sink bin to generate encoded output.
*/
static gboolean
create_encode_file_bin (NvDsSinkEncoderConfig * config, NvDsSinkBinSubBin * bin)
{
GstCaps *caps = NULL;
gboolean ret = FALSE;
gchar elem_name[50];
int probe_id = 0;
gulong bitrate = config->bitrate;
guint profile = config->profile;