Showing the changes to Source/Libraries/TimeSeriesPlatformLibrary/Samples/TemporalSubscriber.cpp (1 of 15 files). Show all
Unified · Split
| @@ -32,9 +32,11 @@ using namespace GSF::TimeSeries; · 31 unmodified lines | |||
| 32 | GSF::Data::DataSetPtr TemporalSubscriber::s_historyDataSet = nullptr; | 32 | GSF::Data::DataSetPtr TemporalSubscriber::s_historyDataSet = nullptr; |
| 33 | GSF::Data::DataTablePtr TemporalSubscriber::s_history = nullptr; | 33 | GSF::Data::DataTablePtr TemporalSubscriber::s_history = nullptr; |
| 34 | 34 | ||
| 35 | TemporalSubscriber::TemporalSubscriber(TemporalSubscriberConnectionPtr connection) : | 35 | TemporalSubscriber::TemporalSubscriber(SubscriberConnectionPtr connection) : |
| 36 | m_connection(std::move(connection)), | 36 | m_connection(std::move(connection)), |
| 37 | m_currentTimestamp(m_connection->GetStartTicks()), | 37 | m_startTimestamp(ToTicks(m_connection->GetStartTimeConstraint())), |
| 38 | m_stopTimestamp(ToTicks(m_connection->GetStopTimeConstraint())), | ||
| 39 | m_currentTimestamp(m_startTimestamp), | ||
| 38 | m_currentRow(0), | 40 | m_currentRow(0), |
| 39 | m_stopped(false) | 41 | m_stopped(false) |
| 40 | { | 42 | { |
| @@ -57,20 +59,22 @@ TemporalSubscriber::TemporalSubscriber(TemporalSubscriberConnectionPtr connectio · 16 unmodified lines | |||
| 57 | TemporalSubscriber::~TemporalSubscriber() | 59 | TemporalSubscriber::~TemporalSubscriber() |
| 58 | { | 60 | { |
| 59 | CompleteTemporalSubscription(); | 61 | CompleteTemporalSubscription(); |
| 62 | m_processTimer.reset(); | ||
| 60 | } | 63 | } |
| 61 | 64 | ||
| 62 | void TemporalSubscriber::SetProcessingInterval(int32_t processingInterval) const | 65 | void TemporalSubscriber::SetProcessingInterval(int32_t processingInterval) const |
| 63 | { | 66 | { |
| 64 | if (processingInterval == -1) | 67 | if (processingInterval == -1) |
| 65 | processingInterval = 33; | 68 | processingInterval = 33; |
| 66 | else if (processingInterval == 0) | ||
| 67 | processingInterval = 1; | ||
| 68 | 69 | ||
| 69 | m_processTimer->SetInterval(processingInterval); | 70 | m_processTimer->SetInterval(processingInterval); |
| 70 | } | 71 | } |
| 71 | 72 | ||
| 72 | void TemporalSubscriber::SendTemporalData() | 73 | void TemporalSubscriber::SendTemporalData() |
| 73 | { | 74 | { |
| 75 | if (m_stopped) | ||
| 76 | return; | ||
| 77 | |||
| 74 | static DataTable& history = *s_history; | 78 | static DataTable& history = *s_history; |
| 75 | static const int32_t signalIDColumn = history["SignalID"]->Index(); | 79 | static const int32_t signalIDColumn = history["SignalID"]->Index(); |
| 76 | static const int32_t timestampColumn = history["Timestamp"]->Index(); | 80 | static const int32_t timestampColumn = history["Timestamp"]->Index(); |
| @@ -103,7 +107,7 @@ void TemporalSubscriber::SendTemporalData() · 26 unmodified lines | |||
| 103 | // Setup next publication timestamp | 107 | // Setup next publication timestamp |
| 104 | m_currentTimestamp += HistoryInterval; | 108 | m_currentTimestamp += HistoryInterval; |
| 105 | 109 | ||
| 106 | if (m_currentTimestamp > m_connection->GetStopTicks()) | 110 | if (m_currentTimestamp > m_stopTimestamp) |
| 107 | CompleteTemporalSubscription(); | 111 | CompleteTemporalSubscription(); |
| 108 | } | 112 | } |
| 109 | 113 | ||
| @@ -112,7 +116,7 @@ void TemporalSubscriber::CompleteTemporalSubscription() · 2 unmodified lines | |||
| 112 | if (m_stopped) | 116 | if (m_stopped) |
| 113 | return; | 117 | return; |
| 114 | 118 | ||
| 115 | m_stopped = true; | ||
| 116 | m_processTimer->Stop(); | 119 | m_processTimer->Stop(); |
| 117 | m_connection->CompleteTemporalSubscription(); | 120 | m_connection->CompleteTemporalSubscription(); |
| 121 | m_stopped = true; | ||
| 118 | } | 122 | } |