Showing the changes to Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberHandler.cpp (1 of 5 files). Show all
Unified · Split
| @@ -22,18 +22,18 @@ · 21 unmodified lines | ||
| 22 | 22 | //****************************************************************************************************** |
| 23 | 23 | |
| 24 | 24 | #include "SubscriberHandler.h" |
| 25 | − | |
| 26 | −// TODO: These includes are just for temporary testing code | |
| 27 | 25 | #include "../Common/Convert.h" |
| 28 | −#include <boost/uuid/uuid_io.hpp> | |
| 29 | 26 | |
| 30 | 27 | using namespace std; |
| 31 | 28 | using namespace GSF; |
| 32 | 29 | using namespace GSF::TimeSeries; |
| 33 | 30 | using namespace GSF::TimeSeries::Transport; |
| 34 | 31 | |
| 32 | +Mutex SubscriberHandler::s_coutLock {}; | |
| 33 | + | |
| 35 | 34 | SubscriberHandler::SubscriberHandler(string name) : |
| 36 | − m_name(std::move(name)) | |
| 35 | + m_name(std::move(name)), | |
| 36 | + m_processCount(0L) | |
| 37 | 37 | { |
| 38 | 38 | } |
| 39 | 39 | |
| @@ -79,9 +79,10 @@ void SubscriberHandler::StatusMessage(const string& message) · 39 unmodified lines | ||
| 79 | 79 | |
| 80 | 80 | status << "[" << m_name << "] " << message; |
| 81 | 81 | |
| 82 | − m_coutLock.lock(); | |
| 82 | + // Calls can come from multiple threads, so we impose a simple lock before write to console | |
| 83 | + s_coutLock.lock(); | |
| 83 | 84 | SubscriberInstance::StatusMessage(status.str()); |
| 84 | − m_coutLock.unlock(); | |
| 85 | + s_coutLock.unlock(); | |
| 85 | 86 | } |
| 86 | 87 | |
| 87 | 88 | void SubscriberHandler::ErrorMessage(const string& message) |
| @@ -92,9 +93,10 @@ void SubscriberHandler::ErrorMessage(const string& message) · 4 unmodified lines | ||
| 92 | 93 | |
| 93 | 94 | status << "[" << m_name << "] " << message; |
| 94 | 95 | |
| 95 | − m_coutLock.lock(); | |
| 96 | + // Calls can come from multiple threads, so we impose a simple lock before write to console | |
| 97 | + s_coutLock.lock(); | |
| 96 | 98 | SubscriberInstance::ErrorMessage(status.str()); |
| 97 | − m_coutLock.unlock(); | |
| 99 | + s_coutLock.unlock(); | |
| 98 | 100 | } |
| 99 | 101 | |
| 100 | 102 | void SubscriberHandler::DataStartTime(time_t unixSOC, uint16_t milliseconds) |
| @@ -124,17 +126,15 @@ void SubscriberHandler::ParsedMetadata() · 23 unmodified lines | ||
| 124 | 126 | // ReSharper disable CppDeclaratorNeverUsed |
| 125 | 127 | void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& measurements) |
| 126 | 128 | { |
| 129 | + // TODO: The following code could be used to generate frame based output, e.g., for IEEE C37.118 | |
| 127 | 130 | // Start processing measurements |
| 128 | 131 | //for (auto &measurement : measurements) |
| 129 | 132 | //{ |
| 130 | − // time_t soc; | |
| 131 | − // uint16_t milliseconds; | |
| 132 | − | |
| 133 | 133 | // // Get adjusted value |
| 134 | 134 | // const float64_t value = measurement->AdjustedValue(); |
| 135 | 135 | |
| 136 | − // // Get time converted to UNIX second of century plus milliseconds | |
| 137 | − // measurement->GetUnixTime(soc, milliseconds); | |
| 136 | + // // Get timestamp | |
| 137 | + // DateTime timestamp = measurement->GetDateTime(); | |
| 138 | 138 | |
| 139 | 139 | // // Handle per measurement quality flags |
| 140 | 140 | // int32_t qualityFlags = measurement->Flags; |
| @@ -152,7 +152,7 @@ void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& me · 11 unmodified lines | ||
| 152 | 152 | |
| 153 | 153 | // // reference.Acronym << target device acronym |
| 154 | 154 | // // reference.Kind << kind of signal (see SignalKind in "Types.h"), like Frequency, Angle, etc |
| 155 | − // // reference.Index << for Phasors, Analogs and Digitals - this is the ordered "index" | |
| 155 | + // // reference.Index << for Phasors, Analogs and Digitals - this is the ordered "index" | |
| 156 | 156 | |
| 157 | 157 | // // TODO: Handle measurement processing here... |
| 158 | 158 | // } |
| @@ -165,16 +165,11 @@ void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& me · 6 unmodified lines | ||
| 165 | 165 | //} |
| 166 | 166 | |
| 167 | 167 | // TODO: *** Temporary Testing Code Below *** -- REMOVE BEFORE USE |
| 168 | − const string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f"; | |
| 169 | − const uint32_t MaxTimestampSize = 80; | |
| 168 | + static const uint64_t interval = 10 * 60; | |
| 169 | + const uint64_t measurementCount = measurements.size(); | |
| 170 | + const bool showMessage = (m_processCount + measurementCount >= (m_processCount / interval + 1) * interval); | |
| 170 | 171 | |
| 171 | − static long processCount = 0; | |
| 172 | − static char timestamp[MaxTimestampSize]; | |
| 173 | − static const long interval = 10 * 60; | |
| 174 | − const long measurementCount = measurements.size(); | |
| 175 | − const bool showMessage = (processCount + measurementCount >= (processCount / interval + 1) * interval); | |
| 176 | − | |
| 177 | − processCount += measurementCount; | |
| 172 | + m_processCount += measurementCount; | |
| 178 | 173 | |
| 179 | 174 | // Only display messages every few seconds |
| 180 | 175 | if (showMessage) |
| @@ -182,12 +177,8 @@ void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& me · 1 unmodified line | ||
| 182 | 177 | stringstream message; |
| 183 | 178 | |
| 184 | 179 | message << GetTotalMeasurementsReceived() << " measurements received so far..." << endl; |
| 185 | − | |
| 186 | − if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp)) | |
| 187 | − message << string(timestamp) << endl; | |
| 188 | − | |
| 189 | − message << "Signal ID: " << boost::lexical_cast<string>(measurements[0]->SignalID) << endl; | |
| 190 | − | |
| 180 | + message << ToString(measurements[0]->GetDateTime()) << endl; | |
| 181 | + message << "Signal ID: " << ToString(measurements[0]->SignalID) << endl; | |
| 191 | 182 | message << "\tPoint\tValue" << endl; |
| 192 | 183 | |
| 193 | 184 | for (const auto& measurement : measurements) |