TimeSeriesPlatformLibrary: (1) Improved operation of SubscriberInstance, (2) Updated function signatures for DataSubscriber to always include subscriber reference since callbacks require a static function, this allows "instance" implementations to have access to source subscriber.

J. Ritchie Carroll committed 8 years ago

Commit ddde4ba1 parent 24ecf3a7 files · +336 −230Browse files

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

Unified · Split

Source/Libraries/TimeSeriesPlatformLibrary/Samples/SubscriberInstance.cpp+211−134

@@ -21,22 +21,21 @@ · 20 unmodified lines
21	21	//
22	22	//******************************************************************************************************
23	23
	24	+#include <functional>
	25	+
24	26	#include "SubscriberInstance.h"
25	27	#include "../Transport/Constants.h"
26	28
27	29	SubscriberInstance::SubscriberInstance() :
28		− m_hostname("localhost"),
29		− m_port(6165),
30		− m_udpPort(0),
31		− m_filterExpression("FILTER ActiveMeasurements WHERE Protocol = 'GatewayTransport'"),
32		− m_startTime(""),
33		− m_stopTime(""),
34		− m_statusMessageCallback(0),
35		− m_errorMessageCallback(0),
36		− m_metadataCallback(0),
37		− m_newMeasurementsCallback(0),
38		− m_historicalReadCompleteCallback(0)
	30	+ m_hostname("localhost"),
	31	+ m_port(6165),
	32	+ m_udpPort(0),
	33	+ m_filterExpression(SubscribeAllExpression),
	34	+ m_startTime(""),
	35	+ m_stopTime("")
39	36	{
	37	+ // Reference this SubscriberInstance in DataSubsciber user data
	38	+ m_subscriber.SetUserData(this);
40	39	}
41	40
42	41	SubscriberInstance::~SubscriberInstance()
@@ -45,186 +44,264 @@ SubscriberInstance::~SubscriberInstance() · 2 unmodified lines
45	44
46	45	// public functions
47	46
48		−void SubscriberInstance::RegisterStatusMessageCallback(MessageCallback statusMessageCallback)
	47	+void SubscriberInstance::Initialize(std::string hostname, gsfts::uint16_t port, gsfts::uint16_t udpPort)
49	48	{
50		− m_statusMessageCallback = statusMessageCallback;
	49	+ m_hostname = hostname;
	50	+ m_port = port;
	51	+ m_udpPort = udpPort;
51	52	}
52	53
53		−void SubscriberInstance::RegisterErrorMessageCallback(MessageCallback errorMessageCallback)
	54	+void SubscriberInstance::EstablishHistoricalRead(std::string startTime, std::string stopTime)
54	55	{
55		− m_errorMessageCallback = errorMessageCallback;
	56	+ m_startTime = startTime;
	57	+ m_stopTime = stopTime;
56	58	}
57	59
58		−void SubscriberInstance::RegisterMetadataCallback(MetadataCallback metadataCallback)
	60	+void SubscriberInstance::SetFilterExpression(std::string filterExpression)
59	61	{
60		− m_metadataCallback = metadataCallback;
	62	+ m_filterExpression = filterExpression;
61	63	}
62	64
63		−void SubscriberInstance::RegisterNewMeasurementsCallback(NewMeasurementsCallback processMeasurementsCallback)
	65	+void SubscriberInstance::Connect()
