Amqp in deepstream amqp_protocol_adaptor

Accelerated Computing Intelligent Video Analytics DeepStream SDK
,
1

deepstream-6.1-jetson

I found the test_amqp_proto_async.c file under amqp_protocol_adaptor, which is an asynchronous message queue. Now I have some problems, that is, when my MQ service stops and restarts, the asynchronous message queue does not push anymore. I want to create a reconnection mechanism now, how can I do it?

3

Please provide complete information as applicable to your setup.

• Hardware Platform (Jetson / GPU)

• DeepStream Version

• JetPack Version (valid for Jetson only)

• TensorRT Version

• NVIDIA GPU Driver Version (valid for GPU only)

• Issue Type( questions, new requirements, bugs)

• 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)

• Requirement details( This is for new requirement. Including the module name-for which plugin or for which sample application, the function description)

4

@fanzh

Hardware Platform：Jetson
DeepStream Version：deepstream-6.2

When I want to test the
/opt/Nvidia/devstream/devstream-6.2/sources/libs/amqp_protocol_adaptor/test_amqp_proto_sync.c file，I can send messages to RabbitMQ normally and see message push on the RabbitMQ management page. I now want to implement: when I stop the RabbitMQ service, the code can continuously reconnect, and when my RabbitMQ service is normal, the code can reconnect successfully and start pushing.

5

@fanzh
I have modified the logic of connect_cb as follows:

void connect_cb(NvDsMsgApiHandle h_ptr, NvDsMsgApiEventType evt) {
    if (evt == NVDS_MSGAPI_EVT_SUCCESS) {
        const char *topics[] = {"alarm"};
        if (nvds_msgapi_subscribe_ptr(conn_handle, (char **)topics, 1, subscribe_cb, &consumed_cnt) != NVDS_MSGAPI_OK) {
			printf("sucess")
        }
    } else {
        h_ptr = nvds_msgapi_connect_ptr(NULL, connect_cb, (char *)AMQP_CFG_FILE); 
    }
}

But it doesn’t work.

6

amqp_protocol_adaptor low-level library implement is opensource in DS7.1. please modify the code for reconnection.

8

May I know your company name? and does this issue still block you?

9

the approach you shared is not quite correct, proper implementation will be more involved as demonstrated in nvmsgbroker source code. The code is trying to reconnect by modifying h_ptr, but this is not going to be reflected in the app as the callback provides a copy and not a reference. The call to nvds_msgapi_connect_ptr here is also improper as it needs a connection string as so:
NvDsMsgApiHandle ah = nvds_msgapi_connect_ptr((char *) AMQP_CONNECT_STR, connect_cb, (char *) AMQP_CFG_FILE);
you need to keep track of this NvDsMsgApiHandle in the app and create a new one with which to call nvds_msgapi operations with when the connection fails.