Added DataSubscriber callback for subscription updated which provides access to SignalIndexCache when publisher updates it. This is important to providing "allowed" signal IDs.

ritchiecarroll · ritchiecarroll · commit 45f5acee2163 · 2019-06-28T22:08:04.000-04:00
diff --git a/src/lib/transport/DataSubscriber.cpp b/src/lib/transport/DataSubscriber.cpp
@@ -664,6 +664,8 @@ void DataSubscriber::HandleUpdateSignalIndexCache(uint8_t* data, uint32_t offset
     SignalIndexCachePtr signalIndexCache = NewSharedPtr<SignalIndexCache>();
     signalIndexCache->Parse(uncompressedBuffer, m_subscriberID);
     m_signalIndexCache.swap(signalIndexCache);
+
+    DispatchSubscriptionUpdated(m_signalIndexCache.get());
 }
 
 // Updates base time offsets.
@@ -888,6 +890,11 @@ void DataSubscriber::Dispatch(const DispatcherFunction& function, const uint8_t*
     m_callbackQueue.Enqueue(dispatcher);
 }
 
+void DataSubscriber::DispatchSubscriptionUpdated(SignalIndexCache* signalIndexCache)
+{
+    Dispatch(&SubscriptionUpdatedDispatcher, reinterpret_cast<uint8_t*>(&signalIndexCache), 0, sizeof(SignalIndexCache**));
+}
+
 // Invokes the status message callback on the callback thread and provides the given message to it.
 void DataSubscriber::DispatchStatusMessage(const string& message)
 {
@@ -953,6 +960,19 @@ void DataSubscriber::MetadataDispatcher(DataSubscriber* source, const vector<uin
         metadataCallback(source, buffer);
 }
 
+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)
+    {
+        const SubscriptionUpdatedCallback subscriptionUpdated = source->m_subscriptionUpdatedCallback;
+
+        if (subscriptionUpdated != nullptr)
+            subscriptionUpdated(source, signalIndexCache->GetReference());
+    }
+}
+
 // 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)
 {
@@ -1017,6 +1037,12 @@ void DataSubscriber::RegisterMetadataCallback(const MetadataCallback& metadataCa
     m_metadataCallback = metadataCallback;
 }
 
+// Registers the subscription updated callback.
+void DataSubscriber::RegisterSubscriptionUpdatedCallback(const SubscriptionUpdatedCallback& subscriptionUpdatedCallback)
+{
+    m_subscriptionUpdatedCallback = subscriptionUpdatedCallback;
+}
+
 // Registers the new measurements callback.
 void DataSubscriber::RegisterNewMeasurementsCallback(const NewMeasurementsCallback& newMeasurementsCallback)
 {
diff --git a/src/lib/transport/DataSubscriber.h b/src/lib/transport/DataSubscriber.h
@@ -158,6 +158,7 @@ namespace transport
         typedef std::function<void(DataSubscriber*, const std::string&)> MessageCallback;
         typedef std::function<void(DataSubscriber*, int64_t)> DataStartTimeCallback;
         typedef std::function<void(DataSubscriber*, const std::vector<uint8_t>&)> MetadataCallback;
+        typedef std::function<void(DataSubscriber*, const SignalIndexCachePtr&)> SubscriptionUpdatedCallback;
         typedef std::function<void(DataSubscriber*, const std::vector<MeasurementPtr>&)> NewMeasurementsCallback;
         typedef std::function<void(DataSubscriber*)> ConfigurationChangedCallback;
         typedef std::function<void(DataSubscriber*)> ConnectionTerminatedCallback;
@@ -225,6 +226,7 @@ namespace transport
         MessageCallback m_errorMessageCallback;
         DataStartTimeCallback m_dataStartTimeCallback;
         MetadataCallback m_metadataCallback;
+        SubscriptionUpdatedCallback m_subscriptionUpdatedCallback;
         NewMeasurementsCallback m_newMeasurementsCallback;
         MessageCallback m_processingCompleteCallback;
         ConfigurationChangedCallback m_configurationChangedCallback;
@@ -258,13 +260,15 @@ namespace transport
         // Dispatchers
         void Dispatch(const DispatcherFunction& function);
         void Dispatch(const DispatcherFunction& function, const uint8_t* data, uint32_t offset, uint32_t length);
+        void DispatchSubscriptionUpdated(SignalIndexCache* signalIndexCache);
         void DispatchStatusMessage(const std::string& message);
         void DispatchErrorMessage(const std::string& message);
 
         static void StatusMessageDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
         static void ErrorMessageDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
         static void DataStartTimeDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
         static void MetadataDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
+        static void SubscriptionUpdatedDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
         static void ProcessingCompleteDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
         static void ConfigurationChangedDispatcher(DataSubscriber* source, const std::vector<uint8_t>& buffer);
 
@@ -292,6 +296,7 @@ namespace transport
         //   void HandleErrorMessage(DataSubscriber* source, const string& message)
         //   void HandleDataStartTime(DataSubscriber* source, int64_t startTime)
         //   void HandleMetadata(DataSubscriber* source, const vector<uint8_t>& metadata)
+        //   void HandleSubscriptionUpdated(DataSubscriber* source, const SignalIndexCachePtr& signalIndexCache)
         //   void HandleNewMeasurements(DataSubscriber* source, const vector<MeasurementPtr>& newMeasurements)
         //   void HandleProcessingComplete(DataSubscriber* source, const string& message)
         //   void HandleConfigurationChanged(DataSubscriber* source)
@@ -303,6 +308,7 @@ namespace transport
         void RegisterErrorMessageCallback(const MessageCallback& errorMessageCallback);
         void RegisterDataStartTimeCallback(const DataStartTimeCallback& dataStartTimeCallback);
         void RegisterMetadataCallback(const MetadataCallback& metadataCallback);
+        void RegisterSubscriptionUpdatedCallback(const SubscriptionUpdatedCallback& subscriptionUpdatedCallback);
         void RegisterNewMeasurementsCallback(const NewMeasurementsCallback& newMeasurementsCallback);
         void RegisterProcessingCompleteCallback(const MessageCallback& processingCompleteCallback);
         void RegisterConfigurationChangedCallback(const ConfigurationChangedCallback& configurationChangedCallback);
diff --git a/src/lib/transport/SignalIndexCache.cpp b/src/lib/transport/SignalIndexCache.cpp
@@ -86,6 +86,16 @@ unordered_set<Guid> SignalIndexCache::GetSignalIDs() const
     return unordered_set<Guid>(m_signalIDList.begin(), m_signalIDList.end());
 }
 
+bool SignalIndexCache::GetSignalIDs(vector<Guid>& signalIDs) const
+{
+    signalIDs.clear();
+
+    for (Guid signalID : m_signalIDList)
+        signalIDs.emplace_back(signalID);
+
+    return !signalIDs.empty();
+}
+
 // Gets the first half of the human-readable measurement
 // key associated with the given 32-bit runtime ID.
 const string& SignalIndexCache::GetSource(const int32_t signalIndex) const
@@ -158,6 +168,11 @@ uint32_t SignalIndexCache::GetBinaryLength() const
     return m_binaryLength;
 }
 