64	66	{
65		− m_newMeasurementsCallback = processMeasurementsCallback;
	67	+ // The connector is declared here because it
	68	+ // is only needed for the initial connection
	69	+ tst::SubscriberConnector connector;
	70	+
	71	+ // Set up helper objects (derived classes can override behavior)
	72	+ connector = CreateSubscriberConnector();
	73	+ m_info = CreateSubscriptionInfo();
	74	+
	75	+ // Register callbacks
	76	+ m_subscriber.RegisterStatusMessageCallback(&HandleStatusMessage);
	77	+ m_subscriber.RegisterErrorMessageCallback(&HandleErrorMessage);
	78	+
	79	+ if (!m_startTime.empty() && !m_stopTime.empty())
	80	+ {
	81	+ m_subscriber.RegisterProcessingCompleteCallback(&HandleProcessingComplete);
	82	+ m_info.StartTime = m_startTime;
	83	+ m_info.StopTime = m_stopTime;
	84	+ }
	85	+
	86	+ if (m_udpPort > 0)
	87	+ {
	88	+ m_info.UdpDataChannel = true;
	89	+ m_info.DataChannelLocalPort = m_udpPort;
	90	+ }
	91	+
	92	+ m_subscriber.RegisterMetadataCallback(this->HandleMetadata);
	93	+ m_subscriber.RegisterConfigurationChangedCallback(this->HandleConfigurationChanged);
	94	+
	95	+ // Connect and subscribe to publisher
	96	+ if (connector.Connect(m_subscriber))
	97	+ {
	98	+ // Request metadata upon successful connection, after metadata is handled
	99	+ // the SubscriberInstance will then subscribe to the desired data
	100	+ m_subscriber.SendServerCommand(tst::ServerCommand::MetadataRefresh);
	101	+ }
	102	+ else
	103	+ {
	104	+ ErrorMessage("All connection attempts failed");
	105	+ }
66	106	}
67	107
68		−void SubscriberInstance::Initialize(std::string hostname, gsfts::uint16_t port, gsfts::uint16_t udpPort)
	108	+void SubscriberInstance::Disconnect()
69	109	{
70		− m_hostname = hostname;
71		− m_port = port;
72		− m_udpPort = udpPort;
	110	+ m_subscriber.Disconnect();
73	111	}
74	112
75		−void SubscriberInstance::EstablishHistoricalRead(std::string startTime, std::string stopTime)
	113	+void SubscriberInstance::SetHistoricalReplayInterval(int32_t replayInterval)
76	114	{
77		− m_startTime = startTime;
78		− m_stopTime = stopTime;
	115	+ if (m_subscriber.IsSubscribed())
	116	+ {
	117	+ replayInterval = m_endianConverter.ConvertBigEndian(replayInterval);
	118	+ m_subscriber.SendServerCommand(tst::ServerCommand::UpdateProcessingInterval, (gsfts::uint8_t*)&replayInterval, 0, 4);
	119	+ }
79	120	}
80	121
81		−void SubscriberInstance::Connect()
	122	+void* SubscriberInstance::GetUserData() const
82	123	{
83		− // The connector is declared here because it
84		− // is only needed for the initial connection
85		− tst::SubscriberConnector connector;
86		−
87		− // Set up helper objects
88		− connector = CreateSubscriberConnector(m_hostname, m_port);
89		− m_info = CreateSubscriptionInfo();
90		−
91		− // Register callbacks
92		− if (m_statusMessageCallback)
93		− m_subscriber.RegisterStatusMessageCallback(m_statusMessageCallback);
94		−
95		− if (m_errorMessageCallback)
96		− m_subscriber.RegisterErrorMessageCallback(m_errorMessageCallback);
97		−
98		− if (!m_startTime.empty() && !m_stopTime.empty())
99		− {
100		− m_subscriber.RegisterProcessingCompleteCallback(this->HandleProcessComplete);
101		− m_info.StartTime = m_startTime;
102		− m_info.StopTime = m_stopTime;
103		− }
104		−
105		− if (m_udpPort > 0)
106		− {
107		− m_info.UdpDataChannel = true;
108		− m_info.DataChannelLocalPort = m_udpPort;
109		− }
110		−
111		− m_subscriber.RegisterMetadataCallback(this->HandleMetadata);
112		− m_subscriber.RegisterConfigurationChangedCallback(this->HandleConfigurationChanged);
113		−
114		− // Connect and subscribe to publisher
115		− if (connector.Connect(m_subscriber))
116		− {
117		− // Request metadata upon successful connection, after metadata is handled
118		− // the SubscriberInstance will then subscribe to the desired data
119		− m_subscriber.SendServerCommand(tst::ServerCommand::MetadataRefresh);
120		− }
121		− else
122		− {
123		− if (m_errorMessageCallback)
124		− m_errorMessageCallback("All connection attempts failed");
125		− }
	124	+ return m_userData;
126	125	}
127	126
128		−void SubscriberInstance::Disconnect()
	127	+void SubscriberInstance::SetUserData(void* userData)
