Updated dispatch reference release callers to always run even when no handler has been registered.

ritchiecarroll · ritchiecarroll · commit 151c4313970d · 2019-07-03T15:29:04.000-04:00
diff --git a/src/lib/transport/DataPublisher.cpp b/src/lib/transport/DataPublisher.cpp
@@ -258,7 +258,7 @@ void DataPublisher::DispatchUserCommand(SubscriberConnection* connection, uint32
 // Dispatcher function for status messages. Decodes the message and provides it to the user via the status message callback.
 void DataPublisher::StatusMessageDispatcher(DataPublisher* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr || buffer.empty())
         return;
 
     const MessageCallback statusMessageCallback = source->m_statusMessageCallback;
@@ -270,7 +270,7 @@ void DataPublisher::StatusMessageDispatcher(DataPublisher* source, const vector<
 // Dispatcher function for error messages. Decodes the message and provides it to the user via the error message callback.
 void DataPublisher::ErrorMessageDispatcher(DataPublisher* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr || buffer.empty())
         return;
 
     const MessageCallback errorMessageCallback = source->m_errorMessageCallback;
@@ -281,95 +281,118 @@ void DataPublisher::ErrorMessageDispatcher(DataPublisher* source, const vector<u
 
 void DataPublisher::ClientConnectedDispatcher(DataPublisher* source, const vector<uint8_t>& buffer)
 {
-    SubscriberConnection* connection = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+    if (source == nullptr || buffer.empty())
+        return;
+
+    SubscriberConnection* connectionPtr = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
 
-    if (source != nullptr)
+    if (connectionPtr != nullptr)
     {
         const SubscriberConnectionCallback clientConnectedCallback = source->m_clientConnectedCallback;
+        const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);
 
         if (clientConnectedCallback != nullptr)
-            clientConnectedCallback(source, source->ReleaseDispatchReference(connection));
+            clientConnectedCallback(source, connectionRef);
     }
 }
 
 void DataPublisher::ClientDisconnectedDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)
 {
-    SubscriberConnection* connection = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+    if (source == nullptr || buffer.empty())
+        return;
 
-    if (source != nullptr)
+    SubscriberConnection* connectionPtr = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+
+    if (connectionPtr != nullptr)
     {
         const SubscriberConnectionCallback clientDisconnectedCallback = source->m_clientDisconnectedCallback;
-        const SubscriberConnectionPtr subscriberConnectionRef = source->ReleaseDispatchReference(connection);
+        const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);
 
         if (clientDisconnectedCallback != nullptr)
-            clientDisconnectedCallback(source, subscriberConnectionRef);
+            clientDisconnectedCallback(source, connectionRef);
 
-        source->RemoveConnection(subscriberConnectionRef);
+        source->RemoveConnection(connectionRef);
     }
 }
 
 void DataPublisher::ProcessingIntervalChangeRequestedDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)
 {
-    SubscriberConnection* connection = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+    if (source == nullptr || buffer.empty())
+        return;
+
+    SubscriberConnection* connectionPtr = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
 
-    if (source != nullptr)
+    if (connectionPtr != nullptr)
     {
         const SubscriberConnectionCallback temporalProcessingIntervalChangeRequestedCallback = source->m_processingIntervalChangeRequestedCallback;
+        const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);
 
         if (temporalProcessingIntervalChangeRequestedCallback != nullptr)
-            temporalProcessingIntervalChangeRequestedCallback(source, source->ReleaseDispatchReference(connection));
+            temporalProcessingIntervalChangeRequestedCallback(source, connectionRef);
     }
 }
 
 void DataPublisher::TemporalSubscriptionRequestedDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)
 {
-    SubscriberConnection* connection = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+    if (source == nullptr || buffer.empty())
+        return;
+
+    SubscriberConnection* connectionPtr = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
 
-    if (source != nullptr)
+    if (connectionPtr != nullptr)
     {
         const SubscriberConnectionCallback temporalSubscriptionRequestedCallback = source->m_temporalSubscriptionRequestedCallback;
+        const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);
 
         if (temporalSubscriptionRequestedCallback != nullptr)
-            temporalSubscriptionRequestedCallback(source, source->ReleaseDispatchReference(connection));
+            temporalSubscriptionRequestedCallback(source, connectionRef);
     }
 }
 
 void DataPublisher::TemporalSubscriptionCanceledDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)
 {
-    SubscriberConnection* connection = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+    if (source == nullptr || buffer.empty())
+        return;
 
-    if (source != nullptr)
+    SubscriberConnection* connectionPtr = *reinterpret_cast<SubscriberConnection**>(const_cast<uint8_t*>(&buffer[0]));
+
+    if (connectionPtr != nullptr)
     {
         const SubscriberConnectionCallback temporalSubscriptionCanceledCallback = source->m_temporalSubscriptionCanceledCallback;
+        const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);
 
         if (temporalSubscriptionCanceledCallback != nullptr)
-            temporalSubscriptionCanceledCallback(source, source->ReleaseDispatchReference(connection));
+            temporalSubscriptionCanceledCallback(source, connectionRef);
     }
 }
 
 void DataPublisher::UserCommandDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)
 {
+    if (source == nullptr || buffer.empty())
+        return;
+
     UserCommandData* userCommandData = *reinterpret_cast<UserCommandData**>(const_cast<uint8_t*>(&buffer[0]));
 
-    if (source != nullptr && userCommandData != nullptr)
+    if (userCommandData != nullptr && userCommandData->connection != nullptr)
     {
         const UserCommandCallback userCommandCallback = source->m_userCommandCallback;
+        const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(userCommandData->connection);
 
         if (userCommandCallback != nullptr)
-            userCommandCallback(source, source->ReleaseDispatchReference(userCommandData->connection), userCommandData->command, userCommandData->data);
+            userCommandCallback(source, connectionRef, userCommandData->command, userCommandData->data);
     }
 
     delete userCommandData;
 }
 
 int32_t DataPublisher::GetColumnIndex(const sttp::data::DataTablePtr& table, const std::string& columnName)
 {
-        const DataColumnPtr& column = table->Column(columnName);
+    const DataColumnPtr& column = table->Column(columnName);
     
-        if (column == nullptr)
-            throw PublisherException("Column name \"" + columnName + "\" was not found in table \"" + table->Name() + "\"");
+    if (column == nullptr)
+        throw PublisherException("Column name \"" + columnName + "\" was not found in table \"" + table->Name() + "\"");
     
-        return column->Index();
+    return column->Index();
 }
 
 void DataPublisher::DefineMetadata(const vector<DeviceMetadataPtr>& deviceMetadata, const vector<MeasurementMetadataPtr>& measurementMetadata, const vector<PhasorMetadataPtr>& phasorMetadata, const int32_t versionNumber)
