Showing the changes to Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberHandler.cpp (1 of 24 files). Show all
Unified · Split
| @@ -29,166 +29,166 @@ using namespace GSF::TimeSeries; · 28 unmodified lines | |||
| 29 | 29 | ||
| 30 | SubscriberHandler::SubscriberHandler(const string& name) | 30 | SubscriberHandler::SubscriberHandler(const string& name) |
| 31 | { | 31 | { |
| 32 | m_name = name; | 32 | m_name = name; |
| 33 | } | 33 | } |
| 34 | 34 | ||
| 35 | SubscriptionInfo SubscriberHandler::CreateSubscriptionInfo() | 35 | SubscriptionInfo SubscriberHandler::CreateSubscriptionInfo() |
| 36 | { | 36 | { |
| 37 | SubscriptionInfo info = SubscriberInstance::CreateSubscriptionInfo(); | 37 | SubscriptionInfo info = SubscriberInstance::CreateSubscriptionInfo(); |
| 38 | 38 | ||
| 39 | // TODO: Modify subscription info properties as desired... | 39 | // TODO: Modify subscription info properties as desired... |
| 40 | 40 | ||
| 41 | // To set up a remotely synchronized subscription, set this flag | 41 | // To set up a remotely synchronized subscription, set this flag |
| 42 | // to true and add the framesPerSecond parameter to the | 42 | // to true and add the framesPerSecond parameter to the |
| 43 | // ExtraConnectionStringParameters. Additionally, the following | 43 | // ExtraConnectionStringParameters. Additionally, the following |
| 44 | // example demonstrates the use of some other useful parameters | 44 | // example demonstrates the use of some other useful parameters |
| 45 | // when setting up remotely synchronized subscriptions. | 45 | // when setting up remotely synchronized subscriptions. |
| 46 | 46 | ||
| 47 | //info.RemotelySynchronized = true; | 47 | //info.RemotelySynchronized = true; |
| 48 | //info.ExtraConnectionStringParameters = "framesPerSecond=30;timeResolution=10000;downsamplingMethod=Closest"; | 48 | //info.ExtraConnectionStringParameters = "framesPerSecond=30;timeResolution=10000;downsamplingMethod=Closest"; |
| 49 | //info.LagTime = 3.0; | 49 | //info.LagTime = 3.0; |
| 50 | //info.LeadTime = 1.0; | 50 | //info.LeadTime = 1.0; |
| 51 | //info.UseLocalClockAsRealTime = false; | 51 | //info.UseLocalClockAsRealTime = false; |
| 52 | 52 | ||
| 53 | // Other example properties (see SubscriptionInfo.h for all properties) | 53 | // Other example properties (see SubscriptionInfo.h for all properties) |
| 54 | //info.Throttled = false; | 54 | //info.Throttled = false; |
| 55 | //info.IncludeTime = true; | 55 | //info.IncludeTime = true; |
| 56 | //info.UseMillisecondResolution = true; | 56 | //info.UseMillisecondResolution = true; |
| 57 | 57 | ||
| 58 | return info; | 58 | return info; |
| 59 | } | 59 | } |
| 60 | 60 | ||
| 61 | void SubscriberHandler::StatusMessage(const string& message) | 61 | void SubscriberHandler::StatusMessage(const string& message) |
| 62 | { | 62 | { |
| 63 | // TODO: Make sure these messages get logged to an appropriate location | 63 | // TODO: Make sure these messages get logged to an appropriate location |
| 64 | // For now, the base class just displays to console: | 64 | // For now, the base class just displays to console: |
| 65 | stringstream status; | 65 | stringstream status; |
| 66 | 66 | ||
| 67 | status << "[" << m_name << "] " << message; | 67 | status << "[" << m_name << "] " << message; |
| 68 | 68 | ||
| 69 | SubscriberInstance::StatusMessage(status.str()); | 69 | SubscriberInstance::StatusMessage(status.str()); |
| 70 | } | 70 | } |
| 71 | 71 | ||
| 72 | void SubscriberHandler::ErrorMessage(const string& message) | 72 | void SubscriberHandler::ErrorMessage(const string& message) |
| 73 | { | 73 | { |
| 74 | // TODO: Make sure these messages get logged to an appropriate location | 74 | // TODO: Make sure these messages get logged to an appropriate location |
| 75 | // For now, the base class just displays to console: | 75 | // For now, the base class just displays to console: |
| 76 | stringstream status; | 76 | stringstream status; |
| 77 | 77 | ||
| 78 | status << "[" << m_name << "] " << message; | 78 | status << "[" << m_name << "] " << message; |
| 79 | 79 | ||
| 80 | SubscriberInstance::ErrorMessage(status.str()); | 80 | SubscriberInstance::ErrorMessage(status.str()); |
| 81 | } | 81 | } |
| 82 | 82 | ||
| 83 | void SubscriberHandler::DataStartTime(time_t unixSOC, int milliseconds) | 83 | void SubscriberHandler::DataStartTime(time_t unixSOC, int milliseconds) |
| 84 | { | 84 | { |
| 85 | // TODO: This reports timestamp of very first received measurement (if useful) | 85 | // TODO: This reports timestamp of very first received measurement (if useful) |
| 86 | } | 86 | } |
| 87 | 87 | ||
| 88 | void SubscriberHandler::ReceivedMetadata(const vector<uint8_t>& payload) | 88 | void SubscriberHandler::ReceivedMetadata(const vector<uint8_t>& payload) |
| 89 | { | 89 | { |
| 90 | stringstream message; | 90 | stringstream message; |
| 91 | message << "Received " << payload.size() << " bytes of metadata, parsing..."; | 91 | message << "Received " << payload.size() << " bytes of metadata, parsing..."; |
| 92 | StatusMessage(message.str()); | 92 | StatusMessage(message.str()); |
| 93 | 93 | ||
| 94 | SubscriberInstance::ReceivedMetadata(payload); | 94 | SubscriberInstance::ReceivedMetadata(payload); |
| 95 | } | 95 | } |
| 96 | 96 | ||
| 97 | void SubscriberHandler::ParsedMetadata() | 97 | void SubscriberHandler::ParsedMetadata() |
| 98 | { | 98 | { |
| 99 | StatusMessage("Metadata successfully parsed."); | 99 | StatusMessage("Metadata successfully parsed."); |
| 100 | } | 100 | } |
| 101 | 101 | ||
| 102 | void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& measurements) | 102 | void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& measurements) |
| 103 | { | 103 | { |
| 104 | // Start processing measurements | 104 | // Start processing measurements |
| 105 | for (auto &measurement: measurements) | 105 | for (auto &measurement : measurements) |
| 106 | { | 106 | { |
| 107 | time_t soc; | 107 | time_t soc; |
| 108 | int16_t milliseconds; | 108 | int16_t milliseconds; |
| 109 | 109 | ||
| 110 | // Get adjusted value | 110 | // Get adjusted value |
| 111 | float64_t value = measurement->AdjustedValue(); | 111 | float64_t value = measurement->AdjustedValue(); |
| 112 | 112 | ||
| 113 | // Get time converted to UNIX second of century plus milliseconds | 113 | // Get time converted to UNIX second of century plus milliseconds |
| 114 | measurement->GetUnixTime(soc, milliseconds); | 114 | measurement->GetUnixTime(soc, milliseconds); |
| 115 | 115 | ||
| 116 | // Handle per measurement quality flags | 116 | // Handle per measurement quality flags |
| 117 | int32_t qualityFlags = measurement->Flags; | 117 | int32_t qualityFlags = measurement->Flags; |
| 118 | 118 | ||
| 119 | ConfigurationFramePtr configurationFrame; | 119 | ConfigurationFramePtr configurationFrame; |
| 120 | 120 | ||
| 121 | // Find associated configuration for measurement | 121 | // Find associated configuration for measurement |
| 122 | if (TryFindTargetConfigurationFrame(measurement->SignalID, configurationFrame)) | 122 | if (TryFindTargetConfigurationFrame(measurement->SignalID, configurationFrame)) |
| 123 | { | 123 | { |
| 124 | MeasurementMetadataPtr measurementMetadata; | 124 | MeasurementMetadataPtr measurementMetadata; |
| 125 | 125 | ||
| 126 | // Lookup measurement metadata - it's faster to find metadata from within configuration frame | 126 | // Lookup measurement metadata - it's faster to find metadata from within configuration frame |
| 127 | if (TryGetMeasurementMetdataFromConfigurationFrame(measurement->SignalID, configurationFrame, measurementMetadata)) | 127 | if (TryGetMeasurementMetdataFromConfigurationFrame(measurement->SignalID, configurationFrame, measurementMetadata)) |
| 128 | { | 128 | { |
| 129 | const SignalReference& reference = measurementMetadata->Reference; | 129 | const SignalReference& reference = measurementMetadata->Reference; |
| 130 | |||
| 131 | // reference.Acronym << target device acronym | ||
| 132 | // reference.Kind << kind of signal (see SignalKind in "Types.h"), like Frequency, Angle, etc | ||
| 133 | // reference.Index << for Phasors, Analogs and Digitals - this is the ordered "index" | ||
| 134 | 130 | ||
| 135 | // TODO: Handle measurement processing here... | 131 | // reference.Acronym << target device acronym |
| 136 | } | 132 | // reference.Kind << kind of signal (see SignalKind in "Types.h"), like Frequency, Angle, etc |
| 137 | } | 133 | // reference.Index << for Phasors, Analogs and Digitals - this is the ordered "index" |
| 138 | } | ||
| 139 | 134 | ||
| 140 | // TODO: *** Temporary Testing Code Below *** -- REMOVE BEFOR USE | 135 | // TODO: Handle measurement processing here... |
| 141 | const string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f"; | 136 | } |
| 142 | const size_t MaxTimestampSize = 80; | 137 | } |
| 138 | } | ||
| 143 | 139 | ||
| 144 | static int processCount = 0; | 140 | // TODO: *** Temporary Testing Code Below *** -- REMOVE BEFORE USE |
| 145 | size_t i; | 141 | const string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f"; |
| 142 | const size_t MaxTimestampSize = 80; | ||
| 146 | 143 | ||
| 147 | char timestamp[MaxTimestampSize]; | 144 | static int processCount = 0; |
| 145 | size_t i; | ||
| 148 | 146 | ||
| 149 | // Only display messages every five | 147 | char timestamp[MaxTimestampSize]; |
| 150 | // seconds (assuming 30 calls per second). | ||
| 151 | if (processCount % 150 == 0) | ||
| 152 | { | ||
| 153 | stringstream message; | ||
| 154 | 148 | ||
| 155 | message << GetTotalMeasurementsReceived() << " measurements received so far..." << endl; | 149 | // Only display messages every five |
| 150 | // seconds (assuming 30 calls per second). | ||
| 151 | if (processCount % 150 == 0) | ||
| 152 | { | ||
| 153 | stringstream message; | ||
| 156 | 154 | ||
| 157 | if (!measurements.empty()) | 155 | message << GetTotalMeasurementsReceived() << " measurements received so far..." << endl; |
| 158 | { | ||
| 159 | if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp)) | ||
| 160 | message << "Timestamp: " << string(timestamp) << endl; | ||
| 161 | 156 | ||
| 162 | message << "Point\tValue" << endl; | 157 | if (!measurements.empty()) |
| 158 | { | ||
| 159 | if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp)) | ||
| 160 | message << "Timestamp: " << string(timestamp) << endl; | ||
| 163 | 161 | ||
| 164 | for (i = 0; i < measurements.size(); ++i) | 162 | message << "Point\tValue" << endl; |
| 165 | message << measurements[i]->ID << '\t' << measurements[i]->Value << endl; | ||
| 166 | 163 | ||
| 167 | message << endl; | 164 | for (i = 0; i < measurements.size(); ++i) |
| 168 | } | 165 | message << measurements[i]->ID << '\t' << measurements[i]->Value << endl; |
| 169 | 166 | ||
| 170 | StatusMessage(message.str()); | 167 | message << endl; |
| 171 | } | 168 | } |
| 172 | 169 | ||
| 173 | ++processCount; | 170 | StatusMessage(message.str()); |
| 171 | } | ||
| 172 | |||
| 173 | ++processCount; | ||
| 174 | } | 174 | } |
| 175 | 175 | ||
| 176 | void SubscriberHandler::ConfigurationChanged() | 176 | void SubscriberHandler::ConfigurationChanged() |
| 177 | { | 177 | { |
| 178 | StatusMessage("Configuration change detected. Metadata refresh requested."); | 178 | StatusMessage("Configuration change detected. Metadata refresh requested."); |
| 179 | } | 179 | } |
| 180 | 180 | ||
| 181 | void SubscriberHandler::HistoricalReadComplete() | 181 | void SubscriberHandler::HistoricalReadComplete() |
| 182 | { | 182 | { |
| 183 | StatusMessage("Historical data read complete."); | 183 | StatusMessage("Historical data read complete."); |
| 184 | } | 184 | } |
| 185 | 185 | ||
| 186 | void SubscriberHandler::ConnectionEstablished() | 186 | void SubscriberHandler::ConnectionEstablished() |
| 187 | { | 187 | { |
| 188 | StatusMessage("Connection established."); | 188 | StatusMessage("Connection established."); |
| 189 | } | 189 | } |
| 190 | 190 | ||
| 191 | void SubscriberHandler::ConnectionTerminated() | 191 | void SubscriberHandler::ConnectionTerminated() |
| 192 | { | 192 | { |
| 193 | StatusMessage("Connection terminated."); | 193 | StatusMessage("Connection terminated."); |
| 194 | } | 194 | } |