129	128	{
130		− m_subscriber.Disconnect();
	129	+ m_userData = userData;
131	130	}
132	131
133		−void SubscriberInstance::SetHistoricalReplayInterval(int32_t replayInterval)
	132	+bool SubscriberInstance::IsMetadataCompressed() const
134	133	{
135		− if (m_subscriber.IsSubscribed())
136		− {
137		− replayInterval = m_endianConverter.ConvertBigEndian(replayInterval);
138		− m_subscriber.SendServerCommand(tst::ServerCommand::UpdateProcessingInterval, (gsfts::uint8_t*)&replayInterval, 0, 4);
139		− }
	134	+ return m_subscriber.IsMetadataCompressed();
140	135	}
141	136
142		−// private functions
	137	+void SubscriberInstance::SetMetadataCompressed(bool compressed)
	138	+{
	139	+ m_subscriber.SetMetadataCompressed(compressed);
	140	+}
143	141
144		−tst::SubscriberConnector SubscriberInstance::CreateSubscriberConnector(std::string hostname, gsfts::uint16_t port)
	142	+long SubscriberInstance::GetTotalCommandChannelBytesReceived() const
145	143	{
146		− // SubscriberConnector is another helper object which allows the
147		− // user to modify settings for auto-reconnects and retry cycles.
148		− tst::SubscriberConnector connector;
	144	+ return m_subscriber.GetTotalCommandChannelBytesReceived();
	145	+}
149	146
150		− if (m_errorMessageCallback)
151		− connector.RegisterErrorMessageCallback(m_errorMessageCallback);
	147	+long SubscriberInstance::GetTotalDataChannelBytesReceived() const
	148	+{
	149	+ return m_subscriber.GetTotalDataChannelBytesReceived();
	150	+}
152	151
153		− connector.RegisterReconnectCallback(this->HandleResubscribe);
	152	+long SubscriberInstance::GetTotalMeasurementsReceived() const
	153	+{
	154	+ return m_subscriber.GetTotalMeasurementsReceived();
	155	+}
154	156
155		− connector.SetHostname(hostname);
156		− connector.SetPort(port);
157		− connector.SetMaxRetries(5);
158		− connector.SetRetryInterval(1000);
159		− connector.SetAutoReconnect(true);
	157	+bool SubscriberInstance::IsConnected() const
	158	+{
	159	+ return m_subscriber.IsConnected();
	160	+}
