AMPS C/C++ Client Class Reference AMPS C/C++ Client Version 5.3.5.7

HAClientImpl.hpp
1 //
3 // Copyright (c) 2010-2026 60East Technologies Inc., All Rights Reserved.
4 //
5 // This computer software is owned by 60East Technologies Inc. and is
6 // protected by U.S. copyright laws and other laws and by international
7 // treaties. This computer software is furnished by 60East Technologies
8 // Inc. pursuant to a written license agreement and may be used, copied,
9 // transmitted, and stored only in accordance with the terms of such
10 // license agreement and with the inclusion of the above copyright notice.
11 // This computer software or any other copies thereof may not be provided
12 // or otherwise made available to any other person.
13 //
14 // U.S. Government Restricted Rights. This computer software: (a) was
15 // developed at private expense and is in all respects the proprietary
16 // information of 60East Technologies Inc.; (b) was not developed with
17 // government funds; (c) is a trade secret of 60East Technologies Inc.
18 // for all purposes of the Freedom of Information Act; and (d) is a
19 // commercial item and thus, pursuant to Section 12.212 of the Federal
20 // Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
21 // Government's use, duplication or disclosure of the computer software
22 // is subject to the restrictions set forth by 60East Technologies Inc..
23 //
25
26 #ifndef _HACLIENTIMPL_H_
27 #define _HACLIENTIMPL_H_
28
29 #include <typeinfo>
30 #include <amps/ampsplusplus.hpp>
31 #include <amps/ServerChooser.hpp>
32 #include <amps/MemorySubscriptionManager.hpp>
33 #include <amps/ReconnectDelayStrategy.hpp>
34 #if __cplusplus >= 201103L || _MSC_VER >= 1900
35 #include <atomic>
36 #endif
37
38 namespace AMPS
39 {
40
41 class HAClientImpl : public ClientImpl
42 {
43 public:
44 HAClientImpl(const std::string& name_)
45 : ClientImpl(name_), _timeout(AMPS_HACLIENT_TIMEOUT_DEFAULT)
46 , _reconnectDelay(AMPS_HACLIENT_RECONNECT_DEFAULT)
47 , _reconnectDelayStrategy(new ExponentialDelayStrategy(_reconnectDelay))
48 , _disconnected(false)
49 {
50 setDisconnectHandler(HADisconnectHandler());
51 setSubscriptionManager(new MemorySubscriptionManager());
52 }
53
54 ~HAClientImpl()
55 {
56 _disconnected = true;
57 _cleanup();
58 }
59
60 void setTimeout(int timeout_)
61 {
62 _timeout = timeout_;
63 }
64
65 int getTimeout() const
66 {
67 return _timeout;
68 }
69
70 unsigned int getReconnectDelay(void) const
71 {
72 return _reconnectDelay;
73 }
74
75 void setReconnectDelay(unsigned int reconnectDelay_)
76 {
77 _reconnectDelay = reconnectDelay_;
78 setReconnectDelayStrategy(new FixedDelayStrategy(
79 (unsigned int)reconnectDelay_));
80 }
81
82 void setReconnectDelayStrategy(const ReconnectDelayStrategy& strategy_)
83 {
84 _reconnectDelayStrategy = strategy_;
85 _reconnectDelay = 0;
86 }
87
88 ReconnectDelayStrategy getReconnectDelayStrategy(void) const
89 {
90 return _reconnectDelayStrategy;
91 }
92
93 std::string getLogonOptions(void) const
94 {
95 return _logonOptions;
96 }
97
98 void setLogonOptions(const std::string& logonOptions_)
99 {
100 _logonOptions = logonOptions_;
101 }
102
103 void setLogonOptions(const char* logonOptions_)
104 {
105 _logonOptions = logonOptions_;
106 }
107
108 ServerChooser getServerChooser() const
109 {
110 return _serverChooser;
111 }
112
113 void setServerChooser(const ServerChooser& serverChooser_)
114 {
115 _serverChooser = serverChooser_;
116 }
117
118 class HADisconnectHandler
119 {
120 public:
121 HADisconnectHandler() {}
122 static void invoke(Client& client, void* );
123 void operator()(Client& client)
124 {
125 invoke(client, NULL);
126 }
127 };
128 void connectAndLogon()
129 {
130 Lock<Mutex> l(_connectAndLogonLock);
131 // AC-1030 In case this is called on a client after delay strategy caused a failure.
132 // Reset delay strategy, then get the duration default and reset again
133 _reconnectDelayStrategy.reset();
134 _reconnectDelay = _reconnectDelayStrategy.getConnectWaitDuration("DUMMY_URI");
135 _reconnectDelayStrategy.reset();
136 try
137 {
138 _disconnected = false;
139 connectAndLogonInternal();
140 }
141 catch (const RetryOperationException&)
142 { // -V565 // -V5002
143 // pass - recv thread was started and is handling reconnect
144 }
145 catch (...)
146 {
147 // Failure, make sure we're disconnected
148 disconnect();
149 throw;
150 }
151 }
152
153 virtual void connect(const std::string& /*uri*/)
154 {
155 connectAndLogon();
156 }
157
158 virtual std::string logon(long /*timeout_*/, Authenticator& /*authenticator_*/,
159 const char* /*options_*/)
160 {
161 if (_disconnected)
162 {
163 throw DisconnectedException("Attempt to call logon on a disconnected HAClient. Use connectAndLogon() instead.");
164 }
165 throw AlreadyConnectedException("Attempt to call logon on an HAClient. Use connectAndLogon() instead.");
166 }
167
168 class DisconnectHandlerDisabler
169 {
170 public:
171 DisconnectHandlerDisabler()
172 : _pClient(NULL), _queueAckTimeout(0), _disconnect(false) { }
173 DisconnectHandlerDisabler(HAClientImpl* pClient_)
174 : _pClient(pClient_)
175 , _queueAckTimeout(0)
176 , _disconnect(false)
177 {
178 setHandler();
179 _queueAckTimeout = _pClient->getAckTimeout();
180 _pClient->setAckTimeout(0);
181 }
182 ~DisconnectHandlerDisabler()
183 {
184 _clear();
185 }
186 void clear()
187 {
188 _clear();
189 if (_disconnect)
190 {
191 _disconnect = false;
192 throw DisconnectedException("Client disconnected during logon.");
193 }
194 }
195 void _clear()
196 {
197 if (_pClient)
198 {
199 amps_client_set_disconnect_handler(
200 _pClient->getHandle(),
201 (amps_handler)ClientImpl::ClientImplDisconnectHandler,
202 _pClient);
203 if (_queueAckTimeout)
204 {
205 _pClient->setAckTimeout(_queueAckTimeout);
206 _queueAckTimeout = 0;
207 }
208 _pClient = NULL;
209 }
210 }
211 void setClient(HAClientImpl* pClient_)
212 {
213 if (!_pClient)
214 {
215 _pClient = pClient_;
216 setHandler();
217 _queueAckTimeout = _pClient->getAckTimeout();
218 _pClient->setAckTimeout(0);
219 amps_client_disconnect(_pClient->getHandle());
220 _disconnect = false;
221 }
222 }
223 void setHandler()
224 {
225 _disconnect = false;
226 amps_client_set_disconnect_handler(
227 _pClient->getHandle(),
228 (amps_handler)HAClientImpl::DisconnectHandlerDisabler::HADoNothingDisconnectHandler,
229 (void*)&_disconnect);
230 }
231 static void HADoNothingDisconnectHandler(amps_handle /*client*/,
232 void* pDisconnect_)
233 {
234 *(bool*)pDisconnect_ = true;
235 }
236
237 private:
238 HAClientImpl* _pClient;
239 int _queueAckTimeout;
240 bool _disconnect;
241 };
242
243 void connectAndLogonInternal()
244 {
245 if (!_serverChooser.isValid())
246 {
247 throw ConnectionException("No server chooser registered with HAClient");
248 }
249 {
250 DisconnectHandlerDisabler disconnectDisabler;
251 TryLock<Mutex> l(_connectLock);
252 if (!l.isLocked())
253 {
254 throw RetryOperationException("Retry, another thread is handling reconnnect");
255 }
256 while (!_disconnected)
257 {
258 std::string uri = _serverChooser.getCurrentURI();
259 if (uri.empty())
260 {
261 throw ConnectionException("No AMPS instances available for connection. " + _serverChooser.getError());
262 }
263 Authenticator& auth = _serverChooser.getCurrentAuthenticator();
264 _sleepBeforeConnecting(uri);
265 try
266 {
267 // Check if another thread disconnected or already connected
268 if (_disconnected)
269 {
270 return;
271 }
272 else if (_connected)
273 {
274 throw RetryOperationException("Another thread has reconnected.");
275 }
276 // Temporarily unset the disconnect handler since we will loop
277 disconnectDisabler.setClient((HAClientImpl*)this);
278 // Connect and logon while holding the _lock
279 {
280 Lock<Mutex> clientLock(_lock);
281 ClientImpl::_connect(uri);
282 try
283 {
284 captureSubscriptionManagerGenerationCount();
285 if (_logonOptions.empty())
286 {
287 ClientImpl::_logon(_timeout, auth);
288 }
289 else
290 {
291 ClientImpl::_logon(_timeout, auth, _logonOptions.c_str());
292 }
293 } // All of the following may not disconnect the client
294 catch (const AuthenticationException&)
295 {
296 ClientImpl::setDisconnected();
297 throw;
298 }
299 catch (const NotEntitledException&)
300 {
301 ClientImpl::setDisconnected();
302 throw;
303 }
304 catch (const DuplicateLogonException&)
305 {
306 ClientImpl::setDisconnected();
307 throw;
308 }
309 catch (const NameInUseException&)
310 {
311 ClientImpl::setDisconnected();
312 throw;
313 }
314 catch (const TimedOutException&)
315 {
316 ClientImpl::setDisconnected();
317 throw;
318 }
319 try
320 {
321 _serverChooser.reportSuccess(getConnectionInfo());
322 // We're clear, reset delay strategy, then get the duration default and reset again
323 _reconnectDelayStrategy.reset();
324 _reconnectDelay = _reconnectDelayStrategy.getConnectWaitDuration("DUMMY_URI");
325 _reconnectDelayStrategy.reset();
326 }
327 catch (const AMPSException&)
328 {
329 Unlock<Mutex> clientUnlock(_lock);
330 ClientImpl::disconnect();
331 throw;
332 }
333 // Resubscribe
334 if (!_disconnected
335 && _subscriptionManager)
336 {
337 Client c(this, false);
338 {
339 Unlock<Mutex> clientUnlock(_lock);
340 _subscriptionManager->resubscribe(c);
341 // Any disconnect during resubscribe will throw here
342 // Because we own _connectLock, the receive thread can attempt
343 // to handle disconnect, but it will get a retry and give up
344 // when it can't lock _connectLock.
345 disconnectDisabler.clear();
346 }
347 broadcastConnectionStateChanged(ConnectionStateListener::Resubscribed);
348 break;
349 }
350 }
351 disconnectDisabler.clear();
352 break;
353 }
354 catch (const RetryOperationException&)
355 {
356 throw;
357 }
358 catch (const AMPSException& ex)
359 {
360 ConnectionInfo ci = getConnectionInfo();
361 // Substitute the URI on the connection info with the one we attempted
362 ci["client.uri"] = std::move(uri); // -V823 we know client.uri already exists
363 _serverChooser.reportFailure(ex, ci);
364 try
365 {
366 ClientImpl::setDisconnected();
367 }
368 catch (const std::exception& e)
369 {
370 try
371 {
372 _exceptionListener->exceptionThrown(e);
373 }
374 catch (...) { } // -V565 // -V5002
375 }
376 catch (...)
377 {
378 try
379 {
380 _exceptionListener->exceptionThrown(UnknownException("Unknown exception calling setDisconnected"));
381 }
382 catch (...) { } // -V565 // -V5002
383 }
384 }
385 }
386 }
387 return;
388 }
389
390 ConnectionInfo gatherConnectionInfo() const
391 {
392 return getConnectionInfo();
393 }
394
395 ConnectionInfo getConnectionInfo() const
396 {
397 ConnectionInfo info = ClientImpl::getConnectionInfo();
398 std::ostringstream writer;
399
400 writer << getReconnectDelay();
401 info["haClient.reconnectDelay"] = writer.str();
402 writer.clear(); writer.str("");
403 writer << _timeout;
404 info["haClient.timeout"] = writer.str();
405
406 return info;
407 }
408
409 bool disconnected() const
410 {
411 return _disconnected;
412 }
413 private:
414
415 void disconnect()
416 {
417 _disconnected = true;
418 // Grabbing this lock ensures no other thread is trying to reconnect
419 Lock<Mutex> l(_connectLock);
420 ClientImpl::disconnect();
421 }
422 void _millisleep(unsigned int millis_)
423 {
424 if (millis_ == 0)
425 {
426 return;
427 }
428 double waitTime = (double)millis_;
429 Timer timer(waitTime);
430 timer.start();
431 while (!timer.checkAndGetRemaining(&waitTime))
432 {
433 if (waitTime - 1000.0 > 0.0)
434 {
435 AMPS_USLEEP(1000000);
436 }
437 else
438 {
439 AMPS_USLEEP(1000UL * (unsigned int)waitTime);
440 }
441 amps_invoke_waiting_function();
442 }
443 }
444 void _sleepBeforeConnecting(const std::string& uri_)
445 {
446 try
447 {
448 _reconnectDelay = _reconnectDelayStrategy.getConnectWaitDuration(uri_);
449 _millisleep(_reconnectDelay);
450 }
451 catch (const ConnectionException&)
452 {
453 throw;
454 }
455 catch (const std::exception& ex_)
456 {
457 _exceptionListener->exceptionThrown(ex_);
458 throw ConnectionException(ex_.what());
459 }
460 catch (...)
461 {
462 throw ConnectionException("Unknown exception thrown by "
463 "the HAClient's delay strategy.");
464 }
465 }
466
467 Mutex _connectLock;
468 Mutex _connectAndLogonLock;
469 int _timeout;
470 unsigned int _reconnectDelay;
471 ReconnectDelayStrategy _reconnectDelayStrategy;
472 ServerChooser _serverChooser;
473 #if __cplusplus >= 201103L || _MSC_VER >= 1900
474 std::atomic<bool> _disconnected;
475 #else
476 volatile bool _disconnected;
477 #endif
478 std::string _logonOptions;
479
480 }; // class HAClientImpl
481
482 }// namespace AMPS
483
484 #endif //_HACLIENTIMPL_H_
MemorySubscriptionManager.hpp
Provides AMPS::MemorySubscriptionManager, used by an AMPS::HAClient to resubmit subscriptions if conn...
ReconnectDelayStrategy.hpp
Provides AMPS::ReconnectDelayStrategy, called by an AMPS::HAClient to determine how long to wait betw...
ServerChooser.hpp
Provides AMPS::ServerChooser, the abstract base class that defines the interface that an AMPS::HAClie...
amps_client_disconnect
AMPSDLL void amps_client_disconnect(amps_handle handle)
Disconnects from the AMPS server, if connected.
amps_handle
void * amps_handle
Opaque handle type used to refer to objects in the AMPS api.
Definition: amps.h:211
amps_client_set_disconnect_handler
AMPSDLL void amps_client_set_disconnect_handler(amps_handle client, amps_handler disconnectHandler, void *userData)
Sets the disconnect handler function to be called when a disconnect occurs.
ampsplusplus.hpp
Core type, function, and class declarations for the AMPS C++ client.