diff --git a/src/lib/transport/DataSubscriber.cpp b/src/lib/transport/DataSubscriber.cpp
@@ -958,7 +958,7 @@ void DataSubscriber::DispatchErrorMessage(const string& message)
 // Dispatcher function for status messages. Decodes the message and provides it to the user via the status message callback.
 void DataSubscriber::StatusMessageDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr || buffer.empty())
         return;
     
     const MessageCallback statusMessageCallback = source->m_statusMessageCallback;
@@ -970,7 +970,7 @@ void DataSubscriber::StatusMessageDispatcher(DataSubscriber* source, const vecto
 // Dispatcher function for error messages. Decodes the message and provides it to the user via the error message callback.
 void DataSubscriber::ErrorMessageDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr || buffer.empty())
         return;
 
     const MessageCallback errorMessageCallback = source->m_errorMessageCallback;
@@ -982,7 +982,7 @@ void DataSubscriber::ErrorMessageDispatcher(DataSubscriber* source, const vector
 // Dispatcher function for data start time. Decodes the start time and provides it to the user via the data start time callback.
 void DataSubscriber::DataStartTimeDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr || buffer.empty())
         return;
 
     const DataStartTimeCallback dataStartTimeCallback = source->m_dataStartTimeCallback;
@@ -997,7 +997,7 @@ void DataSubscriber::DataStartTimeDispatcher(DataSubscriber* source, const vecto
 // Dispatcher function for metadata. Provides encoded metadata to the user via the metadata callback.
 void DataSubscriber::MetadataDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr) // Empty buffer OK
         return;
 
     const MetadataCallback metadataCallback = source->m_metadataCallback;