160	161
161		− return connector;
	162	+bool SubscriberInstance::IsSubscribed() const
	163	+{
	164	+ return m_subscriber.IsSubscribed();
	165	+}
	166	+
	167	+// protected functions
	168	+
	169	+tst::SubscriberConnector SubscriberInstance::CreateSubscriberConnector()
	170	+{
	171	+ // SubscriberConnector is another helper object which allows the
	172	+ // user to modify settings for auto-reconnects and retry cycles.
	173	+ tst::SubscriberConnector connector;
	174	+
	175	+ // Register callbacks
	176	+ connector.RegisterErrorMessageCallback(&HandleErrorMessage);
	177	+ connector.RegisterReconnectCallback(&HandleResubscribe);
	178	+
	179	+ connector.SetHostname(m_hostname);
	180	+ connector.SetPort(m_port);
	181	+ connector.SetMaxRetries(-1);
	182	+ connector.SetRetryInterval(5000);
	183	+ connector.SetAutoReconnect(true);
	184	+
	185	+ return connector;
162	186	}
163	187
164	188	tst::SubscriptionInfo SubscriberInstance::CreateSubscriptionInfo()
165	189	{
166		− // SubscriptionInfo is a helper object which allows the user
167		− // to set up their subscription and reuse subscription settings.
168		− tst::SubscriptionInfo info;
	190	+ // SubscriptionInfo is a helper object which allows the user
	191	+ // to set up their subscription and reuse subscription settings.
	192	+ tst::SubscriptionInfo info;
169	193
170		− // Define desired filter expression
171		− info.FilterExpression = m_filterExpression;
	194	+ // Define desired filter expression
	195	+ info.FilterExpression = m_filterExpression;
172	196
173		− // Establish user new measurements call back (if defined)
174		− if (m_newMeasurementsCallback)
175		− info.NewMeasurementsCallback = m_newMeasurementsCallback;
	197	+ // Establish user new measurements call back (if defined)
	198	+ info.NewMeasurementsCallback = &HandleNewMeasurements;
176	199
177		− // To set up a remotely synchronized subscription, set this flag
178		− // to true and add the framesPerSecond parameter to the
179		− // ExtraConnectionStringParameters. Additionally, the following
180		− // example demonstrates the use of some other useful parameters
181		− // when setting up remotely synchronized subscriptions.
182		− //
183		− //info.RemotelySynchronized = true;
184		− //info.ExtraConnectionStringParameters = "framesPerSecond=30;timeResolution=10000;downsamplingMethod=Closest";
	200	+ // To set up a remotely synchronized subscription, set this flag
	201	+ // to true and add the framesPerSecond parameter to the
	202	+ // ExtraConnectionStringParameters. Additionally, the following
	203	+ // example demonstrates the use of some other useful parameters
	204	+ // when setting up remotely synchronized subscriptions.
	205	+ //
	206	+ //info.RemotelySynchronized = true;
	207	+ //info.ExtraConnectionStringParameters = "framesPerSecond=30;timeResolution=10000;downsamplingMethod=Closest";
185	208
186		− info.RemotelySynchronized = false;
187		− info.Throttled = false;
	209	+ info.RemotelySynchronized = false;
	210	+ info.Throttled = false;
188	211
189		− info.UdpDataChannel = false;
	212	+ info.UdpDataChannel = false;
190	213
191		− info.IncludeTime = true;
192		− info.LagTime = 3.0;
193		− info.LeadTime = 1.0;
194		− info.UseLocalClockAsRealTime = false;
195		− info.UseMillisecondResolution = true;
	214	+ info.IncludeTime = true;
	215	+ info.LagTime = 3.0;
	216	+ info.LeadTime = 1.0;
	217	+ info.UseLocalClockAsRealTime = false;
	218	+ info.UseMillisecondResolution = true;
196	219
197		− return info;
	220	+ return info;
198	221	}
199	222
200		−// Callback that is called when the subscriber auto-reconnects
	223	+void SubscriberInstance::ConnectionTerminated()
	224	+{
	225	+}
	226	+
	227	+void SubscriberInstance::StatusMessage(std::string message)
	228	+{
	229	+ std::cout << message << std::endl << std::endl;
	230	+}
	231	+
	232	+void SubscriberInstance::ErrorMessage(std::string message)
	233	+{
	234	+ std::cerr << message << std::endl << std::endl;
	235	+}
	236	+
	237	+void SubscriberInstance::ReceivedMetadata(std::vector<uint8_t> payload)
	238	+{
	239	+}
	240	+
	241	+void SubscriberInstance::ReceivedNewMeasurements(std::vector<gsfts::Measurement> measurements)
	242	+{
	243	+}
	244	+
	245	+void SubscriberInstance::HistoricalReadComplete()
	246	+{
	247	+}
	248	+
	249	+// private functions
	250	+
	251	+// The following member methods are defined as static so they
	252	+// can be used as callback registrations for DataSubscriber
	253	+
