This plugin sends payload messages to the server using a specified communication protocol. It accepts any buffer that has NvDsPayload metadata attached, and uses the nvds_msgapi_* interface to send the messages to the server. You must implement the nvds_msgapi_* interface for the protocol to be used and specify the implementing library in the proto-lib property.

The following table summarizes the features of the Gst-nvmsgbroker plugin.

The following table describes the Gst properties of the Gst-nvmsgbroker plugin.

You can use the NVIDIA DeepStream messaging interface, nvds_msgapi, to implement a custom protocol message handler and integrate it with DeepStream applications. Such a message handler, known as a protocol adapter, enables you to integrate DeepStream applications with backend data sources, such as data stored in the cloud.

The Gst-nvmsgbroker plugin calls the functions in your protocol adapter as shown in the figure above. These functions support:

The nvds_msgapi interface is defined in the header file source/includes/nvds_msgapi.h. This header file defines a set of function pointers which provide an interface analogous to an interface in C++.

The following sections describe the methods defined by the nvds_msgapi interface.

The function accepts a connection string and configures a connection. The adapter implementation can choose whether or not the function actually makes a connection to accommodate connectionless protocols such as HTTP.

A handle for use in subsequent interface calls if successful, or NULL otherwise.

Both functions send data to the endpoint of a connection. They accept a message topic and a message payload.

The nvds_send() function is synchronous. The nvds_msgapi_send_async() function is asynchronous; it accepts a callback function that is called when the “send” operation is completed.

Both functions allow the API client to control execution of the adapter logic by calling nvds_msgapi_do_work(). See the description of the nvds_msgapi_do_work() function.

The protocol adapter must periodically surrender control to the client during processing of nvds_msgapi_send() and nvds_msgapi_send_async() calls. The client must periodically call nvsd_msgapi_do_work() to let the protocol adapter resume execution. This ensures that the protocol adapter receives sufficient CPU resources. The client can use this convention to control the protocol adapter’s use of multi-threading and thread scheduling. The protocol adapter can use it to support heartbeat functionality, if the underlying protocol requires that.

The nvds_msgapi_do_work() convention is needed when the protocol adapter executes in the client thread. Alternatively, the protocol adapter may execute time-consuming operations in its own thread. In this case the protocol adapter need not surrender control to the client, the client need not call nvsd_msgapi_do_work(), and the implementation of nvds_msgapi_do_work() may be a no-op.

The protocol adapter’s documentation must specify whether the client must call nvds_msgapi_do_work(), and if so, how often.

The function terminates the connection, if the underlying protocol requires it, and frees resources associated with h_ptr.

This function returns a string that identifies the nvds_msgapi version supported by this protocol adapter implementation. The string must use the format "<major>.<minor>", where <major> is a major version number and <minor> is a minor version number. A change in the major version number indicates an API change that may cause incompatibility. When the major version number changes, the minor version number is reset to 1.

The DeepStream 3.0 release includes a protocol adapter that supports Apache Kafka. The adapter provides out-of-the-box capability for DeepStream applications to publish messages to Kafka brokers.

The Kafka adapter uses librdkafka for the underlying protocol implementation. This library must be installed prior to use.

To install librdkakfa, enter these commands:

Install additional dependencies:

You can use the Kafka adapter in an application by setting the Gst-nvmsgbroker plugin’s proto-lib property to the pathname of the adapter’s shared library, libnvds_kafka_proto.so. The plugin’s conn-str property must be set to a string with format:

This instantiates the Gst-nvmsgbroker plugin and makes it use the Kafka protocol adapter to publish messages that the application sends to the broker at the specified broker address and topic.

You can define configuration setting for the Kafka protocol adapter as described by the documentation at:

You can set these options in the Gst-nvmsgbroker configuration file. Like the rest of DeepStream, the configuration file use the gkey format. The Kafka settings must be in a group named [message-broker], and must be specified as part of a key named proto-cfg. The settings can be a series of key-value pairs separated by semicolons, for example:

