History

TimeSeriesPlatformLibrary: Added user data property for publisher instance and performed some subscriber code cleanup

J. Ritchie Carroll committed 7 years ago

Commit 7344d5e1 parent 076441c8 files · +73 −33Browse files

Showing the changes to Source/Libraries/TimeSeriesPlatformLibrary/Samples (3 of 8 files). Show all

Unified · Split

3 files
  1. Source/Libraries/TimeSeriesPlatformLibrary/Samples/InstanceSubscribe.cpp +7 −7
  2. Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberHandler.cpp +29 −8
  3. Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberHandler.h +15 −12
Source/Libraries/TimeSeriesPlatformLibrary/Samples/InstanceSubscribe.cpp+7−7
@@ -22,15 +22,15 @@ · 21 unmodified lines
22//******************************************************************************************************22//******************************************************************************************************
23 23
24#include <iostream>24#include <iostream>
25
26#include "../Common/Convert.h"
27#include "SubscriberHandler.h"25#include "SubscriberHandler.h"
26#include "../Common/Convert.h"
28 27
29using namespace std;28using namespace std;
30using namespace GSF;29using namespace GSF;
30using namespace GSF::TimeSeries;
31using namespace GSF::TimeSeries::Transport;
31 32
32#define TotalInstances 333#define TotalInstances 3
33
34SubscriberHandler* Subscriber[TotalInstances];34SubscriberHandler* Subscriber[TotalInstances];
35 35
36int main(int argc, char* argv[])36int main(int argc, char* argv[])
@@ -55,7 +55,6 @@ int main(int argc, char* argv[]) · 18 unmodified lines
55 for (uint32_t i = 0; i < TotalInstances; i++)55 for (uint32_t i = 0; i < TotalInstances; i++)
56 {56 {
57 SubscriberHandler* subscriber = new SubscriberHandler("Subscriber " + ToString(i + 1));57 SubscriberHandler* subscriber = new SubscriberHandler("Subscriber " + ToString(i + 1));
58
59 subscriber->Initialize(hostname, port);58 subscriber->Initialize(hostname, port);
60 59
61 switch (i)60 switch (i)
@@ -87,13 +86,14 @@ int main(int argc, char* argv[]) · 25 unmodified lines
87 86
88 // Shutdown subscriber instances87 // Shutdown subscriber instances
89 for (uint32_t i = 0; i < TotalInstances; i++)88 for (uint32_t i = 0; i < TotalInstances; i++)
90 {
91 Subscriber[i]->Disconnect();89 Subscriber[i]->Disconnect();
92 delete Subscriber[i];
93 }
94 90
95 // Disconnect the subscriber to stop background threads.91 // Disconnect the subscriber to stop background threads.
96 cout << "Disconnected." << endl;92 cout << "Disconnected." << endl;
97 93
94 // Delete subscriber instances
95 for (uint32_t i = 0; i < TotalInstances; i++)
96 delete Subscriber[i];
97
98 return 0;98 return 0;
99}99}
Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberHandler.cpp+29−8
@@ -31,12 +31,18 @@ · 30 unmodified lines
31using namespace std;31using namespace std;
32using namespace GSF;32using namespace GSF;
33using namespace GSF::TimeSeries;33using namespace GSF::TimeSeries;
34using namespace GSF::TimeSeries::Transport;
34 35
35SubscriberHandler::SubscriberHandler(const string& name) :36SubscriberHandler::SubscriberHandler(string name) :
36 m_name(name)37 m_name(std::move(name))
37{38{
38}39}
39 40
41SubscriberHandler::~SubscriberHandler()
42{
43 cout << "Calling " << m_name << " destructor...";
44}
45
40SubscriptionInfo SubscriberHandler::CreateSubscriptionInfo()46SubscriptionInfo SubscriberHandler::CreateSubscriptionInfo()
41{47{
42 SubscriptionInfo info = SubscriberInstance::CreateSubscriptionInfo();48 SubscriptionInfo info = SubscriberInstance::CreateSubscriptionInfo();
@@ -63,6 +69,12 @@ SubscriptionInfo SubscriberHandler::CreateSubscriptionInfo() · 20 unmodified lines
63 return info;69 return info;
64}70}
65 71
72void SubscriberHandler::SetupSubscriberConnector(SubscriberConnector& connector)
73{
74 // TODO: Modify connector properties as desired...
75 //connector.SetMaxRetries(-1);
76}
77
66void SubscriberHandler::StatusMessage(const string& message)78void SubscriberHandler::StatusMessage(const string& message)
67{79{
68 // TODO: Make sure these messages get logged to an appropriate location80 // TODO: Make sure these messages get logged to an appropriate location
@@ -71,7 +83,9 @@ void SubscriberHandler::StatusMessage(const string& message) · 2 unmodified lines
71 83
72 status << "[" << m_name << "] " << message;84 status << "[" << m_name << "] " << message;
73 85
86 m_coutLock.lock();
74 SubscriberInstance::StatusMessage(status.str());87 SubscriberInstance::StatusMessage(status.str());
88 m_coutLock.unlock();
75}89}
76 90
77void SubscriberHandler::ErrorMessage(const string& message)91void SubscriberHandler::ErrorMessage(const string& message)
@@ -82,12 +96,19 @@ void SubscriberHandler::ErrorMessage(const string& message) · 4 unmodified lines
82 96
83 status << "[" << m_name << "] " << message;97 status << "[" << m_name << "] " << message;
84 98
99 m_coutLock.lock();
85 SubscriberInstance::ErrorMessage(status.str());100 SubscriberInstance::ErrorMessage(status.str());
101 m_coutLock.unlock();
86}102}
87 103
88void SubscriberHandler::DataStartTime(time_t unixSOC, uint16_t milliseconds)104void SubscriberHandler::DataStartTime(time_t unixSOC, uint16_t milliseconds)
89{105{
90 // TODO: This reports timestamp of very first received measurement (if useful)106 // TODO: This reports timestamp, time_t format, of very first received measurement (if useful)
107}
108
109void SubscriberHandler::DataStartTime(DateTime startTime)
110{
111 // TODO: This reports timestamp, ptime format, of very first received measurement (if useful)
91}112}
92 113
93void SubscriberHandler::ReceivedMetadata(const vector<uint8_t>& payload)114void SubscriberHandler::ReceivedMetadata(const vector<uint8_t>& payload)
@@ -167,16 +188,16 @@ void SubscriberHandler::ReceivedNewMeasurements(const vector<MeasurementPtr>& me · 73 unmodified lines
167 message << GetTotalMeasurementsReceived() << " measurements received so far..." << endl;188 message << GetTotalMeasurementsReceived() << " measurements received so far..." << endl;
168 189
169 if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp))190 if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp))
170 message << string(timestamp);191 message << string(timestamp) << endl;
171 192
172 message << "Signal ID: " << boost::lexical_cast<string>(measurements[0]->SignalID) << endl;193 message << "Signal ID: " << boost::lexical_cast<string>(measurements[0]->SignalID) << endl;
173 194
174 message << "Point\tValue" << endl;195 message << "\tPoint\tValue" << endl;
175 196
176 for (const auto& measurement : measurements)197 for (const auto& measurement : measurements)
177 message << '\t' << measurement->ID << '\t' << measurement->Value;198 message << '\t' << measurement->ID << '\t' << measurement->Value << endl;
178 199
179 message << '\t' << GetTotalMeasurementsReceived() << endl;200 message << "Total measurements: " << GetTotalMeasurementsReceived() << endl;
180 201
181 StatusMessage(message.str());202 StatusMessage(message.str());
182 }203 }
@@ -200,4 +221,4 @@ void SubscriberHandler::ConnectionEstablished() · 17 unmodified lines
200void SubscriberHandler::ConnectionTerminated()221void SubscriberHandler::ConnectionTerminated()
201{222{
202 StatusMessage("Connection terminated.");223 StatusMessage("Connection terminated.");
203}224}
Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberHandler.h+15−12
@@ -24,24 +24,24 @@ · 23 unmodified lines
24#ifndef __SUBSCRIBER_HANDLER_H24#ifndef __SUBSCRIBER_HANDLER_H
25#define __SUBSCRIBER_HANDLER_H25#define __SUBSCRIBER_HANDLER_H
26 26
27#include "../Common/CommonTypes.h"
27#include "../Transport/SubscriberInstance.h"28#include "../Transport/SubscriberInstance.h"
28 29
29using namespace std;30class SubscriberHandler : public GSF::TimeSeries::Transport::SubscriberInstance
30using namespace GSF::TimeSeries;
31using namespace GSF::TimeSeries::Transport;
32
33class SubscriberHandler : public SubscriberInstance
34{31{
35private:32private:
36 string m_name;33 std::string m_name;
34 GSF::Mutex m_coutLock;
37 35
38protected:36protected:
39 SubscriptionInfo CreateSubscriptionInfo() override;37 GSF::TimeSeries::Transport::SubscriptionInfo CreateSubscriptionInfo() override;
40 void StatusMessage(const string& message) override;38 void SetupSubscriberConnector(GSF::TimeSeries::Transport::SubscriberConnector& connector) override;
41 void ErrorMessage(const string& message) override;39 void StatusMessage(const std::string& message) override;
40 void ErrorMessage(const std::string& message) override;
42 void DataStartTime(time_t unixSOC, uint16_t milliseconds) override;41 void DataStartTime(time_t unixSOC, uint16_t milliseconds) override;
43 void ReceivedMetadata(const vector<uint8_t>& payload) override;42 void DataStartTime(GSF::DateTime startTime) override;
44 void ReceivedNewMeasurements(const vector<MeasurementPtr>& measurements) override;43 void ReceivedMetadata(const std::vector<uint8_t>& payload) override;
44 void ReceivedNewMeasurements(const std::vector<GSF::TimeSeries::MeasurementPtr>& measurements) override;
45 void ParsedMetadata() override;45 void ParsedMetadata() override;
46 void ConfigurationChanged() override;46 void ConfigurationChanged() override;
47 void HistoricalReadComplete() override;47 void HistoricalReadComplete() override;
@@ -49,7 +49,10 @@ protected: · 1 unmodified line
49 void ConnectionTerminated() override;49 void ConnectionTerminated() override;
50 50
51public:51public:
52 SubscriberHandler(const string& name);52 SubscriberHandler(std::string name);
53 ~SubscriberHandler();
53};54};
54 55
56typedef GSF::SharedPtr<SubscriberHandler> SubscriberHandlerPtr;
57
55#endif58#endif

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