History

TimeSeriesPlatformLibrary: Updated SimplePublish sample to properly run.

J. Ritchie Carroll committed 7 years ago

Commit 4d190cd1 parent 285fcea7 files · +108 −99Browse files

Unified · Split

7 files
  1. Source/Applications/TimeSeries Platform Library Samples/SimplePublish/SimplePublish.vcxproj +2 −2
  2. Source/Libraries/TimeSeriesPlatformLibrary/Common/CommonTypes.h +1 −0
  3. Source/Libraries/TimeSeriesPlatformLibrary/Common/Convert.cpp +7 −0
  4. Source/Libraries/TimeSeriesPlatformLibrary/Common/Convert.h +2 −0
  5. Source/Libraries/TimeSeriesPlatformLibrary/Samples/SimplePublish.cpp +45 −56
  6. Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataPublisher.cpp +48 −41
  7. Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataPublisher.h +3 −0
Source/Applications/TimeSeries Platform Library Samples/SimplePublish/SimplePublish.vcxproj+2−2
@@ -61,7 +61,7 @@ · 60 unmodified lines
6161 </PrecompiledHeader>
6262 <WarningLevel>Level3</WarningLevel>
6363 <Optimization>Disabled</Optimization>
64− <PreprocessorDefinitions>_WIN32_WINNT=0x0601;WIN32;BOOST_CONFIG_SUPPRESS_OUTDATED_MESSAGE;_DEBUG;_CONSOLE;%(PreprocessorDefinitions)</PreprocessorDefinitions>
64+ <PreprocessorDefinitions>_WIN32_WINNT=0x0601;WIN32;BOOST_CONFIG_SUPPRESS_OUTDATED_MESSAGE;_SILENCE_FPOS_SEEKPOS_DEPRECATION_WARNING;_DEBUG;_CONSOLE;%(PreprocessorDefinitions)</PreprocessorDefinitions>
6565 <SDLCheck>true</SDLCheck>
6666 <AdditionalIncludeDirectories>$(SolutionDir)\..\..\boost</AdditionalIncludeDirectories>
6767 </ClCompile>
@@ -79,7 +79,7 @@ · 11 unmodified lines
7979 <Optimization>MaxSpeed</Optimization>
8080 <FunctionLevelLinking>true</FunctionLevelLinking>
8181 <IntrinsicFunctions>true</IntrinsicFunctions>
82− <PreprocessorDefinitions>_WIN32_WINNT=0x0601;WIN32;BOOST_CONFIG_SUPPRESS_OUTDATED_MESSAGE;NDEBUG;_CONSOLE;%(PreprocessorDefinitions)</PreprocessorDefinitions>
82+ <PreprocessorDefinitions>_WIN32_WINNT=0x0601;WIN32;BOOST_CONFIG_SUPPRESS_OUTDATED_MESSAGE;_SILENCE_FPOS_SEEKPOS_DEPRECATION_WARNING;NDEBUG;_CONSOLE;%(PreprocessorDefinitions)</PreprocessorDefinitions>
8383 <SDLCheck>true</SDLCheck>
8484 <AdditionalIncludeDirectories>$(SolutionDir)\..\..\boost</AdditionalIncludeDirectories>
8585 </ClCompile>
Source/Libraries/TimeSeriesPlatformLibrary/Common/CommonTypes.h+1−0
@@ -206,6 +206,7 @@ namespace GSF · 205 unmodified lines
206206 typedef boost::any Object;
207207 typedef boost::uuids::uuid Guid;
208208 typedef boost::posix_time::ptime DateTime;
209+ typedef boost::posix_time::time_duration TimeSpan;
209210 typedef boost::system::error_code ErrorCode;
210211 typedef boost::system::system_error SystemError;
211212 typedef boost::exception Exception;
Source/Libraries/TimeSeriesPlatformLibrary/Common/Convert.cpp+7−0
@@ -254,6 +254,13 @@ std::string GSF::ToString(const DateTime& value, const char* format) · 253 unmodified lines
254254 return stream.str();
255255 }
256256
257+std::string GSF::ToString(const TimeSpan& value)
258+{
259+ // TODO: Consider improving elapsed time string with hours, minutes, etc.
260+ const float64_t seconds = value.total_milliseconds() / 1000.0;
261+ return ToString(seconds) + " seconds";
262+}
263+
257264 wstring GSF::ToUTF16(const string& value)
258265 {
259266 wstring_convert<codecvt_utf8<wchar_t>, wchar_t> converter;
Source/Libraries/TimeSeriesPlatformLibrary/Common/Convert.h+2−0
@@ -55,6 +55,8 @@ namespace GSF · 54 unmodified lines
5555
5656 std::string ToString(const DateTime& value, const char* format = "%Y-%m-%d %H:%M:%S%F");
5757
58+ std::string ToString(const TimeSpan& value);
59+
5860 std::wstring ToUTF16(const std::string& value);
5961 std::string ToUTF8(const std::wstring& value);
6062
Source/Libraries/TimeSeriesPlatformLibrary/Samples/SimplePublish.cpp+45−56
@@ -23,9 +23,7 @@ · 22 unmodified lines
2323
2424 #include <iostream>
2525 #include <string>
26−#include <vector>
2726
28−#include "../Common/Convert.h"
2927 #include "../Transport/DataPublisher.h"
3028
3129 using namespace std;
@@ -33,7 +31,7 @@ using namespace GSF; · 1 unmodified line
3331 using namespace GSF::TimeSeries;
3432 using namespace GSF::TimeSeries::Transport;
3533
36−DataPublisher Subscriber;
34+DataPublisherPtr Publisher;
3735
3836 bool RunPublisher(uint16_t port);
3937 //void ProcessMeasurements(DataPublisher* source, const vector<MeasurementPtr>& measurements);
@@ -86,27 +84,15 @@ int main(int argc, char* argv[]) · 46 unmodified lines
8684 // - Publish
8785 bool RunPublisher(uint16_t port)
8886 {
89− // SubscriptionInfo is a helper object which allows the user
90− // to set up their subscription and reuse subscription settings.
91− //SubscriptionInfo info;
92− //info.FilterExpression = "PPA:1;PPA:2;PPA:3;PPA:4;PPA:5;PPA:6;PPA:7;PPA:8;PPA:9;PPA:10;PPA:11;PPA:12;PPA:13;PPA:14";
93−
94− // Register callbacks
95− Subscriber.RegisterStatusMessageCallback(&DisplayStatusMessage);
96− Subscriber.RegisterErrorMessageCallback(&DisplayErrorMessage);
97− //Subscriber.RegisterNewMeasurementsCallback(&ProcessMeasurements);
98−
99− cout << endl << "Listening on port: " << port << "..." << endl << endl;
100−
10187 string errorMessage;
10288 bool connected = false;
10389
10490 try
10591 {
106− //Subscriber.Connect(hostname, port);
92+ Publisher = NewSharedPtr<DataPublisher>(port);
10793 connected = true;
10894 }
109− catch (SubscriberException& ex)
95+ catch (PublisherException& ex)
11096 {
11197 errorMessage = ex.what();
11298 }
@@ -116,57 +102,60 @@ bool RunPublisher(uint16_t port) · 3 unmodified lines
116102 }
117103 catch (...)
118104 {
119− errorMessage = current_exception_diagnostic_information(true);
105+ errorMessage = boost::current_exception_diagnostic_information(true);
120106 }
121107
122108 if (connected)
123109 {
124− cout << "Connected! Subscribing to data..." << endl << endl;
125− Subscriber.Subscribe(info);
110+ cout << endl << "Listening on port: " << port << "..." << endl << endl;
111+
112+ // Register callbacks
113+ Publisher->RegisterStatusMessageCallback(&DisplayStatusMessage);
114+ Publisher->RegisterErrorMessageCallback(&DisplayErrorMessage);
126115 }
127116 else
128117 {
129− cerr << "Failed to connect to \"" << hostname << ":" << port << "\": " << errorMessage;
118+ cerr << "Failed to listen on port: " << port << ": " << errorMessage;
130119 }
131120
132121 return connected;
133122 }
134123
135−// Callback which is called when the subscriber has
136−// received a new packet of measurements from the publisher.
137−void ProcessMeasurements(DataPublisher* source, const vector<MeasurementPtr>& measurements)
138−{
139− const string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f";
140− const uint32_t MaxTimestampSize = 80;
141−
142− static long processCount = 0;
143− static char timestamp[MaxTimestampSize];
144− static const long interval = 5 * 60;
145− const long measurementCount = measurements.size();
146− const bool showMessage = (processCount + measurementCount >= (processCount / interval + 1) * interval);
147−
148− processCount += measurementCount;
149−
150− // Only display messages every few seconds
151− if (showMessage)
152− {
153− stringstream message;
154−
155− message << source->GetTotalMeasurementsReceived() << " measurements received so far..." << endl;
156−
157− if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp))
158− message << "Timestamp: " << string(timestamp) << endl;
159−
160− message << "Point\tValue" << endl;
161−
162− for (const auto& measurement : measurements)
163− message << measurement->ID << '\t' << measurement->Value << endl;
164−
165− message << endl;
166−
167− cout << message.str();
168− }
169−}
124+//// Callback which is called when the subscriber has
125+//// received a new packet of measurements from the publisher.
126+//void ProcessMeasurements(DataPublisher* source, const vector<MeasurementPtr>& measurements)
127+//{
128+// const string TimestampFormat = "%Y-%m-%d %H:%M:%S.%f";
129+// const uint32_t MaxTimestampSize = 80;
130+//
131+// static long processCount = 0;
132+// static char timestamp[MaxTimestampSize];
133+// static const long interval = 5 * 60;
134+// const long measurementCount = measurements.size();
135+// const bool showMessage = (processCount + measurementCount >= (processCount / interval + 1) * interval);
136+//
137+// processCount += measurementCount;
138+//
139+// // Only display messages every few seconds
140+// if (showMessage)
141+// {
142+// stringstream message;
143+//
144+// message << source->GetTotalMeasurementsReceived() << " measurements received so far..." << endl;
145+//
146+// if (TicksToString(timestamp, MaxTimestampSize, TimestampFormat, measurements[0]->Timestamp))
147+// message << "Timestamp: " << string(timestamp) << endl;
148+//
149+// message << "Point\tValue" << endl;
150+//
151+// for (const auto& measurement : measurements)
152+// message << measurement->ID << '\t' << measurement->Value << endl;
153+//
154+// message << endl;
155+//
156+// cout << message.str();
157+// }
158+//}
170159
171160 // Callback which is called to display status messages from the subscriber.
172161 void DisplayStatusMessage(DataPublisher* source, const string& message)
Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataPublisher.cpp+48−41
@@ -67,10 +67,11 @@ struct ClientConnectedInfo · 66 unmodified lines
6767
6868 ClientConnection::ClientConnection(DataPublisherPtr parent, IOContext& commandChannelService, IOContext& dataChannelService) :
6969 m_parent(std::move(parent)),
70+ m_commandChannelService(commandChannelService),
7071 m_subscriberID(NewGuid()),
7172 m_operationalModes(OperationalModes::NoFlags),
7273 m_encoding(OperationalEncoding::UTF8),
73− m_commandChannelSocket(commandChannelService),
74+ m_commandChannelSocket(m_commandChannelService),
7475 m_udpPort(0),
7576 m_dataChannelSocket(dataChannelService),
7677 m_timeIndex(0),
@@ -81,42 +82,6 @@ ClientConnection::ClientConnection(DataPublisherPtr parent, IOContext& commandCh · 4 unmodified lines
8182 m_pingTimer.SetAutoReset(true);
8283 m_pingTimer.SetCallback(&ClientConnection::PingTimerElapsed);
8384 m_pingTimer.SetUserData(this);
84−
85− // Attempt to lookup remote connection identification for logging purposes
86− auto remoteEndPoint = m_commandChannelSocket.remote_endpoint();
87− m_ipAddress = remoteEndPoint.address();
88−
89− if (remoteEndPoint.protocol() == tcp::v6())
90− m_connectionID = "[" + m_ipAddress.to_string() + "]:" + ToString(remoteEndPoint.port());
91− else
92− m_connectionID = m_ipAddress.to_string() + ":" + ToString(remoteEndPoint.port());
93−
94− try
95− {
96− DnsResolver resolver(commandChannelService);
97− const DnsResolver::query query(m_ipAddress.to_string(), ToString(remoteEndPoint.port()));
98− DnsResolver::iterator iterator = resolver.resolve(query);
99− const DnsResolver::iterator end;
100−
101− while (iterator != end)
102− {
103− auto endPoint = *iterator++;
104−
105− if (!endPoint.host_name().empty())
106− {
107− m_hostName = endPoint.host_name();
108− m_connectionID = m_hostName + " (" + m_connectionID + ")";
109− break;
110− }
111− }
112− }
113− catch (...)
114− { //-V565
115− // DNS lookup failure is not a catastrophe
116− }
117−
118− if (m_hostName.empty())
119− m_hostName = m_ipAddress.to_string();
12085 }
12186
12287 ClientConnection::~ClientConnection() = default;
@@ -190,7 +155,43 @@ std::vector<uint8_t> ClientConnection::IVs(int cipherIndex) · 67 unmodified lines
190155
191156 void ClientConnection::Start()
192157 {
193−
158+ // Attempt to lookup remote connection identification for logging purposes
159+ auto remoteEndPoint = m_commandChannelSocket.remote_endpoint();
160+ m_ipAddress = remoteEndPoint.address();
161+
162+ if (remoteEndPoint.protocol() == tcp::v6())
163+ m_connectionID = "[" + m_ipAddress.to_string() + "]:" + ToString(remoteEndPoint.port());
164+ else
165+ m_connectionID = m_ipAddress.to_string() + ":" + ToString(remoteEndPoint.port());
166+
167+ try
168+ {
169+ DnsResolver resolver(m_commandChannelService);
170+ const DnsResolver::query query(m_ipAddress.to_string(), ToString(remoteEndPoint.port()));
171+ DnsResolver::iterator iterator = resolver.resolve(query);
172+ const DnsResolver::iterator end;
173+
174+ while (iterator != end)
175+ {
176+ auto endPoint = *iterator++;
177+
178+ if (!endPoint.host_name().empty())
179+ {
180+ m_hostName = endPoint.host_name();
181+ m_connectionID = m_hostName + " (" + m_connectionID + ")";
182+ break;
183+ }
184+ }
185+ }
186+ catch (...)
187+ { //-V565
188+ // DNS lookup failure is not a catastrophe
189+ }
190+
191+ if (m_hostName.empty())
192+ m_hostName = m_ipAddress.to_string();
193+
194+ m_pingTimer.Start();
194195 }
195196
196197 void ClientConnection::CommandChannelSendAsync(uint8_t* data, uint32_t offset, uint32_t length)
@@ -241,7 +242,6 @@ DataPublisher::DataPublisher(const TcpEndPoint& endpoint) : · 44 unmodified lines
241242 m_errorMessageCallback(nullptr),
242243 m_clientConnectedCallback(nullptr)
243244 {
244− m_commandChannelService.restart();
245245 m_callbackThread = Thread(bind(&DataPublisher::RunCallbackThread, this));
246246 m_commandChannelAcceptThread = Thread(bind(&DataPublisher::RunCommandChannelAcceptThread, this));
247247 }
@@ -253,6 +253,13 @@ DataPublisher::DataPublisher(uint16_t port, bool ipV6) : · 5 unmodified lines
253253
254254 DataPublisher::~DataPublisher() = default;
255255
256+DataPublisher::CallbackDispatcher::CallbackDispatcher() :
257+ Source(nullptr),
258+ Data(nullptr),
259+ Function(nullptr)
260+{
261+}
262+
256263 void DataPublisher::RunCallbackThread()
257264 {
258265 while (true)
@@ -346,8 +353,8 @@ void DataPublisher::HandleMetadataRefresh(const ClientConnectionPtr& connection, · 87 unmodified lines
346353
347354 if (rowCount > 0)
348355 {
349− const auto elapsedTime = (UtcNow() - startTime).total_seconds();
350− DispatchStatusMessage(ToString(rowCount) + " records spanning " + ToString(tables.size()) + " tables of meta-data prepared in " + ToString(elapsedTime) + " seconds, sending response to " + connection->GetConnectionID() + "...");
356+ const TimeSpan elapsedTime = UtcNow() - startTime;
357+ DispatchStatusMessage(ToString(rowCount) + " records spanning " + ToString(tables.size()) + " tables of meta-data prepared in " + ToString(elapsedTime) + ", sending response to " + connection->GetConnectionID() + "...");
351358 }
352359 else
353360 {
Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataPublisher.h+3−0
@@ -65,6 +65,7 @@ namespace Transport · 64 unmodified lines
6565 {
6666 private:
6767 const DataPublisherPtr m_parent;
68+ GSF::IOContext& m_commandChannelService;
6869 GSF::Guid m_subscriberID;
6970 std::string m_connectionID;
7071 uint32_t m_operationalModes;
@@ -273,6 +274,8 @@ namespace Transport · 202 unmodified lines
273274
274275 friend class ClientConnection;
275276 };
277+
278+ typedef SharedPtr<DataPublisher> DataPublisherPtr;
276279 }}}
277280
278281 #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