Add draft implementation of the subscription manager module #1091

aggarw13 · 2020-07-31T02:21:57Z

Subscription Manager For MQTT

Subscription Manager will be utilized in demo that exhibits connection sharing in a single-threaded demo

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

muneebahmed10 · 2020-07-31T02:39:53Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+    }
+    else
+    {
+        /* Should the topic string be copied? */


To accommodate varying use cases yes, so the pTopicFilter member of the SubscriptionManager_Record_t would need to be an array. However, since it's just a demo and you control everything, it's up to you; if you only use constant strings then I don't think it's a major issue as long as there's a comment calling that out.

Yeah, I think for the single-threaded demo purpose, I will keep it simple and avoid the memory allocation problem (currently, the record list is statically allocated for a fixed size, but with local copy of the topic filter, it would require a max limit on the topic filter size, and thereby, increase the record list structure size)

muneebahmed10 · 2020-07-31T02:41:21Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+        {
+            recordFound = true;
+        }
+    } while( ( matchingRecordIndex < recordListSize ) && ( recordFound == false ) );


What if more than one application registers a different callback to the same topic filter? Then this would only remove the first one.

Currently, the subscription manager is geared toward(s) a single-threaded demo that will showcase connection sharing for multiple libraries (like Jobs, Shadow) that need different callbacks for processing their topic-specific PUBLISH messages.

However, even in the case of a multi-threaded use-case, a single application's call to remove its callback shouldn't remove another application's callback for the same topic filter. In that case, the subscription manager would need to store some thread/application-specific information in the records.

It is a good point that Muneeb mentioned. What is the purpose of this Subscription manager? Is it only for a single task use case? Is there a plan to improve it to support it for connection sharing between multiple tasks?
If we are extending it to multiple tasks, yea we would need some way to identify the task from which it is called.

muneebahmed10 · 2020-07-31T02:43:56Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+
+    size_t recordListCursor = 0u;
+
+    while( findRecordForTopic( pPublishInfo->pTopicName,


I don't see why findRecordForTopic is necessary. Since you're iterating over every entry in the list, why not just use a for loop in this function?

Good point, removed

muneebahmed10 · 2020-07-31T02:47:53Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+
+    /* If  an exact match is required, return the result of the comparison above.
+     * Otherwise, attempt to match with MQTT wildcards in topic filters. */
+    if( requiresExactMatch == false )


Would we ever want to set this to true? I see you don't and that makes sense to me, so I think this function could just be removed and topicFilterMatch used in its place

I took the topic matching logic from the v4_beta branch here.
Looking at the call-site of this function in the v4_beta implementation, it seems like this parameter was added for a different use-case (of updating registered callback against the same topic for which subscription already exists).

I will remove the requiresExactMatch logic

muneebahmed10 · 2020-07-31T02:49:09Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+typedef struct SubscriptionManager_Record
+{
+    const char * pTopicFilter;
+    size_t topicFilterLength;


This can be a uint16_t, since filter lengths must fit in 2 bytes

…ogic based on feedback

leegeth · 2020-07-31T18:26:22Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+
+typedef struct SubscriptionManager_Record
+{
+    const char * pTopicFilter;


Which module is responsible for topic filter memory? Will application be responsible for all topic filter buffers if it subscribes to different topics?

Yes, the single-threaded application will be responsible to maintain the lifetime of the topic string memory that is provides to the subscription manager. (Ideally, in a multi-threaded use-case, the data should be copied into the record, but I am thinking of keeping it simple for the single-threaded demo)

demos/mqtt/mqtt_demo_connection_sharing/CMakeLists.txt

leegeth · 2020-07-31T18:28:57Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+#include <string.h>
+#include "mqtt_subscription_manager.h"
+
+#define MAX_SUBSCRIPTION_CALLBACK_RECORDS    5


I think this would need to be the default value. we would need to wrap with #ifndef to this define

Will probably look to move it to the demo_config.h

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

leegeth · 2020-07-31T19:01:21Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+                                         filterIndex,
+                                         &nameIndex,
+                                         &status );
+        }


Don't we have to check for the topics beginning with $ as well?

I think the '$' prefix reservation is specific to AWS IoT, so I'd think no

http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718107, "Topics beginning with $", it is part of the spec. Not really specific to AWS IoT.

The spec mentions that "The Server MUST NOT match Topic Filters starting with a wildcard character (# or +) with Topic Names beginning with a $ character".
The client does not need to perform any special handling for topics starting with "$". Our service (Jobs, Shadow, Provisioning, etc) subscribe to these topics (starting with "$") and the topic matching on receiving incoming PUBLISHes only involves performing an exact character match for the starting "$" in the topic 🙂

how does the topic matching happens if there are 2 subscriptions in the client for topics “+/monitor/Clients” and “$SYS/monitor/Clients”?

Good point, have added logic to avoid wildcard match for incoming PUBLISH topic names starting with the "$" character

leegeth · 2020-07-31T21:59:35Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+                        callbackRecordList[ listIndex ].requiresExactMatch ) == true )
+        {
+            /* Invoke the callback associated with the record as the topics match. */
+            callbackRecordList[ listIndex++ ].callback( packetIdentifier, pPublishInfo );


I think this will result in an infinite loop if matchTopic returns false

Good catch! Fixed

leegeth · 2020-07-31T22:06:52Z

demos/mqtt/mqtt_demo_connection_sharing/subscription-manager/mqtt_subscription_manager.c

+            ( strncmp( pTopicFilter, pRecord->pTopicFilter, topicFileterLength ) == 0 ) )
+        {
+            recordFound = true;
+        }


