sources/includes/nvds_mqtt_proto.h

Go to the documentation of this file.

/*
 * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#include "mqtt_protocol.h"
#include "mosquitto.h"
 
#include <mutex>
#include <unordered_map>
#include <errno.h>
#include <syslog.h>
#include "glib.h"
#include <string.h>
#include <string>
#include "nvds_msgapi.h"
#include "nvds_utils.h"
 
using namespace std;
 
#define MAX_FIELD_LEN 1024
 
// Structure to hold extracted data from protobuf payload
struct ExtractedData {
    std::string sensor_id;  // Extracted from metadata
    std::string timestamp;  // Not used - timestamp is always current system time
};
 
// Structure to hold protobuf wrapper message
struct ProtobufWrapper {
    std::string payload;        // Raw protobuf binary data (no encoding needed)
    std::string key;           // sensorId extracted from metadata
    std::string timestamp;     // current system timestamp when data was sent
};
#define DEFAULT_LOOP_TIMEOUT 2000
#define DEFAULT_KEEP_ALIVE 60
/* Message details:
 * send_callback = user callback func
 * user_ptr = user pointer passed by async send
 */
struct send_msg_info_t {
  nvds_msgapi_send_cb_t send_callback;
  void *user_ptr;
};
 
/* Details of mqtt connection handle:
 * mosq : mosquitto client object
 * sub_callback : user subscription callback func
 * connect_cb : user connection callback func
 * user_ctx : user pointer passed by sub
 * username: username for login to server
 * password: password for login to server
 * client_id: name of MQTT client
 * loop_timeout : time in ms for the call to loop to wait for network activity
 * keep_alive : number of seconds after which broker should send PING if no messages have been exchanged
 * subscription_on : Flag to check if subscription is ON
 * send_msg_info_map : map message info to id assigned by mosquitto broker
 * map_lock : mutex lock for accessing above map
 * enable_tls : flag to check if TLS encryption is enabled by the broker
 * cafile : path to a TLS certificate authority file
 * capath : path to a directory containing TLS CA files
 * certfile : path to the client TLS certificate file
 * keyfile : path to the client TLS key file
 * disconnect : bool for checking if disconnect has been called
 * set_threaded : bool for setting mosquitto_threaded_set in proto adaptor
 */
typedef struct {
   struct mosquitto *mosq = NULL;
   nvds_msgapi_subscribe_request_cb_t  sub_callback;
   nvds_msgapi_connect_cb_t connect_cb;
   void* user_ctx;
   char connection_str[MAX_FIELD_LEN] = {0};
   char username[MAX_FIELD_LEN] = {0};
   char password[MAX_FIELD_LEN] = {0};
   char client_id[MAX_FIELD_LEN] = {0};
   int loop_timeout = DEFAULT_LOOP_TIMEOUT;
   int keep_alive = DEFAULT_KEEP_ALIVE;
   bool subscription_on = false;
   std::unordered_map<int , send_msg_info_t> send_msg_info_map;
   std::mutex map_lock;
   bool enable_tls = false;
   char cafile[MAX_FIELD_LEN] = {0};
   char capath[MAX_FIELD_LEN] = {0};
   char certfile[MAX_FIELD_LEN] = {0};
   char keyfile[MAX_FIELD_LEN] = {0};
   bool disconnect = false;
   bool set_threaded = true;
} NvDsMqttClientHandle;
 
NvDsMsgApiErrorType nvds_mqtt_read_config(NvDsMqttClientHandle *mh, char *config_path);
void mosq_mqtt_log_callback(struct mosquitto *mosq, void *obj, int level, const char *str);
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc, const mosquitto_property *properties);
void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flags, const mosquitto_property *properties);
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid, int reason_code, const mosquitto_property *properties);
bool is_valid_mqtt_connection_str(char *connection_str, std::string &burl, std::string &bport);
 
// Function declarations for protobuf wrapper
ExtractedData extract_data_from_payload(const uint8_t *payload, size_t nbuf);
std::string create_protobuf_wrapper(const uint8_t *payload, size_t nbuf);
std::string create_enhanced_payload(const uint8_t *payload, size_t nbuf);
std::string serialize_protobuf_wrapper(const ProtobufWrapper& wrapper);