Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Deepstream 5.0 Subscription Implementation #1

Open
wants to merge 32 commits into
base: master
Choose a base branch
from
Open

Deepstream 5.0 Subscription Implementation #1

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
32 commits
Select commit Hold shift + click to select a range
2e29abb
Add msgapi_subscribe implementation
Shubhamturakhia Apr 5, 2021
2a7a3d6
Modify broker.c file
Shubhamturakhia Apr 6, 2021
9a5bc7a
Added subcribe implementation
Shubhamturakhia Apr 7, 2021
b2c6940
Added Subscribe implementation
Shubhamturakhia Apr 7, 2021
bf90453
Minor changes
Shubhamturakhia Apr 7, 2021
67dffec
Changes according to PR review
Shubhamturakhia Apr 7, 2021
5ca5006
Add callback wrapper
Shubhamturakhia Apr 8, 2021
4821a94
Added unsubscribe feature
Shubhamturakhia Apr 12, 2021
adc30df
Changes as per PR review
Shubhamturakhia Apr 12, 2021
f8f0099
Minor change
Shubhamturakhia Apr 14, 2021
7aa84e7
Minor change
Shubhamturakhia Apr 14, 2021
9a8ff42
Fix formatting
Shubhamturakhia Apr 14, 2021
bd3310d
Update aws_protocol_adaptor/device_client/aws_nvmsgbroker.c
Shubhamturakhia Apr 14, 2021
7e4b8d9
Add msgapi connection signature mwthod
Shubhamturakhia Apr 15, 2021
f2cd2b9
Merge branch 'DS-5.0-subscription-implementation' of https://github.c…
Shubhamturakhia Apr 15, 2021
3707004
added connection signature method
Shubhamturakhia Apr 16, 2021
2fd20b1
Minor changes
Shubhamturakhia Apr 16, 2021
072dfbb
Modified utilities for conn signature
Shubhamturakhia Apr 16, 2021
93deacc
Formatting and minor refactoring
Shubhamturakhia Apr 19, 2021
9c1449a
Changes as per PR review
Shubhamturakhia Apr 20, 2021
fa94cc3
Modifications for Hash code
Shubhamturakhia Apr 21, 2021
525ee45
Changes as per review
Shubhamturakhia Apr 21, 2021
579d7b4
Minor indentation change
Shubhamturakhia Apr 21, 2021
d330130
Hash code refactoring
Shubhamturakhia Apr 22, 2021
819cdc0
Add hash code implementation for config file
Shubhamturakhia Apr 23, 2021
0bac762
Fix Incorrect Return Types
lummish Apr 23, 2021
ef454fb
Minor changes
Shubhamturakhia Apr 23, 2021
9c61435
Resolve Compilation Errors
lummish Apr 23, 2021
d9fa5a6
Link OpenSSL in Makefile
lummish Apr 23, 2021
54fff91
Statically Link to libssl and libcrypto
lummish Apr 27, 2021
c99d2de
Reorder SSL Linkages
lummish Apr 27, 2021
0c7895e
Reorder Linkage to SSL Libraries
lummish Apr 27, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 42 additions & 2 deletions aws_protocol_adaptor/device_client/aws_nvmsgbroker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,44 @@ NvDsMsgApiErrorType nvds_msgapi_send_async(NvDsMsgApiHandle h_ptr, char *topic,
return NVDS_MSGAPI_OK;
}

/* ************************************************************************* */
// Subscribe def
/* ************************************************************************* */
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Get rid of this comment.


void subscribe_callback_handler(AWS_IoT_Client *pClient, char *topicName, uint16_t topicNameLen,
IoT_Publish_Message_Params *params, void *pData)
{
IOT_INFO("Subscribe callback");
IOT_INFO("%.*s\t%.*s", topicNameLen, topicName, (int)params->payloadLen, (char *)params->payload);
}
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this here? See comments below.


