TimeSeriesPlatformLibrary: Simplfied data publisher threading implementations.

J. Ritchie Carroll committed 7 years ago

Commit 68c05e01 parent aea1d009 files · +219 −232Browse files

Showing the changes to Source/Libraries/TimeSeriesPlatformLibrary/Samples/PublisherHandler.cpp (1 of 9 files). Show all

Unified · Split

Source/Libraries/TimeSeriesPlatformLibrary/Samples/PublisherHandler.cpp+169−169

@@ -31,12 +31,12 @@ using namespace GSF::TimeSeries::Transport; · 30 unmodified lines
31	31	Mutex PublisherHandler::s_coutLock {};
32	32
33	33	PublisherHandler::PublisherHandler(string name, uint16_t port, bool ipV6) :
34		− PublisherInstance(port, ipV6),
35		− m_name(std::move(name)),
36		− m_processCount(0L),
37		− m_metadataVersion(0)
	34	+ PublisherInstance(port, ipV6),
	35	+ m_name(std::move(name)),
	36	+ m_processCount(0L),
	37	+ m_metadataVersion(0)
38	38	{
39		− Initialize();
	39	+ Initialize();
40	40	}
41	41
42	42	void PublisherHandler::StatusMessage(const string& message)
@@ -69,184 +69,184 @@ void PublisherHandler::ErrorMessage(const string& message) · 26 unmodified lines
69	69
70	70	void PublisherHandler::ClientConnected(const SubscriberConnectionPtr& connection)
71	71	{
72		− StatusMessage("Client \"" + connection->GetConnectionID() + "\" with subscriber ID " + ToString(connection->GetSubscriberID()) + " connected...\n\n");
	72	+ StatusMessage("Client \"" + connection->GetConnectionID() + "\" with subscriber ID " + ToString(connection->GetSubscriberID()) + " connected...\n\n");
73	73	}
74	74
75	75	void PublisherHandler::ClientDisconnected(const SubscriberConnectionPtr& connection)
76	76	{
77		− StatusMessage("Client \"" + connection->GetConnectionID() + "\" with subscriber ID " + ToString(connection->GetSubscriberID()) + " disconnected...\n\n");
	77	+ StatusMessage("Client \"" + connection->GetConnectionID() + "\" with subscriber ID " + ToString(connection->GetSubscriberID()) + " disconnected...\n\n");
78	78	}
79	79
80	80	void PublisherHandler::DefineMetadata()
81	81	{
82		− // This sample just generates random Guid measurement and device identifiers - for a production system,
83		− // these Guid values would need to persist between runs defining a permanent association between the
84		− // defined metadata and the identifier...
85		−
86		− DeviceMetadataPtr device1Metadata = NewSharedPtr<DeviceMetadata>();
87		− const datetime_t timestamp = UtcNow();
88		−
89		− // Add a device
90		− device1Metadata->Name = "Test PMU";
91		− device1Metadata->Acronym = ToUpper(Replace(device1Metadata->Name, " ", "", false));
92		− device1Metadata->UniqueID = NewGuid();
93		− device1Metadata->Longitude = 300;
94		− device1Metadata->Latitude = 200;
95		− device1Metadata->FramesPerSecond = 30;
96		− device1Metadata->ProtocolName = "GEP";
97		− device1Metadata->UpdatedOn = timestamp;
98		−
99		− m_deviceMetadata.emplace_back(device1Metadata);
100		−
101		− const string& pointTagPrefix = device1Metadata->Acronym + ".";
102		− const string& measurementSource = "PPA:";
103		− int runtimeIndex = 1;
104		−
105		− // Add a frequency measurement
106		− MeasurementMetadataPtr measurement1Metadata = NewSharedPtr<MeasurementMetadata>();
107		− measurement1Metadata->ID = measurementSource + ToString(runtimeIndex++);
108		− measurement1Metadata->PointTag = pointTagPrefix + "FREQ";
109		− measurement1Metadata->SignalID = NewGuid();
110		− measurement1Metadata->DeviceAcronym = device1Metadata->Acronym;
111		− measurement1Metadata->Reference.Acronym = device1Metadata->Acronym;
112		− measurement1Metadata->Reference.Kind = Frequency;
113		− measurement1Metadata->Reference.Index = 0;
114		− measurement1Metadata->PhasorSourceIndex = 0;
115		− measurement1Metadata->UpdatedOn = timestamp;
116		−
117		− // Add a dF/dt measurement
118		− MeasurementMetadataPtr measurement2Metadata = NewSharedPtr<MeasurementMetadata>();
119		− measurement2Metadata->ID = measurementSource + ToString(runtimeIndex++);
120		− measurement2Metadata->PointTag = pointTagPrefix + "DFDT";
121		− measurement2Metadata->SignalID = NewGuid();
122		− measurement2Metadata->DeviceAcronym = device1Metadata->Acronym;
123		− measurement2Metadata->Reference.Acronym = device1Metadata->Acronym;
124		− measurement2Metadata->Reference.Kind = DfDt;
125		− measurement2Metadata->Reference.Index = 0;
126		− measurement2Metadata->PhasorSourceIndex = 0;
127		− measurement2Metadata->UpdatedOn = timestamp;
128		−
129		− // Add a phase angle measurement
130		− MeasurementMetadataPtr measurement3Metadata = NewSharedPtr<MeasurementMetadata>();
131		− measurement3Metadata->ID = measurementSource + ToString(runtimeIndex++);
132		− measurement3Metadata->PointTag = pointTagPrefix + "VPHA";
133		− measurement3Metadata->SignalID = NewGuid();
134		− measurement3Metadata->DeviceAcronym = device1Metadata->Acronym;
135		− measurement3Metadata->Reference.Acronym = device1Metadata->Acronym;
136		− measurement3Metadata->Reference.Kind = Angle;
137		− measurement3Metadata->Reference.Index = 1; // First phase angle
138		− measurement3Metadata->PhasorSourceIndex = 1; // Match to Phasor.SourceIndex = 1
139		− measurement3Metadata->UpdatedOn = timestamp;
140		−
141		− // Add a phase magnitude measurement
142		− MeasurementMetadataPtr measurement4Metadata = NewSharedPtr<MeasurementMetadata>();
143		− measurement4Metadata->ID = measurementSource + ToString(runtimeIndex++);
144		− measurement4Metadata->PointTag = pointTagPrefix + "VPHM";
145		− measurement4Metadata->SignalID = NewGuid();
146		− measurement4Metadata->DeviceAcronym = device1Metadata->Acronym;
147		− measurement4Metadata->Reference.Acronym = device1Metadata->Acronym;
148		− measurement4Metadata->Reference.Kind = Magnitude;
149		− measurement4Metadata->Reference.Index = 1; // First phase magnitude
150		− measurement4Metadata->PhasorSourceIndex = 1; // Match to Phasor.SourceIndex = 1
151		− measurement4Metadata->UpdatedOn = timestamp;
152		−
153		− m_measurementMetadata.emplace_back(measurement1Metadata);
154		− m_measurementMetadata.emplace_back(measurement2Metadata);
155		− m_measurementMetadata.emplace_back(measurement3Metadata);
156		− m_measurementMetadata.emplace_back(measurement4Metadata);
157		−
158		− // Add a phasor
159		− PhasorMetadataPtr phasor1Metadata = NewSharedPtr<PhasorMetadata>();
160		− phasor1Metadata->DeviceAcronym = device1Metadata->Acronym;
161		− phasor1Metadata->Label = device1Metadata->Name + " Voltage Phasor";
162		− phasor1Metadata->Type = "V"; // Voltage phasor
163		− phasor1Metadata->Phase = "+"; // Positive sequence
164		− phasor1Metadata->SourceIndex = 1; // Phasor number 1
165		− phasor1Metadata->UpdatedOn = timestamp;
166		−
167		− m_phasorMetadata.emplace_back(phasor1Metadata);
168		−
169		− m_metadataVersion++;
170		−
171		− // Pass meta-data to publisher instance for proper conditioning
172		− PublisherInstance::DefineMetadata(m_deviceMetadata, m_measurementMetadata, m_phasorMetadata, m_metadataVersion);
	82	+ // This sample just generates random Guid measurement and device identifiers - for a production system,
	83	+ // these Guid values would need to persist between runs defining a permanent association between the
	84	+ // defined metadata and the identifier...
	85	+
	86	+ DeviceMetadataPtr device1Metadata = NewSharedPtr<DeviceMetadata>();
	87	+ const datetime_t timestamp = UtcNow();
	88	+
	89	+ // Add a device
	90	+ device1Metadata->Name = "Test PMU";
	91	+ device1Metadata->Acronym = ToUpper(Replace(device1Metadata->Name, " ", "", false));
	92	+ device1Metadata->UniqueID = NewGuid();
	93	+ device1Metadata->Longitude = 300;
	94	+ device1Metadata->Latitude = 200;
	95	+ device1Metadata->FramesPerSecond = 30;
	96	+ device1Metadata->ProtocolName = "GEP";
	97	+ device1Metadata->UpdatedOn = timestamp;
	98	+
	99	+ m_deviceMetadata.emplace_back(device1Metadata);
	100	+
	101	+ const string& pointTagPrefix = device1Metadata->Acronym + ".";
	102	+ const string& measurementSource = "PPA:";
	103	+ int runtimeIndex = 1;
	104	+
	105	+ // Add a frequency measurement
	106	+ MeasurementMetadataPtr measurement1Metadata = NewSharedPtr<MeasurementMetadata>();
	107	+ measurement1Metadata->ID = measurementSource + ToString(runtimeIndex++);
	108	+ measurement1Metadata->PointTag = pointTagPrefix + "FREQ";
	109	+ measurement1Metadata->SignalID = NewGuid();
	110	+ measurement1Metadata->DeviceAcronym = device1Metadata->Acronym;
	111	+ measurement1Metadata->Reference.Acronym = device1Metadata->Acronym;
	112	+ measurement1Metadata->Reference.Kind = Frequency;
	113	+ measurement1Metadata->Reference.Index = 0;
	114	+ measurement1Metadata->PhasorSourceIndex = 0;
	115	+ measurement1Metadata->UpdatedOn = timestamp;
	116	+
	117	+ // Add a dF/dt measurement
	118	+ MeasurementMetadataPtr measurement2Metadata = NewSharedPtr<MeasurementMetadata>();
	119	+ measurement2Metadata->ID = measurementSource + ToString(runtimeIndex++);
	120	+ measurement2Metadata->PointTag = pointTagPrefix + "DFDT";
	121	+ measurement2Metadata->SignalID = NewGuid();
	122	+ measurement2Metadata->DeviceAcronym = device1Metadata->Acronym;
	123	+ measurement2Metadata->Reference.Acronym = device1Metadata->Acronym;
	124	+ measurement2Metadata->Reference.Kind = DfDt;
	125	+ measurement2Metadata->Reference.Index = 0;
	126	+ measurement2Metadata->PhasorSourceIndex = 0;
	127	+ measurement2Metadata->UpdatedOn = timestamp;
	128	+
	129	+ // Add a phase angle measurement
	130	+ MeasurementMetadataPtr measurement3Metadata = NewSharedPtr<MeasurementMetadata>();
	131	+ measurement3Metadata->ID = measurementSource + ToString(runtimeIndex++);
	132	+ measurement3Metadata->PointTag = pointTagPrefix + "VPHA";
	133	+ measurement3Metadata->SignalID = NewGuid();
	134	+ measurement3Metadata->DeviceAcronym = device1Metadata->Acronym;
	135	+ measurement3Metadata->Reference.Acronym = device1Metadata->Acronym;
	136	+ measurement3Metadata->Reference.Kind = Angle;
	137	+ measurement3Metadata->Reference.Index = 1; // First phase angle
	138	+ measurement3Metadata->PhasorSourceIndex = 1; // Match to Phasor.SourceIndex = 1
	139	+ measurement3Metadata->UpdatedOn = timestamp;
	140	+
	141	+ // Add a phase magnitude measurement
	142	+ MeasurementMetadataPtr measurement4Metadata = NewSharedPtr<MeasurementMetadata>();
	143	+ measurement4Metadata->ID = measurementSource + ToString(runtimeIndex++);
	144	+ measurement4Metadata->PointTag = pointTagPrefix + "VPHM";
	145	+ measurement4Metadata->SignalID = NewGuid();
	146	+ measurement4Metadata->DeviceAcronym = device1Metadata->Acronym;
	147	+ measurement4Metadata->Reference.Acronym = device1Metadata->Acronym;
	148	+ measurement4Metadata->Reference.Kind = Magnitude;
	149	+ measurement4Metadata->Reference.Index = 1; // First phase magnitude
	150	+ measurement4Metadata->PhasorSourceIndex = 1; // Match to Phasor.SourceIndex = 1
	151	+ measurement4Metadata->UpdatedOn = timestamp;
	152	+
	153	+ m_measurementMetadata.emplace_back(measurement1Metadata);
	154	+ m_measurementMetadata.emplace_back(measurement2Metadata);
	155	+ m_measurementMetadata.emplace_back(measurement3Metadata);
	156	+ m_measurementMetadata.emplace_back(measurement4Metadata);
	157	+
	158	+ // Add a phasor
	159	+ PhasorMetadataPtr phasor1Metadata = NewSharedPtr<PhasorMetadata>();
	160	+ phasor1Metadata->DeviceAcronym = device1Metadata->Acronym;
	161	+ phasor1Metadata->Label = device1Metadata->Name + " Voltage Phasor";
	162	+ phasor1Metadata->Type = "V"; // Voltage phasor
	163	+ phasor1Metadata->Phase = "+"; // Positive sequence
	164	+ phasor1Metadata->SourceIndex = 1; // Phasor number 1
	165	+ phasor1Metadata->UpdatedOn = timestamp;
	166	+
	167	+ m_phasorMetadata.emplace_back(phasor1Metadata);
	168	+
	169	+ m_metadataVersion++;
	170	+
	171	+ // Pass meta-data to publisher instance for proper conditioning
	172	+ PublisherInstance::DefineMetadata(m_deviceMetadata, m_measurementMetadata, m_phasorMetadata, m_metadataVersion);
173	173	}
174	174
175	175	void PublisherHandler::Start()
176	176	{
177		− static float64_t randMax = float64_t(RAND_MAX);
178		− static const uint64_t interval = 1000;
179		−
180		− const int32_t maxConnections = GetMaximumAllowedConnections();
181		− StatusMessage("\nListening on port: " + ToString(GetPort()) + ", max connections = " + (maxConnections == -1 ? "unlimited" : ToString(maxConnections)) + "...\n");
182		−
183		− // Setup meta-data
184		− DefineMetadata();
185		−
186		− // Setup data publication timer - for this publishing sample we send
187		− // data type reasonable random values every 33 milliseconds
188		− m_publishTimer = NewSharedPtr<Timer>(33, [this](Timer, void)
189		− {
190		− static uint32_t count = m_measurementMetadata.size();
191		− const int64_t timestamp = ToTicks(UtcNow());
192		− vector<MeasurementPtr> measurements;
193		−
194		− measurements.reserve(count);
195		−
196		− // Create new measurement values for publication
197		− for (size_t i = 0; i < count; i++)
198		− {
199		− const MeasurementMetadataPtr metadata = m_measurementMetadata[i];
200		− MeasurementPtr measurement = NewSharedPtr<Measurement>();
201		−
202		− measurement->SignalID = metadata->SignalID;
203		− measurement->Timestamp = timestamp;
204		−
205		− const float64_t randFraction = rand() / randMax;
206		− const float64_t sign = randFraction > 0.5 ? 1.0 : -1.0;
207		− float64_t value;
208		−
209		− switch (metadata->Reference.Kind)
210		− {
211		− case Frequency:
212		− value = 60.0 + sign * randFraction * 0.1;
213		− break;
214		− case DfDt:
215		− value = sign * randFraction * 2;
216		− break;
217		− case Magnitude:
218		− value = 500 + sign * randFraction * 50;
219		− break;
220		− case Angle:
221		− value = sign * randFraction * 180;
222		− break;
223		− default:
224		− value = sign * randFraction * UInt32::MaxValue;
225		− break;
226		− }
227		−
228		− measurement->Value = value;
229		−
230		− measurements.push_back(measurement);
231		− }
232		−
233		− // Publish measurements
234		− PublishMeasurements(measurements);
235		−
236		− // Display a processing message every few seconds
237		− const bool showMessage = m_processCount + count >= (m_processCount / interval + 1) * interval && GetTotalMeasurementsSent() > 0;
238		− m_processCount += count;
239		−
240		− if (showMessage)
241		− StatusMessage(ToString(GetTotalMeasurementsSent()) + " measurements published so far...\n");
242		− },
243		− true);
244		−
245		− // Start data publication
246		− m_publishTimer->Start();
	177	+ static float64_t randMax = float64_t(RAND_MAX);
	178	+ static const uint64_t interval = 1000;
	179	+
	180	+ const int32_t maxConnections = GetMaximumAllowedConnections();
	181	+ StatusMessage("\nListening on port: " + ToString(GetPort()) + ", max connections = " + (maxConnections == -1 ? "unlimited" : ToString(maxConnections)) + "...\n");
	182	+
	183	+ // Setup meta-data
	184	+ DefineMetadata();
	185	+
	186	+ // Setup data publication timer - for this publishing sample we send
	187	+ // data type reasonable random values every 33 milliseconds
	188	+ m_publishTimer = NewSharedPtr<Timer>(33, [this](Timer, void)
	189	+ {
	190	+ static uint32_t count = m_measurementMetadata.size();
	191	+ const int64_t timestamp = ToTicks(UtcNow());
	192	+ vector<MeasurementPtr> measurements;
	193	+
	194	+ measurements.reserve(count);
	195	+
	196	+ // Create new measurement values for publication
	197	+ for (size_t i = 0; i < count; i++)
	198	+ {
	199	+ const MeasurementMetadataPtr metadata = m_measurementMetadata[i];
	200	+ MeasurementPtr measurement = NewSharedPtr<Measurement>();
	201	+
	202	+ measurement->SignalID = metadata->SignalID;
	203	+ measurement->Timestamp = timestamp;
	204	+
	205	+ const float64_t randFraction = rand() / randMax;
	206	+ const float64_t sign = randFraction > 0.5 ? 1.0 : -1.0;
	207	+ float64_t value;
	208	+
	209	+ switch (metadata->Reference.Kind)
	210	+ {
	211	+ case Frequency:
	212	+ value = 60.0 + sign * randFraction * 0.1;
	213	+ break;
	214	+ case DfDt:
	215	+ value = sign * randFraction * 2;
	216	+ break;
	217	+ case Magnitude:
	218	+ value = 500 + sign * randFraction * 50;
	219	+ break;
	220	+ case Angle:
	221	+ value = sign * randFraction * 180;
	222	+ break;
	223	+ default:
	224	+ value = sign * randFraction * UInt32::MaxValue;
	225	+ break;
	226	+ }
	227	+
	228	+ measurement->Value = value;
	229	+
	230	+ measurements.push_back(measurement);
	231	+ }
	232	+
	233	+ // Publish measurements
	234	+ PublishMeasurements(measurements);
	235	+
	236	+ // Display a processing message every few seconds
	237	+ const bool showMessage = m_processCount + count >= (m_processCount / interval + 1) * interval && GetTotalMeasurementsSent() > 0;
	238	+ m_processCount += count;
	239	+
	240	+ if (showMessage)
	241	+ StatusMessage(ToString(GetTotalMeasurementsSent()) + " measurements published so far...\n");
	242	+ },
	243	+ true);
	244	+
	245	+ // Start data publication
	246	+ m_publishTimer->Start();
247	247	}
248	248
249	249	void PublisherHandler::Stop() const
250	250	{
251		− m_publishTimer->Stop();
	251	+ m_publishTimer->Stop();
252	252	}

TimeSeriesPlatformLibrary: Simplfied data publisher threading implementations.

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code