The Kafka adapter lets you specify the name of the field in messages that is to be used to define the partition key. For each message, the specified message field is extracted and send to the topic partitioner along with the message. The partitioner uses it to identify the partition in the Kafka cluster that handles the message. The partition key information must be specified in the Gst-nvmsgbroker configuration file’s [message-broker] group, using an entry named partition-key.

Fields embedded in a hierarchy of JSON objects in the message are specified using dotted notation. For example, for the sample JSON message shown below, the id field in the sensor object is identified as sensor.id,

You can integrate the Kafka adapter into custom user code by using the nvds_msgapi interface to call its functions. Note the following points with regard to the functions defined by the interface:

The Kafka adapter generates log messages based on the nvds_logger framework to help you monitor execution. The adapter generates separate logs for the INFO, DEBUG, and ERROR severity levels, as described in nvds_logger: The Logger Framework. You can limit the log messages generated by setting the level at which log messages are filtered as part of the logging setup script.

The DeepStream 4.0.1 release includes protocol adapters that supports direct messaging from device to cloud (using the Azure device client adapter) and through Azure IoT Edge runtime (using the Azure module client adapter). The adapters provide out-of-the-box capability for DeepStream applications to publish messages to Azure IoT Hub using the MQTT protocol.

The Azure IoT protocol adapters are encapsulated by their respective shared libraries found within the deepstream package at the location:

The Azure device client adapter library is named libnvds_azure_proto.so.

The Azure module client adapter library is named libnvds_azure_edge_proto.so.

Azure adapters use libiothub_client.so from the Azure IoT C SDK (v1.2.8) for the underlying protocol implementation. After you install the deepstream package you can find the precompiled library at:

You can also compile libiothub_client.so manually by entering these commands:

To install some other required dependencies, enter one of these commands.

Azure IoT adapter needs a functioning Azure IoT Hub instance to which is can publish messages. To set up an Azure IoT Hub instance if required, see the instructions at:

After you create the Azure IoT instance, create a device entry corresponding to the device that is running DeepStream.

To set up Azure IoT Edge runtime on the edge device, see the instructions at:

Place Azure IoT specific information in a custom configuration file named, e.g., cfg_azure.txt. The entries in the configuration file vary slightly between the Azure device client and the module client.

Here is useful information about some of the configuration file properties:

To use the Azure device client adapter in an application, set the Gst-nvmsgbroker plugin’s proto-lib property to the pathname of the adapter’s shared library - libnvds_azure_proto.so for the device client case, or libnvds_azure_edge_proto.so for the module client case.

The next step in using the adapter is to specify the connection details. The procedure for specifying connection details is different for the Azure device client and module client cases, as described in the following sections.

Set the plugin’s conn-str property to the full Azure connection string in the format:

Alternatively, you can specify the connection string details in the Azure configuration file:

Leave the connection string empty, since the Azure IoT Edge library automatically fetches the connection string from from the file /etc/iotedge/config.yaml.

Once the connection details have been configured, you can integrate the Azure device client and module client adapters into custom user code by using the nvds_msgapi interface to call its functions. Note the following points with regard to the functions defined by the interface:

The Azure device client and module client use different logging mechanisms.

The Azure device client adapter uses the nvds_logger framework to generate log messages which can help you monitor execution. The adapter generates separate logs for the INFO, DEBUG, and ERROR severity levels, as described in nvds_logger: Logging Framework. You can limit the generated log messages by setting the level at which log messages are filtered in the logging setup script.

The log messages from the Azure module client adapter library are emitted to stdout, and the log output is captured in the docker/iotedge module logs.

You can specify a message topic in a GStreamer property topic. However, the Azure device client and module client use the topic property in different ways.

The Azure device client does not support topics. Thus the value of the topic property is ignored, and you cannot use it to filter messages on Azure IoT Hub.

The Azure module client uses the topic property to determine the route of messages, i.e. how messages are passed within a system. For more information about message routes, see:

DeepStream release 4.0.1 includes an AMQP protocol adapter that DeepStream applications can use out of the box to publish messages using AMQP 0-9-1 message protocol.