@@ -1008,21 +1008,25 @@ void DataSubscriber::MetadataDispatcher(DataSubscriber* source, const vector<uin
 
 void DataSubscriber::SubscriptionUpdatedDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer)
 {
-    SignalIndexCache* signalIndexCache = *reinterpret_cast<SignalIndexCache**>(const_cast<uint8_t*>(&buffer[0]));
+    if (source == nullptr || buffer.empty())
+        return;
+
+    SignalIndexCache* signalIndexCachePtr = *reinterpret_cast<SignalIndexCache**>(const_cast<uint8_t*>(&buffer[0]));
 
-    if (source != nullptr)
+    if (signalIndexCachePtr != nullptr)
     {
         const SubscriptionUpdatedCallback subscriptionUpdated = source->m_subscriptionUpdatedCallback;
+        const SignalIndexCachePtr signalIndexCacheRef = source->ReleaseDispatchReference(signalIndexCachePtr);
 
         if (subscriptionUpdated != nullptr)
-            subscriptionUpdated(source, source->ReleaseDispatchReference(signalIndexCache));
+            subscriptionUpdated(source, signalIndexCacheRef);
     }
 }
 
 // Dispatcher for processing complete message that is sent by the server at the end of a temporal session.
 void DataSubscriber::ProcessingCompleteDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr) // Empty buffer OK
         return;
 
     const MessageCallback processingCompleteCallback = source->m_processingCompleteCallback;