NvDsMsgApiErrorType nvds_msgapi_subscribe(NvDsMsgApiHandle h_ptr, char **topics, int num_topics, nvds_msgapi_subscribe_request_cb_t cb, void *user_ctx)
{
printf("Subscribe called\n");
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You should use the IOT logging functions.


if ((h_ptr == NULL) || (topics == NULL) || (num_topics <= 0))
{
IOT_ERROR("Essensial args missing for function nvds_msgapi_subscribe: %d, %d, %d\n", (h_ptr == NULL), (topics == NULL), (num_topics == NULL);
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Typo: "essensial"

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thats the typo in the whole file I will correct that in all other places as well

return NVDS_MSGAPI_ERR;
}
if (!cb)
{
IOT_ERROR("Callback function for nvds_msgapi_send cannot be NULL\n");
return NVDS_MSGAPI_ERR;
}
IoT_Error_t rc = FAILURE;
AWS_IoT_Client *client = (AWS_IoT_Client *)h_ptr;
rc = aws_iot_mqtt_subscribe(client, topics, strlen(topics), QOS0, subscribe_callback_handler, NULL);
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please review the documentation for this function.

There are a few problems here:

  1. This likely will not compile as you are not using the correct interface for aws_iot_mqtt_subscribe
  2. aws_iot_mqtt_subscribe only subscribes to a single topic (see 1).
  3. You're calling strlen on an array of strings, not a string.
  4. You're not using cb at all. Instead you've provided your own callback, so the argument (and by extension, the subscription) is essentially useless.
  5. You're not using user_ctx

if (SUCCESS != rc)
{
IOT_ERROR("Unable to subscribe, error: %d\n", rc);
return NVDS_MSGAPI_ERR;
}
IOT_INFO("Successfully subscribed");
g_free(client);
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You realize this will destroy the client, right? Why would we want to do that?

return NVDS_MSGAPI_OK;
}

/* ************************************************************************* */
// Do Work function def
/* ************************************************************************* */
Expand Down Expand Up @@ -277,7 +315,8 @@ void nvds_msgapi_do_work(NvDsMsgApiHandle h_ptr)
}
return;
}
while (! g_queue_is_empty(work_queue)){
while (!g_queue_is_empty(work_queue))
{
Work *work_node = (Work *)g_queue_pop_head(work_queue);
AWS_IoT_Client *client = (AWS_IoT_Client *)work_node->h_ptr;
rc = _mqtt_msg_send(client, work_node->topic, work_node->payload, work_node->payload_size);
Expand All @@ -291,7 +330,8 @@ void nvds_msgapi_do_work(NvDsMsgApiHandle h_ptr)
g_free(work_node);
return;
}
if (work_node->call_back_handler != NULL){
if (work_node->call_back_handler != NULL)
{
IOT_INFO("Pointer callback.");
work_node->call_back_handler(work_node->user_ptr, NVDS_MSGAPI_OK);
}
Expand Down
1 change: 1 addition & 0 deletions aws_protocol_adaptor/device_client/aws_nvmsgbroker.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,4 +41,5 @@ NvDsMsgApiHandle nvds_msgapi_connect(char *connection_str, nvds_msgapi_connect_c
NvDsMsgApiErrorType nvds_msgapi_disconnect(NvDsMsgApiHandle h_ptr);
NvDsMsgApiErrorType nvds_msgapi_send(NvDsMsgApiHandle conn, char *topic, const uint8_t *payload, size_t nbuf);
NvDsMsgApiErrorType nvds_msgapi_send_async(NvDsMsgApiHandle h_ptr, char *topic, const uint8_t *payload, size_t nbuf, nvds_msgapi_send_cb_t send_callback, void *user_ptr);
NvDsMsgApiErrorType nvds_msgapi_subscribe(NvDsMsgApiHandle h_ptr, char **topics, int num_topics, nvds_msgapi_subscribe_request_cb_t cb, void *user_ctx);
void nvds_msgapi_do_work(NvDsMsgApiHandle h_ptr);