possible Infinite loop. matchingRecordIndex needs to be incremented.

Good catch, fixed!

… ID with MQTT context in Subscription Manager

…c in CMakeLists.txt of new demo

muneebahmed10 · 2020-08-03T21:52:04Z

demos/mqtt/mqtt_demo_connection_sharing/mqtt_demo_connection_sharing.c

+            ( void ) SubscriptionManager_RemoveCallback( DEMO_APP_1_TOPIC,
+                                                         strlen( DEMO_APP_1_TOPIC ) );
+        }
+    }


The logic of the above two blocks is repeated twice more, why not make it a function?

muneebahmed10 · 2020-08-03T21:53:12Z

demos/mqtt/mqtt_demo_connection_sharing/mqtt_demo_connection_sharing.c

+        LogInfo( ( "Unsubscribing from the MQTT topic %.*s.",
+                   strlen( DEMO_APP_1_TOPIC ),
+                   DEMO_APP_1_TOPIC ) );
+        returnStatus = unsubscribeFromTopic( &mqttContext, DEMO_APP_1_TOPIC );


Should the callbacks be removed from the subscription manager?

muneebahmed10 · 2020-08-03T21:54:53Z

demos/mqtt/mqtt_demo_connection_sharing/mqtt_demo_connection_sharing.c

+    /* Start with everything at 0. */
+    ( void ) memset( ( void * ) pSubscriptionList, 0x00, sizeof( pSubscriptionList ) );
+
+    /* This example subscribes to only one topic and uses QOS2. */


This comment isn't accurate anymore

muneebahmed10 · 2020-08-03T21:55:05Z

demos/mqtt/mqtt_demo_connection_sharing/mqtt_demo_connection_sharing.c

+    ( void ) memset( ( void * ) pSubscriptionList, 0x00, sizeof( pSubscriptionList ) );
+
+    /* This example subscribes to and unsubscribes from only one topic
+     * and uses QOS2. */


muneebahmed10 · 2020-08-03T22:01:28Z

demos/mqtt/mqtt_demo_connection_sharing/mqtt_demo_connection_sharing.c

+
+    handleIncomingPublish( pPublishInfo );
+
+    /* Send a Publish to the notify topic to signal that the App 1 callback has been invoked! */


sarenameas · 2020-08-19T17:44:18Z

So this will be closed?

Add draft implementation of the subscription manager module

833038d

muneebahmed10 reviewed Jul 31, 2020

View reviewed changes

Make simplification changes to topic matching and dispatch handling l…

f9f4d82

…ogic based on feedback

leegeth reviewed Jul 31, 2020

View reviewed changes

demos/mqtt/mqtt_demo_connection_sharing/CMakeLists.txt Outdated Show resolved Hide resolved

leegeth reviewed Jul 31, 2020

View reviewed changes

muneebahmed10 reviewed Jul 31, 2020

View reviewed changes

leegeth reviewed Jul 31, 2020

View reviewed changes

aggarw13 added 7 commits July 31, 2020 22:47

Fix bugs identified in review and update signatures to replace packet…

519b6c0

… ID with MQTT context in Subscription Manager

Add PoC implementation of connection sharing demo

176c3b8

Fix build issues of the demo target

47df1f6

Remove certificate download logic and fix header path dependency logi…

2468beb

…c in CMakeLists.txt of new demo

Fixes to get demo to function correctly

8c95534

Remove qos parameter from publishToTopic functions

77af0d3

Add documentation to subscription manager files

2dbe5f5

aggarw13 mentioned this pull request Aug 4, 2020

Add subscription manager for MQTT #1094

Merged

muneebahmed10 reviewed Aug 5, 2020

View reviewed changes

aggarw13 closed this Aug 19, 2020

aggarw13 deleted the mqtt-demo/connection-sharing branch August 19, 2020 23:33


		size_t recordListCursor = 0u;

		while( findRecordForTopic( pPublishInfo->pTopicName,


		handleIncomingPublish( pPublishInfo );

		/* Send a Publish to the notify topic to signal that the App 1 callback has been invoked! */

Add draft implementation of the subscription manager module #1091

Add draft implementation of the subscription manager module #1091

Uh oh!

Conversation

aggarw13 commented Jul 31, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Subscription Manager For MQTT

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sarenameas commented Aug 19, 2020

Uh oh!

Uh oh!

aggarw13 commented Jul 31, 2020 •

edited

Loading