The AMQP protocol adapter shared library is located in the deepstream package at:

AMQP protocol adapter for DeepStream uses the librabbitmq.so library, built from rabbitmq-c (v0.8.0) for the underlying AMQP protocol implementation. To build the library, enter these commands:

To copy the built librabbitmq.so library to its final location, enter this command.

Install additional dependencies:

The AMQP protocol communicates with an AMQP 0-9-1 compliant message broker. If you do not have a functioning broker already, you can deploy one by installing the rabbitmq-server package, available at:

You can install this package on your local system or on the remote machine where you want the broker to be installed.

To install the package, enter the command:

To determine whether the rabbitmq service is running, enter the command:

If rabbitmq is not running, enter this command to start it:

You can place AMQP protocol adapter specific information in a custom configuration named, for example, cfg_amqp.txt. Here is an example of configuration file entries for an AMQP broker installed on the local machine:

The properties in the configuration file are:

To use the AMQP protocol client adapter in a DeepStream application, set the Gst-nvmsgbroker plugin’s proto-lib property to the pathname of the adapter’s shared library, libnvds_amqp_proto.so.

You can specify the AMQP connection details in the AMQP adapter specific configuration file (e.g., cfg_amqp.txt) as described above. This is the recommended method. The path to the AMQP configuration file is specified by the Gst property config:

Alternatively, you can specify the AMQP protocol’s hostname, port number, and username in the Gst plugin’s conn-str property, and specify the password in the configuration file. In the Gst properties:

In the AMPQ configuration file:

You can set the Gst-nvmsgbroker plugin’s topic property to specify the message topic.

Alternatively, you can specify a topic in the AMQP configuration file (cfg_amqp.txt). In the Gst properties, set:

In the AMQP configuration file:

Once you have configured the connection, you can integrate the AMQP protocol adapter into your application by using the nvds_msgapi interface to call its functions. Note the following points about the functions defined by the interface:

The AMQP protocol adapter expects the client to manage usage and retirement of the connection handle. The client must ensure that once a handle is disconnected, it is not used for either a “send” call or a call to nvds_msgapi_do_work(). While the library attempts to ensure graceful failure, if the application calls these functions with retired handles, it does not do so in a thread-safe manner.

The AMQP protocol adapter uses the nvds_logger framework to generate log messages which can help you monitor execution. The adapter generates separate logs for the INFO, DEBUG, and ERROR severity levels, as described in nvds_logger: Logging Framework. You can limit the log messages being generated by setting the level at which log messages are filtered in the logging setup script.

DeepStream provides a logging framework named nvds_logger. The Kafka protocol adapter uses this framework to generate a run time log. nvds_logger is based on syslog, and offers many related features, including:

To enable logging, run the setup_nvds_logger.sh script. Note that this script must be run with sudo. You may have to modify the permissions associated with this script to make it executable.

The script accepts an optional parameter specifying the pathname of log file to be written. By default, the pathname is /tmp/nvds/ds.log.

Once logging is enabled, you can access the generated log messages by reading the log file.

By default, you must have sudo permissions to read the log file. Standard techniques for syslog-based logging configuration can eliminate this requirement.

nvds_logger allows logs to be associate with a severity level similar to that which syslog offers. You can filter log messages based on severity level by modifying the setup script. By default, the script enables logging for messages at the INFO level (level 6) and above. You can modify this as outlined in the comments in the script:

It is recommended that you limit the size of log files by retiring them periodically. logrotate is a popular utility for this purpose. You can use it in cron jobs so that the log files are automatically archived periodically, and are discarded after a desired interval.

You can implement modules that use the logger by including sources/includes/nvds_logger.h in the source code and linking to the libnvds_logger.so library.

Generating logs programmatically involves three steps:

Note the nvds_logger is a process-based logging mechanism, so the recommended procedure is to call nvds_log_open() from the main application routine rather than the individual plugins. Similarly, call nvds_log_close() from the main application when it shuts down the application before exit.