@@ -1041,7 +1045,7 @@ void DataSubscriber::ProcessingCompleteDispatcher(DataSubscriber* source, const
 // Dispatcher for processing complete message that is sent by the server at the end of a temporal session.
 void DataSubscriber::ConfigurationChangedDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
 {
-    if (source == nullptr)
+    if (source == nullptr) // Empty buffer OK
         return;
 
     const ConfigurationChangedCallback configurationChangedCallback = source->m_configurationChangedCallback;

Original file line number	Diff line number	Diff line change
`@@ -258,7 +258,7 @@ void DataPublisher::DispatchUserCommand(SubscriberConnection* connection, uint32`
`258`	`258`	`// Dispatcher function for status messages. Decodes the message and provides it to the user via the status message callback.`
`259`	`259`	`void DataPublisher::StatusMessageDispatcher(DataPublisher* source, const vector<uint8_t>& buffer)`
`260`	`260`	`{`
`261`		`- if (source == nullptr)`
	`261`	`+ if (source == nullptr \|\| buffer.empty())`
`262`	`262`	`return;`
`263`	`263`
`264`	`264`	`const MessageCallback statusMessageCallback = source->m_statusMessageCallback;`
`@@ -270,7 +270,7 @@ void DataPublisher::StatusMessageDispatcher(DataPublisher* source, const vector<`
`270`	`270`	`// Dispatcher function for error messages. Decodes the message and provides it to the user via the error message callback.`
`271`	`271`	`void DataPublisher::ErrorMessageDispatcher(DataPublisher* source, const vector<uint8_t>& buffer)`
`272`	`272`	`{`
`273`		`- if (source == nullptr)`
	`273`	`+ if (source == nullptr \|\| buffer.empty())`
`274`	`274`	`return;`
`275`	`275`
`276`	`276`	`const MessageCallback errorMessageCallback = source->m_errorMessageCallback;`
`@@ -281,95 +281,118 @@ void DataPublisher::ErrorMessageDispatcher(DataPublisher* source, const vector<u`
`281`	`281`
`282`	`282`	`void DataPublisher::ClientConnectedDispatcher(DataPublisher* source, const vector<uint8_t>& buffer)`
`283`	`283`	`{`
`284`		`- SubscriberConnection* connection = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`284`	`+ if (source == nullptr \|\| buffer.empty())`
	`285`	`+ return;`
	`286`	`+`
	`287`	`+ SubscriberConnection* connectionPtr = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
`285`	`288`
`286`		`- if (source != nullptr)`
	`289`	`+ if (connectionPtr != nullptr)`
`287`	`290`	`{`
`288`	`291`	`const SubscriberConnectionCallback clientConnectedCallback = source->m_clientConnectedCallback;`
	`292`	`+ const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);`
`289`	`293`
`290`	`294`	`if (clientConnectedCallback != nullptr)`
`291`		`- clientConnectedCallback(source, source->ReleaseDispatchReference(connection));`
	`295`	`+ clientConnectedCallback(source, connectionRef);`
`292`	`296`	`}`
`293`	`297`	`}`
`294`	`298`
`295`	`299`	`void DataPublisher::ClientDisconnectedDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)`
`296`	`300`	`{`
`297`		`- SubscriberConnection* connection = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`301`	`+ if (source == nullptr \|\| buffer.empty())`
	`302`	`+ return;`
`298`	`303`
`299`		`- if (source != nullptr)`
	`304`	`+ SubscriberConnection* connectionPtr = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`305`	`+`
	`306`	`+ if (connectionPtr != nullptr)`
`300`	`307`	`{`
`301`	`308`	`const SubscriberConnectionCallback clientDisconnectedCallback = source->m_clientDisconnectedCallback;`
`302`		`- const SubscriberConnectionPtr subscriberConnectionRef = source->ReleaseDispatchReference(connection);`
	`309`	`+ const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);`
`303`	`310`
`304`	`311`	`if (clientDisconnectedCallback != nullptr)`
`305`		`- clientDisconnectedCallback(source, subscriberConnectionRef);`
	`312`	`+ clientDisconnectedCallback(source, connectionRef);`
`306`	`313`
`307`		`- source->RemoveConnection(subscriberConnectionRef);`
	`314`	`+ source->RemoveConnection(connectionRef);`
`308`	`315`	`}`
`309`	`316`	`}`
`310`	`317`
`311`	`318`	`void DataPublisher::ProcessingIntervalChangeRequestedDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)`
`312`	`319`	`{`
`313`		`- SubscriberConnection* connection = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`320`	`+ if (source == nullptr \|\| buffer.empty())`
	`321`	`+ return;`
	`322`	`+`
	`323`	`+ SubscriberConnection* connectionPtr = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
`314`	`324`
`315`		`- if (source != nullptr)`
	`325`	`+ if (connectionPtr != nullptr)`
`316`	`326`	`{`
`317`	`327`	`const SubscriberConnectionCallback temporalProcessingIntervalChangeRequestedCallback = source->m_processingIntervalChangeRequestedCallback;`
	`328`	`+ const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);`
`318`	`329`
`319`	`330`	`if (temporalProcessingIntervalChangeRequestedCallback != nullptr)`
`320`		`- temporalProcessingIntervalChangeRequestedCallback(source, source->ReleaseDispatchReference(connection));`
	`331`	`+ temporalProcessingIntervalChangeRequestedCallback(source, connectionRef);`
`321`	`332`	`}`
`322`	`333`	`}`
`323`	`334`
`324`	`335`	`void DataPublisher::TemporalSubscriptionRequestedDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)`
`325`	`336`	`{`
`326`		`- SubscriberConnection* connection = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`337`	`+ if (source == nullptr \|\| buffer.empty())`
	`338`	`+ return;`
	`339`	`+`
	`340`	`+ SubscriberConnection* connectionPtr = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
`327`	`341`
`328`		`- if (source != nullptr)`
	`342`	`+ if (connectionPtr != nullptr)`
`329`	`343`	`{`
`330`	`344`	`const SubscriberConnectionCallback temporalSubscriptionRequestedCallback = source->m_temporalSubscriptionRequestedCallback;`
	`345`	`+ const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);`
`331`	`346`
`332`	`347`	`if (temporalSubscriptionRequestedCallback != nullptr)`
`333`		`- temporalSubscriptionRequestedCallback(source, source->ReleaseDispatchReference(connection));`
	`348`	`+ temporalSubscriptionRequestedCallback(source, connectionRef);`
`334`	`349`	`}`
`335`	`350`	`}`
`336`	`351`
`337`	`352`	`void DataPublisher::TemporalSubscriptionCanceledDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)`
`338`	`353`	`{`
`339`		`- SubscriberConnection* connection = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`354`	`+ if (source == nullptr \|\| buffer.empty())`
	`355`	`+ return;`
`340`	`356`
`341`		`- if (source != nullptr)`
	`357`	`+ SubscriberConnection* connectionPtr = reinterpret_cast<SubscriberConnection>(const_cast<uint8_t>(&buffer[0]));`
	`358`	`+`
	`359`	`+ if (connectionPtr != nullptr)`
`342`	`360`	`{`
`343`	`361`	`const SubscriberConnectionCallback temporalSubscriptionCanceledCallback = source->m_temporalSubscriptionCanceledCallback;`
	`362`	`+ const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(connectionPtr);`
`344`	`363`
`345`	`364`	`if (temporalSubscriptionCanceledCallback != nullptr)`
`346`		`- temporalSubscriptionCanceledCallback(source, source->ReleaseDispatchReference(connection));`
	`365`	`+ temporalSubscriptionCanceledCallback(source, connectionRef);`
`347`	`366`	`}`
`348`	`367`	`}`
`349`	`368`
`350`	`369`	`void DataPublisher::UserCommandDispatcher(DataPublisher* source, const std::vector<uint8_t>& buffer)`
`351`	`370`	`{`
	`371`	`+ if (source == nullptr \|\| buffer.empty())`
	`372`	`+ return;`
	`373`	`+`
`352`	`374`	`UserCommandData* userCommandData = reinterpret_cast<UserCommandData>(const_cast<uint8_t>(&buffer[0]));`
`353`	`375`
`354`		`- if (source != nullptr && userCommandData != nullptr)`
	`376`	`+ if (userCommandData != nullptr && userCommandData->connection != nullptr)`
`355`	`377`	`{`
`356`	`378`	`const UserCommandCallback userCommandCallback = source->m_userCommandCallback;`
	`379`	`+ const SubscriberConnectionPtr connectionRef = source->ReleaseDispatchReference(userCommandData->connection);`
`357`	`380`
`358`	`381`	`if (userCommandCallback != nullptr)`
`359`		`- userCommandCallback(source, source->ReleaseDispatchReference(userCommandData->connection), userCommandData->command, userCommandData->data);`
	`382`	`+ userCommandCallback(source, connectionRef, userCommandData->command, userCommandData->data);`
`360`	`383`	`}`
`361`	`384`
`362`	`385`	`delete userCommandData;`
`363`	`386`	`}`
`364`	`387`
`365`	`388`	`int32_t DataPublisher::GetColumnIndex(const sttp::data::DataTablePtr& table, const std::string& columnName)`
`366`	`389`	`{`
`367`		`- const DataColumnPtr& column = table->Column(columnName);`
	`390`	`+ const DataColumnPtr& column = table->Column(columnName);`
`368`	`391`
`369`		`- if (column == nullptr)`
`370`		`- throw PublisherException("Column name \"" + columnName + "\" was not found in table \"" + table->Name() + "\"");`
	`392`	`+ if (column == nullptr)`
	`393`	`+ throw PublisherException("Column name \"" + columnName + "\" was not found in table \"" + table->Name() + "\"");`
`371`	`394`
`372`		`- return column->Index();`
	`395`	`+ return column->Index();`
`373`	`396`	`}`
`374`	`397`
`375`	`398`	`void DataPublisher::DefineMetadata(const vector<DeviceMetadataPtr>& deviceMetadata, const vector<MeasurementMetadataPtr>& measurementMetadata, const vector<PhasorMetadataPtr>& phasorMetadata, const int32_t versionNumber)`