201	254	void SubscriberInstance::HandleResubscribe(tst::DataSubscriber* source)
202	255	{
203		− if (source->IsConnected())
204		− source->Subscribe(m_info);
	256	+ if (source->IsConnected())
	257	+ source->Subscribe(source->GetCurrentSubscription());
205	258	}
206	259
207		−void SubscriberInstance::HandleMetadata(std::vector<uint8_t> payload)
	260	+void SubscriberInstance::HandleStatusMessage(tst::DataSubscriber* source, std::string message)
208	261	{
209		− // Call user defined handle meta-data function
210		− if (m_metadataCallback)
211		− m_metadataCallback(payload);
	262	+ SubscriberInstance* instance = (SubscriberInstance*)source->GetUserData();
	263	+ instance->StatusMessage(message);
	264	+}
212	265
213		− // Start subscription after successful meta-data parse
214		− m_subscriber.Subscribe(m_info);
	266	+void SubscriberInstance::HandleErrorMessage(tst::DataSubscriber* source, std::string message)
	267	+{
	268	+ SubscriberInstance* instance = (SubscriberInstance*)source->GetUserData();
	269	+ instance->ErrorMessage(message);
215	270	}
216	271
217		−void SubscriberInstance::HandleConfigurationChanged()
	272	+void SubscriberInstance::HandleConnectionTerminated(tst::DataSubscriber* source)
218	273	{
219		− // When publisher configuration has changed, request updated metadata
220		− m_subscriber.SendServerCommand(tst::ServerCommand::MetadataRefresh);
	274	+ SubscriberInstance* instance = (SubscriberInstance*)source->GetUserData();
	275	+ instance->ConnectionTerminated();
221	276	}
222	277
223		−void SubscriberInstance::HandleProcessComplete(std::string message)
	278	+void SubscriberInstance::HandleMetadata(tst::DataSubscriber* source, std::vector<uint8_t> payload)
224	279	{
225		− if (m_statusMessageCallback)
226		− m_statusMessageCallback(message);
	280	+ SubscriberInstance* instance = (SubscriberInstance*)source->GetUserData();
	281	+
	282	+ // Call virtual method to handle metadata payload
	283	+ instance->ReceivedMetadata(payload);
	284	+
	285	+ // Start subscription after successful user meta-data handling
	286	+ source->Subscribe(instance->m_info);
	287	+}
227	288
228		− if (m_historicalReadCompleteCallback)
229		− m_historicalReadCompleteCallback();
	289	+void SubscriberInstance::HandleNewMeasurements(tst::DataSubscriber* source, std::vector<gsfts::Measurement> measurements)
	290	+{
	291	+ SubscriberInstance* instance = (SubscriberInstance*)source->GetUserData();
	292	+ instance->ReceivedNewMeasurements(measurements);
	293	+}
	294	+
	295	+void SubscriberInstance::HandleConfigurationChanged(tst::DataSubscriber* source)
	296	+{
	297	+ // When publisher configuration has changed, request updated metadata
	298	+ source->SendServerCommand(tst::ServerCommand::MetadataRefresh);
	299	+}
	300	+
	301	+void SubscriberInstance::HandleProcessingComplete(tst::DataSubscriber* source, std::string message)
	302	+{
	303	+ SubscriberInstance* instance = (SubscriberInstance*)source->GetUserData();
	304	+
	305	+ instance->StatusMessage(message);
	306	+ instance->HistoricalReadComplete();
230	307	}

TimeSeriesPlatformLibrary: (1) Improved operation of SubscriberInstance, (2) Updated function signatures for DataSubscriber to always include subscriber reference since callbacks require a static function, this allows "instance" implementations to have access to source subscriber.

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code