+SignalIndexCachePtr SignalIndexCache::GetReference()
+{
+    return shared_from_this();
+}
+
 void SignalIndexCache::RecalculateBinaryLength(const SubscriberConnection& connection)
 {
     uint32_t binaryLength = 28;
diff --git a/src/lib/transport/SignalIndexCache.h b/src/lib/transport/SignalIndexCache.h
@@ -33,12 +33,15 @@ namespace sttp {
 namespace transport
 {
     class SubscriberConnection;
-    typedef SharedPtr<SubscriberConnection> SubscriberConnectionPtr;
+    typedef sttp::SharedPtr<SubscriberConnection> SubscriberConnectionPtr;
+
+    class SignalIndexCache;
+    typedef sttp::SharedPtr<SignalIndexCache> SignalIndexCachePtr;
 
     // Maps 16-bit runtime IDs to 128-bit globally unique IDs.
     // Additionally provides reverse lookup and an extra mapping
     // to human-readable measurement keys.
-    class SignalIndexCache
+    class SignalIndexCache : public sttp::EnableSharedThisPtr<SignalIndexCache> // NOLINT
     {
     private:
         std::unordered_map<int32_t, uint32_t> m_reference;
@@ -63,9 +66,12 @@ namespace transport
         // Gets the globally unique signal ID associated with the given 16-bit runtime ID.
         sttp::Guid GetSignalID(int32_t signalIndex) const;
 
-        //Gets the full list of signal IDs as an unordered set
+        // Gets the full list of signal IDs as an unordered set
         std::unordered_set<sttp::Guid> GetSignalIDs() const;
 
+        // Populates given list with a copy of signal IDs
+        bool GetSignalIDs(std::vector<sttp::Guid>& signalIDs) const;
+
         // Gets the first half of the human-readable measurement
         // key associated with the given 16-bit runtime ID.
         const std::string& GetSource(int32_t signalIndex) const;
@@ -88,14 +94,16 @@ namespace transport
         // a vector size, for an exact size call RecalculateBinaryLength first
         uint32_t GetBinaryLength() const;
 
+        SignalIndexCachePtr GetReference();
+
         void RecalculateBinaryLength(const SubscriberConnection& connection);
 
         void Parse(const std::vector<uint8_t>& buffer, Guid& subscriberID);
 
         void Serialize(const SubscriberConnection& connection, std::vector<uint8_t>& buffer);
     };
 
-    typedef SharedPtr<SignalIndexCache> SignalIndexCachePtr;
+    typedef sttp::SharedPtr<SignalIndexCache> SignalIndexCachePtr;
 }}
 
 #endif
diff --git a/src/lib/transport/SubscriberInstance.cpp b/src/lib/transport/SubscriberInstance.cpp
@@ -1001,6 +1001,10 @@ void SubscriberInstance::ParsedMetadata()
 {
 }
 
+void SubscriberInstance::SubscriptionUpdated(const SignalIndexCachePtr& signalIndexCache)
+{
+}
+
 void SubscriberInstance::ReceivedNewMeasurements(const vector<MeasurementPtr>& measurements)
 {
     if (m_receiveSimpleMeasurements)
@@ -1109,6 +1113,12 @@ void SubscriberInstance::HandleMetadata(DataSubscriber* source, const vector<uin
         source->Subscribe();
 }
 
+void SubscriberInstance::HandleSubscriptionUpdated(DataSubscriber* source, const SignalIndexCachePtr& signalIndexCache)
+{
+    SubscriberInstance* instance = static_cast<SubscriberInstance*>(source->GetUserData());
+    instance->SubscriptionUpdated(signalIndexCache);
+}
+
 void SubscriberInstance::HandleNewMeasurements(DataSubscriber* source, const vector<MeasurementPtr>& measurements)
 {
     SubscriberInstance* instance = static_cast<SubscriberInstance*>(source->GetUserData());
diff --git a/src/lib/transport/SubscriberInstance.h b/src/lib/transport/SubscriberInstance.h
@@ -68,6 +68,7 @@ namespace transport
         static void HandleErrorMessage(DataSubscriber* source, const std::string& message);
         static void HandleDataStartTime(DataSubscriber* source, int64_t startTime);
         static void HandleMetadata(DataSubscriber* source, const std::vector<uint8_t>& payload);
+        static void HandleSubscriptionUpdated(DataSubscriber* source, const SignalIndexCachePtr& signalIndexCache);
         static void HandleNewMeasurements(DataSubscriber* source, const std::vector<MeasurementPtr>& measurements);
         static void HandleProcessingComplete(DataSubscriber* source, const std::string& message);
         static void HandleConfigurationChanged(DataSubscriber* source);
@@ -86,6 +87,7 @@ namespace transport
         virtual void DataStartTime(datetime_t startTime);
         virtual void ReceivedMetadata(const std::vector<uint8_t>& payload);
         virtual void ParsedMetadata();
+        virtual void SubscriptionUpdated(const SignalIndexCachePtr& signalIndexCache);
         virtual void ReceivedNewMeasurements(const std::vector<MeasurementPtr>& measurements);
         virtual void ReceivedNewMeasurements(const SimpleMeasurement* measurements, int32_t length);
         virtual void ConfigurationChanged();
diff --git a/src/samples/InstanceSubscribe/SubscriberHandler.cpp b/src/samples/InstanceSubscribe/SubscriberHandler.cpp
@@ -187,6 +187,11 @@ void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& me
     }
 }
 
+void SubscriberHandler::SubscriptionUpdated(const SignalIndexCachePtr& signalIndexCache)
+{
+    StatusMessage("Publisher provided " + ToString(signalIndexCache->Count()) + " measurements in response to subscription.");
+}
+
 void SubscriberHandler::ConfigurationChanged()
 {
     StatusMessage("Configuration change detected. Metadata refresh requested.");
diff --git a/src/samples/InstanceSubscribe/SubscriberHandler.h b/src/samples/InstanceSubscribe/SubscriberHandler.h
@@ -44,6 +44,7 @@ class SubscriberHandler : public sttp::transport::SubscriberInstance
     void DataStartTime(sttp::datetime_t startTime) override;
     void ReceivedMetadata(const std::vector<uint8_t>& payload) override;
     void ReceivedNewMeasurements(const std::vector<sttp::transport::MeasurementPtr>& measurements) override;
+    void SubscriptionUpdated(const sttp::transport::SignalIndexCachePtr& signalIndexCache) override;
     void ParsedMetadata() override;
     void ConfigurationChanged() override;
     void HistoricalReadComplete() override;

Original file line number	Diff line number	Diff line change
`@@ -1001,6 +1001,10 @@ void SubscriberInstance::ParsedMetadata()`
`1001`	`1001`	`{`
`1002`	`1002`	`}`
`1003`	`1003`
	`1004`	`+void SubscriberInstance::SubscriptionUpdated(const SignalIndexCachePtr& signalIndexCache)`
	`1005`	`+{`
	`1006`	`+}`
	`1007`	`+`
`1004`	`1008`	`void SubscriberInstance::ReceivedNewMeasurements(const vector<MeasurementPtr>& measurements)`
`1005`	`1009`	`{`
`1006`	`1010`	`if (m_receiveSimpleMeasurements)`
`@@ -1109,6 +1113,12 @@ void SubscriberInstance::HandleMetadata(DataSubscriber* source, const vector<uin`
`1109`	`1113`	`source->Subscribe();`
`1110`	`1114`	`}`
`1111`	`1115`
	`1116`	`+void SubscriberInstance::HandleSubscriptionUpdated(DataSubscriber* source, const SignalIndexCachePtr& signalIndexCache)`
	`1117`	`+{`
	`1118`	`+ SubscriberInstance* instance = static_cast<SubscriberInstance*>(source->GetUserData());`
	`1119`	`+ instance->SubscriptionUpdated(signalIndexCache);`
	`1120`	`+}`
	`1121`	`+`
`1112`	`1122`	`void SubscriberInstance::HandleNewMeasurements(DataSubscriber* source, const vector<MeasurementPtr>& measurements)`
`1113`	`1123`	`{`
`1114`	`1124`	`SubscriberInstance* instance = static_cast<SubscriberInstance*>(source->GetUserData());`
Original file line number	Diff line number	Diff line change
`@@ -187,6 +187,11 @@ void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& me`
`187`	`187`	`}`
`188`	`188`	`}`
`189`	`189`
	`190`	`+void SubscriberHandler::SubscriptionUpdated(const SignalIndexCachePtr& signalIndexCache)`
	`191`	`+{`
	`192`	`+ StatusMessage("Publisher provided " + ToString(signalIndexCache->Count()) + " measurements in response to subscription.");`
	`193`	`+}`
	`194`	`+`
`190`	`195`	`void SubscriberHandler::ConfigurationChanged()`
`191`	`196`	`{`
`192`	`197`	`StatusMessage("Configuration change detected. Metadata refresh requested.");`