History

TimeSeriesPlatformLibrary: Simplified temporal subscription connection processing.

J. Ritchie Carroll committed 7 years ago

Commit eef23ad1 parent ee9de0113 files · +470 −127Browse files

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

Unified · Split

Source/Libraries/TimeSeriesPlatformLibrary/Samples/AdvancedPublish.cpp+22−16
@@ -48,8 +48,9 @@ void DisplayClientConnected(DataPublisher* source, const SubscriberConnectionPtr · 47 unmodified lines
48void DisplayClientDisconnected(DataPublisher* source, const SubscriberConnectionPtr& connection);48void DisplayClientDisconnected(DataPublisher* source, const SubscriberConnectionPtr& connection);
49void DisplayStatusMessage(DataPublisher* source, const string& message);49void DisplayStatusMessage(DataPublisher* source, const string& message);
50void DisplayErrorMessage(DataPublisher* source, const string& message);50void DisplayErrorMessage(DataPublisher* source, const string& message);
51void HandleTemporalSubscriptionRequested(DataPublisher* source, const SubscriberConnectionPtr& connection);51void HandleTemporalSubscriptionRequested(DataPublisher* source, const TemporalSubscriberConnectionPtr& connection);
52void HandleProcessingIntervalChangeRequested(DataPublisher* source, const SubscriberConnectionPtr& connection);52void HandleTemporalProcessingIntervalChangeRequested(DataPublisher* source, const TemporalSubscriberConnectionPtr& connection);
53void HandleTemporalSubscriptionCanceled(DataPublisher* source, const TemporalSubscriberConnectionPtr& connection);
53 54
54void LoadMetadataToPublish(vector<DeviceMetadataPtr>& deviceMetadata, vector<MeasurementMetadataPtr>& measurementMetadata, vector<PhasorMetadataPtr>& phasorMetadata)55void LoadMetadataToPublish(vector<DeviceMetadataPtr>& deviceMetadata, vector<MeasurementMetadataPtr>& measurementMetadata, vector<PhasorMetadataPtr>& phasorMetadata)
55{56{
@@ -222,7 +223,8 @@ bool RunPublisher(uint16_t port, bool genHistory) · 166 unmodified lines
222 Publisher->RegisterStatusMessageCallback(&DisplayStatusMessage);223 Publisher->RegisterStatusMessageCallback(&DisplayStatusMessage);
223 Publisher->RegisterErrorMessageCallback(&DisplayErrorMessage);224 Publisher->RegisterErrorMessageCallback(&DisplayErrorMessage);
224 Publisher->RegisterTemporalSubscriptionRequestedCallback(&HandleTemporalSubscriptionRequested);225 Publisher->RegisterTemporalSubscriptionRequestedCallback(&HandleTemporalSubscriptionRequested);
225 Publisher->RegisterProcessingIntervalChangeRequestedCallback(&HandleProcessingIntervalChangeRequested);226 Publisher->RegisterTemporalProcessingIntervalChangeRequestedCallback(&HandleTemporalProcessingIntervalChangeRequested);
227 Publisher->RegisterTemporalSubscriptionCanceledCallback(&HandleTemporalSubscriptionCanceled);
226 228
227 // Enable temporal subscription support - this allows historical data requests as well as real-time229 // Enable temporal subscription support - this allows historical data requests as well as real-time
228 Publisher->SetSupportsTemporalSubscriptions(true);230 Publisher->SetSupportsTemporalSubscriptions(true);
@@ -333,28 +335,18 @@ void DisplayErrorMessage(DataPublisher* source, const string& message) · 104 unmodified lines
333 cerr << message << endl << endl;335 cerr << message << endl << endl;
334}336}
335 337
336void HandleTemporalSubscriptionRequested(DataPublisher* source, const SubscriberConnectionPtr& connection)338void HandleTemporalSubscriptionRequested(DataPublisher* source, const TemporalSubscriberConnectionPtr& connection)
337{339{
338 cout << "Client \"" << connection->GetConnectionID() << "\" with subscriber ID " << ToString(connection->GetSubscriberID()) << " has requested a temporal subscription starting at " << ToString(connection->GetStartTimeConstraint()) << endl << endl;340 cout << "Client \"" << connection->GetConnectionID() << "\" with subscriber ID " << ToString(connection->GetSubscriberID()) << " has requested a temporal subscription starting at " << ToString(connection->GetStartTimeConstraint()) << endl << endl;
339 341
340 TemporalSubscriberPtr temporalSubscription = NewSharedPtr<TemporalSubscriber>(connection, [](const GSF::Guid& instanceID)342 TemporalSubscriberPtr temporalSubscription = NewSharedPtr<TemporalSubscriber>(connection);
341 {
342 Thread([instanceID]{
343 TemporalSubscriptionsLock.lock();
344
345 if (TemporalSubscriptions.find(instanceID) != TemporalSubscriptions.end())
346 TemporalSubscriptions.erase(instanceID);
347
348 TemporalSubscriptionsLock.unlock();
349 });
350 });
351 343
352 TemporalSubscriptionsLock.lock();344 TemporalSubscriptionsLock.lock();
353 TemporalSubscriptions.insert_or_assign(connection->GetInstanceID(), temporalSubscription);345 TemporalSubscriptions.insert_or_assign(connection->GetInstanceID(), temporalSubscription);
354 TemporalSubscriptionsLock.unlock();346 TemporalSubscriptionsLock.unlock();
355}347}
356 348
357void HandleProcessingIntervalChangeRequested(DataPublisher* source, const SubscriberConnectionPtr& connection)349void HandleTemporalProcessingIntervalChangeRequested(DataPublisher* source, const TemporalSubscriberConnectionPtr& connection)
358{350{
359 static const TemporalSubscriberPtr nullTemporalSubscription = nullptr;351 static const TemporalSubscriberPtr nullTemporalSubscription = nullptr;
360 const GSF::Guid& instanceID = connection->GetInstanceID();352 const GSF::Guid& instanceID = connection->GetInstanceID();
@@ -371,3 +363,17 @@ void HandleProcessingIntervalChangeRequested(DataPublisher* source, const Subscr · 10 unmodified lines
371 if (temporalSubscription != nullTemporalSubscription)363 if (temporalSubscription != nullTemporalSubscription)
372 cout << "Client \"" << connection->GetConnectionID() << "\" with subscriber ID " << ToString(connection->GetSubscriberID()) << " has requested to change its temporal processing interval to " << ToString(processingInterval) << "ms" << endl << endl;364 cout << "Client \"" << connection->GetConnectionID() << "\" with subscriber ID " << ToString(connection->GetSubscriberID()) << " has requested to change its temporal processing interval to " << ToString(processingInterval) << "ms" << endl << endl;
373}365}
366
367void HandleTemporalSubscriptionCanceled(DataPublisher* source, const TemporalSubscriberConnectionPtr& connection)
368{
369 const GSF::Guid& instanceID = connection->GetInstanceID();
370
371 Thread([instanceID]{
372 TemporalSubscriptionsLock.lock();
373
374 if (TemporalSubscriptions.find(instanceID) != TemporalSubscriptions.end())
375 TemporalSubscriptions.erase(instanceID);
376
377 TemporalSubscriptionsLock.unlock();
378 });
379}

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t