History

TimeSeriesPlatformLibrary: (1) Updated library and samples so that they run on Windows, (2) Overall code cleanup, bug fixes and improvements,

J. Ritchie Carroll committed 8 years ago

Commit 4c65c881 parent 713fa6232 files · +1496 −869Browse files

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

Unified · Split

Source/Libraries/TimeSeriesPlatformLibrary/Samples/AdvancedSubscribe.cpp+57−42
@@ -26,27 +26,26 @@ · 25 unmodified lines
26#include <vector>26#include <vector>
27 27
28#include "../Common/Convert.h"28#include "../Common/Convert.h"
29#include "../Common/Measurement.h"
30#include "../Transport/DataSubscriber.h"29#include "../Transport/DataSubscriber.h"
31 30
32namespace gsfts = GSF::TimeSeries;31using namespace GSF::TimeSeries;
33namespace tst = gsfts::Transport;32using namespace GSF::TimeSeries::Transport;
34 33
35tst::DataSubscriber Subscriber;34DataSubscriber Subscriber;
36tst::SubscriptionInfo Info;35SubscriptionInfo Info;
37 36
38// Create helper objects for subscription.37// Create helper objects for subscription.
39tst::SubscriberConnector CreateSubscriberConnector(std::string hostname, uint16_t port);38SubscriberConnector CreateSubscriberConnector(string hostname, uint16_t port);
40tst::SubscriptionInfo CreateSubscriptionInfo();39SubscriptionInfo CreateSubscriptionInfo();
41 40
42// Handlers for subscriber callbacks.41// Handlers for subscriber callbacks.
43void Resubscribe(tst::DataSubscriber* source);42void Resubscribe(DataSubscriber* source);
44void ProcessMeasurements(tst::DataSubscriber* source, std::vector<gsfts::Measurement> newMeasurements);43void ProcessMeasurements(DataSubscriber* source, vector<Measurement> newMeasurements);
45void DisplayStatusMessage(tst::DataSubscriber* source, std::string message);44void DisplayStatusMessage(DataSubscriber* source, string message);
46void DisplayErrorMessage(tst::DataSubscriber* source, std::string message);45void DisplayErrorMessage(DataSubscriber* source, string message);
47 46
48// Runs the subscriber.47// Runs the subscriber.
49void RunSubscriber(std::string hostname, gsfts::uint16_t port);48void RunSubscriber(string hostname, uint16_t port);
50 49
51// Sample application to demonstrate the more advanced use of the subscriber API.50// Sample application to demonstrate the more advanced use of the subscriber API.
52//51//
@@ -60,31 +59,32 @@ void RunSubscriber(std::string hostname, gsfts::uint16_t port); · 7 unmodified lines
60// Measurements are transmitted via a separate UDP data channel.59// Measurements are transmitted via a separate UDP data channel.
61int main(int argc, char* argv[])60int main(int argc, char* argv[])
62{61{
63 std::string hostname;62 string hostname;
64 gsfts::uint16_t port;63 uint16_t port;
65 64
66 // Ensure that the necessary65 // Ensure that the necessary
67 // command line arguments are given.66 // command line arguments are given.
68 if (argc < 3)67 if (argc < 3)
69 {68 {
70 std::cout << "Usage:" << std::endl;69 cout << "Usage:" << endl;
71 std::cout << " AdvancedSubscribe HOSTNAME PORT" << std::endl;70 cout << " AdvancedSubscribe HOSTNAME PORT" << endl;
72 return 0;71 return 0;
73 }72 }
74 73
75 // Get hostname and port.74 // Get hostname and port.
76 hostname = argv[1];75 hostname = argv[1];
77 std::stringstream(argv[2]) >> port;76 stringstream(argv[2]) >> port;
78 77
79 // Run the subscriber.78 // Run the subscriber.
80 RunSubscriber(hostname, port);79 RunSubscriber(hostname, port);
81 80
82 // Wait until the user presses enter before quitting.81 // Wait until the user presses enter before quitting.
83 std::string line;82 string line;
84 std::getline(std::cin, line);83 getline(cin, line);
85 84
86 // Disconnect the subscriber to stop background threads.85 // Disconnect the subscriber to stop background threads.
87 Subscriber.Disconnect();86 Subscriber.Disconnect();
87 cout << "Disconnected." << endl;
88 88
89 return 0;89 return 0;
90}90}
@@ -94,11 +94,11 @@ int main(int argc, char* argv[]) · 3 unmodified lines
94// - Register callbacks94// - Register callbacks
95// - Connect to publisher95// - Connect to publisher
96// - Subscribe96// - Subscribe
97void RunSubscriber(std::string hostname, gsfts::uint16_t port)97void RunSubscriber(string hostname, uint16_t port)
98{98{
99 // The connector is declared here because it99 // The connector is declared here because it
100 // is only needed for the initial connection100 // is only needed for the initial connection
101 tst::SubscriberConnector connector;101 SubscriberConnector connector;
102 102
103 // Set up helper objects103 // Set up helper objects
104 connector = CreateSubscriberConnector(hostname, port);104 connector = CreateSubscriberConnector(hostname, port);
@@ -108,18 +108,25 @@ void RunSubscriber(std::string hostname, gsfts::uint16_t port) · 3 unmodified lines
108 Subscriber.RegisterStatusMessageCallback(&DisplayStatusMessage);108 Subscriber.RegisterStatusMessageCallback(&DisplayStatusMessage);
109 Subscriber.RegisterErrorMessageCallback(&DisplayErrorMessage);109 Subscriber.RegisterErrorMessageCallback(&DisplayErrorMessage);
110 110
111 cout << endl << "Connecting to " << hostname << ":" << port << "..." << endl << endl;
112
111 // Connect and subscribe to publisher113 // Connect and subscribe to publisher
112 if (connector.Connect(Subscriber))114 if (connector.Connect(Subscriber))
115 {
116 cout << "Connected! Subscribing to data..." << endl << endl;
113 Subscriber.Subscribe(Info);117 Subscriber.Subscribe(Info);
118 }
114 else119 else
115 std::cerr << "All connection attempts failed" << std::endl;120 {
121 cout << "Connection attempts exceeded. Press enter to exit." << endl;
122 }
116}123}
117 124
118tst::SubscriptionInfo CreateSubscriptionInfo()125SubscriptionInfo CreateSubscriptionInfo()
119{126{
120 // SubscriptionInfo is a helper object which allows the user127 // SubscriptionInfo is a helper object which allows the user
121 // to set up their subscription and reuse subscription settings.128 // to set up their subscription and reuse subscription settings.
122 tst::SubscriptionInfo info;129 SubscriptionInfo info;
123 130
124 // The following filter expression formats are also available:131 // The following filter expression formats are also available:
125 //132 //
@@ -162,11 +169,11 @@ tst::SubscriptionInfo CreateSubscriptionInfo() · 36 unmodified lines
162 return info;169 return info;
163}170}
164 171
165tst::SubscriberConnector CreateSubscriberConnector(std::string hostname, gsfts::uint16_t port)172SubscriberConnector CreateSubscriberConnector(string hostname, uint16_t port)
166{173{
167 // SubscriberConnector is another helper object which allows the174 // SubscriberConnector is another helper object which allows the
168 // user to modify settings for auto-reconnects and retry cycles.175 // user to modify settings for auto-reconnects and retry cycles.
169 tst::SubscriberConnector connector;176 SubscriberConnector connector;
170 177
171 connector.RegisterErrorMessageCallback(&DisplayErrorMessage);178 connector.RegisterErrorMessageCallback(&DisplayErrorMessage);
172 connector.RegisterReconnectCallback(&Resubscribe);179 connector.RegisterReconnectCallback(&Resubscribe);
@@ -174,7 +181,7 @@ tst::SubscriberConnector CreateSubscriberConnector(std::string hostname, gsfts:: · 1 unmodified line
174 connector.SetHostname(hostname);181 connector.SetHostname(hostname);
175 connector.SetPort(port);182 connector.SetPort(port);
176 connector.SetMaxRetries(5);183 connector.SetMaxRetries(5);
177 connector.SetRetryInterval(1000);184 connector.SetRetryInterval(1500);
178 connector.SetAutoReconnect(true);185 connector.SetAutoReconnect(true);
179 186
180 return connector;187 return connector;
@@ -182,13 +189,13 @@ tst::SubscriberConnector CreateSubscriberConnector(std::string hostname, gsfts:: · 1 unmodified line
182 189
183// Callback which is called when the subscriber has190// Callback which is called when the subscriber has
184// received a new packet of measurements from the publisher.191// received a new packet of measurements from the publisher.
185void ProcessMeasurements(tst::DataSubscriber* source, std::vector<gsfts::Measurement> newMeasurements)192void ProcessMeasurements(DataSubscriber* source, vector<Measurement> newMeasurements)
186{193{
187 const std::string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f";194 const string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f";
188 const std::size_t MaxTimestampSize = 80;195 const size_t MaxTimestampSize = 80;
189 196
190 static int processCount = 0;197 static int processCount = 0;
191 std::size_t i;198 size_t i;
192 199
193 char timestamp[MaxTimestampSize];200 char timestamp[MaxTimestampSize];
194 201
@@ -196,19 +203,19 @@ void ProcessMeasurements(tst::DataSubscriber* source, std::vector<gsfts::Measure · 1 unmodified line
196 // seconds (assuming 30 calls per second).203 // seconds (assuming 30 calls per second).
197 if (processCount % 150 == 0)204 if (processCount % 150 == 0)
198 {205 {
199 std::cout << Subscriber.GetTotalMeasurementsReceived() << " measurements received so far..." << std::endl;206 cout << Subscriber.GetTotalMeasurementsReceived() << " measurements received so far..." << endl;
200 207
201 if (newMeasurements.size() > 0)208 if (newMeasurements.size() > 0)
202 {209 {
203 if (gsfts::TicksToString(timestamp, MaxTimestampSize, TimestampFormat, newMeasurements[0].Timestamp))210 if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, newMeasurements[0].Timestamp))
204 std::cout << "Timestamp: " << std::string(timestamp) << std::endl;211 cout << "Timestamp: " << string(timestamp) << endl;
205 212
206 std::cout << "Point\tValue" << std::endl;213 cout << "Point\tValue" << endl;
207 214
208 for (i = 0; i < newMeasurements.size(); ++i)215 for (i = 0; i < newMeasurements.size(); ++i)
209 std::cout << newMeasurements[i].ID << '\t' << newMeasurements[i].Value << std::endl;216 cout << newMeasurements[i].ID << '\t' << newMeasurements[i].Value << endl;
210 217
211 std::cout << std::endl;218 cout << endl;
212 }219 }
213 }220 }
214 221
@@ -216,20 +223,28 @@ void ProcessMeasurements(tst::DataSubscriber* source, std::vector<gsfts::Measure · 1 unmodified line
216}223}
217 224
218// Callback that is called when the subscriber auto-reconnects.225// Callback that is called when the subscriber auto-reconnects.
219void Resubscribe(tst::DataSubscriber* source)226void Resubscribe(DataSubscriber* source)
220{227{
221 if (source->IsConnected())228 if (source->IsConnected())
229 {
230 cout << "Reconnected! Subscribing to data..." << endl << endl;
222 source->Subscribe(Info);231 source->Subscribe(Info);
232 }
233 else
234 {
235 source->Disconnect();
236 cout << "Connection retry attempts exceeded. Press enter to exit." << endl;
237 }
223}238}
224 239
225// Callback which is called to display status messages from the subscriber.240// Callback which is called to display status messages from the subscriber.
226void DisplayStatusMessage(tst::DataSubscriber* source, std::string message)241void DisplayStatusMessage(DataSubscriber* source, string message)
227{242{
228 std::cout << message << std::endl << std::endl;243 cout << message << endl << endl;
229}244}
230 245
231// Callback which is called to display error messages from the connector and subscriber.246// Callback which is called to display error messages from the connector and subscriber.
232void DisplayErrorMessage(tst::DataSubscriber* source, std::string message)247void DisplayErrorMessage(DataSubscriber* source, string message)
233{248{
234 std::cerr << message << std::endl << std::endl;249 cerr << message << endl << endl;
235}250}

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