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

MessageRouter.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 #ifndef _MESSAGEROUTER_HPP_
26 #define _MESSAGEROUTER_HPP_
27 #include <map>
28 #include "amps/ampscrc.hpp"
29 #include "amps/util.hpp"
30 #include "amps/Message.hpp"
31
32 namespace AMPS
33 {
38 template <typename Func, typename Object>
39 class Handler
40 {
41 protected:
42 friend class MessageStream;
43 Func _func;
44 void* _userData;
45 std::function<void(Object)> _callable;
46 bool _isValid;
47
48 public:
49 // No op function for handlers
50 static void noOpHandler(Object) {;}
51
52 typedef Func FunctionType;
55 Handler() : _func(NULL), _userData(NULL)
56 , _callable(Handler<Func, Object>::noOpHandler)
57 , _isValid(false)
58 {
59 }
60
66 Handler(Func func_, void* userData_)
67 : _func(func_), _userData(userData_)
68 , _callable(noOpHandler)
69 , _isValid(true)
70 {
71 }
72
75 Handler(const Handler& orig_)
76 : _func(orig_._func), _userData(orig_._userData)
77 , _callable(orig_._callable)
78 , _isValid(orig_._isValid)
79 {
80 }
83 template <typename T>
84 Handler(const T& callback_)
85 : _func(NULL), _userData(NULL), _callable(callback_), _isValid(true)
86 {
87 }
88 void invoke(Object message)
89 {
90 if (_func)
91 {
92 _func(message, _userData);
93 }
94 else
95 {
96 _callable(message);
97 }
98 }
99
100 Handler& operator=(const Handler& rhs_)
101 {
102 if (this != &rhs_)
103 {
104 _func = rhs_._func;
105 _userData = rhs_._userData;
106 _callable = rhs_._callable;
107 _isValid = rhs_._isValid;
108 }
109 return *this;
110 }
111
112 bool isValid(void) const
113 {
114 return _isValid;
115 }
116 Func function(void) const
117 {
118 return _func;
119 }
120 void* userData(void) const
121 {
122 return _userData;
123 }
124 };
125 class Message;
128 typedef void(*MessageHandlerFunc)(const Message&, void* userData);
129
130 typedef Handler<MessageHandlerFunc, const Message&> MessageHandler;
131
135 class MessageRouter
136 {
137 private:
138 MessageHandler _emptyMessageHandler;
139 typedef amps_uint64_t (*CRCFunction)(const char*, size_t, amps_uint64_t);
140 // Function used to calculate the CRC if one is used
141 CRCFunction _crc;
142 class MessageRoute
143 {
144 MessageHandler _messageHandler;
145 unsigned _requestedAcks, _systemAcks, _terminationAck;
146 public:
147 MessageRoute() : _requestedAcks(0), _systemAcks(0), _terminationAck(0) {;}
148 MessageRoute(const MessageRoute& rhs_) :
149 _messageHandler(rhs_._messageHandler),
150 _requestedAcks (rhs_._requestedAcks),
151 _systemAcks (rhs_._systemAcks),
152 _terminationAck(rhs_._terminationAck)
153 {;}
154 const MessageRoute& operator=(const MessageRoute& rhs_)
155 {
156 _messageHandler = rhs_._messageHandler;
157 _requestedAcks = rhs_._requestedAcks;
158 _systemAcks = rhs_._systemAcks;
159 _terminationAck = rhs_._terminationAck;
160 return *this;
161 }
162 MessageRoute(MessageHandler messageHandler_, unsigned requestedAcks_,
163 unsigned systemAcks_, Message::Command::Type commandType_) :
164 _messageHandler(messageHandler_),
165 _requestedAcks(requestedAcks_),
166 _systemAcks(systemAcks_),
167 _terminationAck(0)
168 {
169 bool isSubscribeOrSOW = commandType_ & Message::Command::Subscribe
170 || commandType_ & Message::Command::DeltaSubscribe
171 || commandType_ & Message::Command::SOWAndSubscribe
172 || commandType_ & Message::Command::SOWAndDeltaSubscribe
173 || commandType_ & Message::Command::SOW;
174 if (!isSubscribeOrSOW)
175 {
176 // The ack to terminate the route on is whatever the highest
177 // bit set in requestedAcks is.
178 unsigned bitCounter = (requestedAcks_ | systemAcks_) >> 1;
179 _terminationAck = 1;
180 while (bitCounter > 0)
181 {
182 bitCounter >>= 1;
183 _terminationAck <<= 1;
184 }
185 }
186 else if (commandType_ == Message::Command::SOW)
187 {
188 if (requestedAcks_ >= Message::AckType::Completed)
189 {
190 // The ack to terminate the route on is whatever the highest
191 // bit set in requestedAcks is.
192 unsigned bitCounter = (requestedAcks_ | systemAcks_) >> 1;
193 _terminationAck = 1;
194 while (bitCounter > 0)
195 {
196 bitCounter >>= 1;
197 _terminationAck <<= 1;
198 }
199 }
200 else
201 {
202 _terminationAck = Message::AckType::Completed;
203 }
204 }
205 }
206
207 // Deliver an ack to registered handler if the ack type was requested
208 unsigned deliverAck(const Message& message_, unsigned ackType_)
209 {
210 if ( (_requestedAcks & ackType_) == 0)
211 {
212 return 0;
213 }
214 try
215 {
216 _messageHandler.invoke(message_);
217 }
218 catch (std::exception& ex)
219 {
220 std::cerr << ex.what() << std::endl;
221 }
222 return 1;
223 }
224 bool isTerminationAck(unsigned ackType_) const
225 {
226 return ackType_ == _terminationAck;
227 }
228 unsigned deliverData(const Message& message_)
229 {
230 _messageHandler.invoke(message_);
231 return 1;
232 }
233 const MessageHandler& getMessageHandler() const
234 {
235 return _messageHandler;
236 }
237 MessageHandler& getMessageHandler()
238 {
239 return _messageHandler;
240 }
241 };
242
243 public:
244 MessageRouter()
245 : _previousCommandId(0),
246 _lookupGenerationCount(0),
247 _generationCount(0)
248 {
249 #ifndef AMPS_SSE_42
250 _crc = AMPS::CRC<0>::crcNoSSE;
251 #else
252 if (AMPS::CRC<0>::isSSE42Enabled())
253 {
254 _crc = AMPS::CRC<0>::crc;
255 }
256 else
257 {
258 _crc = AMPS::CRC<0>::crcNoSSE;
259 }
260 #endif
261 }
262
263 int addRoute(const Field& commandId_, const AMPS::MessageHandler& messageHandler_,
264 unsigned requestedAcks_, unsigned systemAcks_, Message::Command::Type commandType_)
265 {
266 Lock<Mutex> lock(_lock);
267 RouteMap::iterator i = _routes.find(commandId_);
268 if (i == _routes.end())
269 {
270 _routes[commandId_.deepCopy()] = MessageRoute(messageHandler_, requestedAcks_, systemAcks_, commandType_);
271 return 1;
272 }
273 else
274 {
275 bool isSubscribe = commandType_ & Message::Command::Subscribe
276 || commandType_ & Message::Command::DeltaSubscribe
277 || commandType_ & Message::Command::SOWAndSubscribe
278 || commandType_ & Message::Command::SOWAndDeltaSubscribe;
279
280 // Only replace a non-subscribe with a subscribe
281 if (isSubscribe
282 && !i->second.isTerminationAck(0))
283 {
284 void* routeData = i->second.getMessageHandler().userData();;
285 i->second = MessageRoute(messageHandler_, requestedAcks_, systemAcks_, commandType_);
286 if (routeData)
287 {
288 Unlock<Mutex> u(_lock);
289 amps_invoke_remove_route_function(routeData);
290 }
291 return 1;
292 }
293 }
294 return 0;
295 }
296
297 // returns true if a route was removed.
298 bool removeRoute(const Field& commandId_)
299 {
300 Lock<Mutex> lock(_lock);
301 RouteMap::iterator i = _routes.find(commandId_);
302 if (i == _routes.end())
303 {
304 return false;
305 }
306 return _removeRoute(i);
307 }
308
309 void clear()
310 {
311 AMPS_FETCH_ADD(&_generationCount, 1);
312 std::vector<void*> removeData;
313 {
314 Lock<Mutex> lock(_lock);
315 for (RouteMap::iterator i = _routes.begin(); i != _routes.end(); ++i)
316 {
317 // Make a non-const copy of Field and clear it, which will clear i
318 // as well but won't actually affect the map, which is unaware that
319 // the key's shared pointer has been deleted.
320 Field f = i->first;
321 void* data = i->second.getMessageHandler().userData();
322 removeData.push_back(data);
323 f.clear();
324 }
325 _routes.clear();
326 }
327 for (size_t i = 0; i < removeData.size(); ++i)
328 {
329 amps_invoke_remove_route_function(removeData[i]);
330 }
331 }
332
333 // Returns true if a route exists for a single id.
334 bool hasRoute(const Field& commandId_) const
335 {
336 Lock<Mutex> lock(_lock);
337 RouteMap::const_iterator it = _routes.find(commandId_);
338 return it != _routes.end();
339 }
340
341 // Find a single route and return true if here, setting result_ to the handler.
342 bool getRoute(const Field& commandId_, MessageHandler& result_) const
343 {
344 Lock<Mutex> lock(_lock);
345 RouteMap::const_iterator it = _routes.find(commandId_);
346 if (it != _routes.end())
347 {
348 result_ = it->second.getMessageHandler();
349 return true;
350 }
351 else
352 {
353 result_ = _emptyMessageHandler;
354 return false;
355 }
356 }
357
358 // RouteCache is the result type for a parseRoutes(); we do extra work
359 // to avoid hitting the map or its lock when the subids field on
360 // publish messages does not change.
361 struct RouteLookup
362 {
363 size_t idOffset;
364 size_t idLength;
365 MessageHandler handler;
366 };
367 class RouteCache : public std::vector<RouteLookup>
368 {
369 RouteCache(const RouteCache&);
370 void operator=(const RouteCache&);
371 public:
372 RouteCache(void)
373 : _generationCount(0),
374 _hashVal(0)
375 {;}
376
377 void invalidateCache(void)
378 {
379 #if __cplusplus >= 201100L || _MSC_VER >= 1900
380 _generationCount.store(0);
381 #else
382 _generationCount = 0;
383 #endif
384 _hashVal = 0;
385 clear();
386 }
387 #if __cplusplus >= 201100L || _MSC_VER >= 1900
388 void invalidateCache(const std::atomic<uint_fast64_t>& generationCount_, amps_uint64_t hashVal_)
389 {
390 _generationCount.store(generationCount_);
391 _hashVal = hashVal_;
392 clear();
393 }
394 #else
395 void invalidateCache(const AMPS_ATOMIC_TYPE& generationCount_, amps_uint64_t hashVal_)
396 {
397 _generationCount = generationCount_;
398 _hashVal = hashVal_;
399 clear();
400 }
401 #endif
402
403 #if __cplusplus >= 201100L || _MSC_VER >= 1900
404 bool isCacheHit(const std::atomic<uint_fast64_t>& generationCount_, amps_uint64_t hashVal_) const
405 {
406 return _generationCount == generationCount_ && _hashVal == hashVal_;
407 }
408 #else
409 bool isCacheHit(const AMPS_ATOMIC_TYPE& generationCount_, amps_uint64_t hashVal_) const
410 {
411 return _generationCount == generationCount_ && _hashVal == hashVal_;
412 }
413 #endif
414
415 private:
416 #if __cplusplus >= 201100L || _MSC_VER >= 1900
417 std::atomic<uint_fast64_t> _generationCount;
418 #else
419 AMPS_ATOMIC_TYPE _generationCount;
420 #endif
421 amps_uint64_t _hashVal;
422 };
423
424 // Parses the command id list into the route lookup vector and assigns
425 // the found handlers into the list. Only intended to be called by the
426 // message handler thread. Returns the number of command/sub IDs parsed.
427 size_t parseRoutes(const Field& commandIdList_, RouteCache& result_)
428 {
429 // Super shortcut: if the whole subID list is the same as the previous one,
430 // then assume the result_ contains all the right handlers already, and that
431 // the offsets and lengths of subIds are unchanged.
432 amps_uint64_t listHash = _crc(commandIdList_.data(), commandIdList_.len(), 0);
433 if (result_.isCacheHit(_generationCount, listHash))
434 {
435 return result_.size();
436 }
437 result_.invalidateCache(_generationCount, listHash);
438
439 // Lock required now that we'll be using the route map.
440 Lock<Mutex> lockGuard(_lock);
441 size_t resultCount = 0;
442 const char* pStart = commandIdList_.data();
443 for (const char* p = pStart, *e = commandIdList_.len() + pStart; p < e;
444 ++p, ++resultCount)
445 {
446 const char* delimiter = p;
447 while (delimiter != e && *delimiter != ',')
448 {
449 ++delimiter;
450 }
451 AMPS::Field subId(p, (size_t)(delimiter - p));
452 result_.emplace_back(RouteLookup());
453 // Push back and then copy over fields; would emplace_back if available on
454 // all supported compilers.
455 RouteLookup& result = result_[resultCount];
456 result.idOffset = (size_t)(p - pStart);
457 result.idLength = (size_t)(delimiter - p);
458
459 RouteMap::const_iterator it = _routes.find(subId);
460 if (it != _routes.end())
461 {
462 result.handler = it->second.getMessageHandler();
463 }
464 else
465 {
466 result.handler = _emptyMessageHandler;
467 }
468 p = delimiter;
469 }
470 return resultCount;
471 }
472 unsigned deliverAck(const Message& ackMessage_, unsigned ackType_)
473 {
474 assert(ackMessage_.getCommand() == "ack");
475 unsigned messagesDelivered = 0;
476 Field key;
477
478 // Call _deliverAck, which will deliver to any waiting handlers
479 // AND remove the route if it's a termination ack
480 if (key = ackMessage_.getCommandId(), !key.empty())
481 {
482 messagesDelivered += _deliverAck(ackMessage_, ackType_, key);
483 }
484 if (key = ackMessage_.getQueryID(),
485 !key.empty() && messagesDelivered == 0)
486 {
487 messagesDelivered += _deliverAck(ackMessage_, ackType_, key);
488 }
489 if (key = ackMessage_.getSubscriptionId(),
490 !key.empty() && messagesDelivered == 0)
491 {
492 messagesDelivered += _deliverAck(ackMessage_, ackType_, key);
493 }
494 return messagesDelivered;
495 }
496
497 // deliverData may only be called by the message handler thread.
498 unsigned deliverData(const Message& dataMessage_, const Field& commandId_)
499 {
500 unsigned messagesDelivered = 0;
501 amps_uint64_t hval = _crc(commandId_.data(), commandId_.len(), 0);
502 if (_previousCommandId == hval &&
503 _lookupGenerationCount == _generationCount)
504 {
505 messagesDelivered += _previousHandler.deliverData(dataMessage_);
506 }
507 else
508 {
509 Lock<Mutex> lock(_lock);
510 RouteMap::iterator it = _routes.find(commandId_);
511 if (it != _routes.end())
512 {
513 _previousCommandId = hval;
514 #if __cplusplus >= 201100L || _MSC_VER >= 1900
515 _lookupGenerationCount.store(_generationCount);
516 #else
517 _lookupGenerationCount = _generationCount;
518 #endif
519 _previousHandler = it->second;
520 messagesDelivered += it->second.deliverData(dataMessage_);
521 }
522 }
523 return messagesDelivered;
524 }
525
526 void invalidateCache(void)
527 {
528 _previousCommandId = 0;
529 }
530
531 void unsubscribeAll(void)
532 {
533 AMPS_FETCH_ADD(&_generationCount, 1);
534 std::vector<Field> removeIds;
535 std::vector<void*> removeData;
536 Lock<Mutex> lock(_lock);
537 for (RouteMap::iterator it = _routes.begin(); it != _routes.end(); ++it)
538 {
539 if (it->second.isTerminationAck(0))
540 {
541 removeIds.push_back(it->first);
542 removeData.push_back(it->second.getMessageHandler().userData());
543 }
544 }
545 for (size_t i = 0; i < removeIds.size(); ++i)
546 {
547 // it can't be end() b/c we have the lock and found id above
548 RouteMap::iterator it = _routes.find(removeIds[i]);
549 // Make a non-const copy of Field and clear it, which will clear i as well
550 Field f = it->first; // -V783
551 f.clear();
552 _routes.erase(it);
553 }
554 Unlock<Mutex> u(_lock);
555 for (size_t i = 0; i < removeData.size(); ++i)
556 {
557 amps_invoke_remove_route_function(removeData[i]);
558 }
559 }
560
561 private:
562 typedef std::map<Field, MessageRoute> RouteMap;
563 RouteMap _routes;
564 mutable Mutex _lock;
565
566 MessageRoute _previousHandler;
567 amps_uint64_t _previousCommandId;
568 #if __cplusplus >= 201100L || _MSC_VER >= 1900
569 mutable std::atomic<uint_fast64_t> _lookupGenerationCount;
570 mutable std::atomic<uint_fast64_t> _generationCount;
571 #else
572 mutable AMPS_ATOMIC_TYPE _lookupGenerationCount;
573 mutable AMPS_ATOMIC_TYPE _generationCount;
574 #endif
575
576
577 // Deliver the ack to any waiting handlers
578 // AND remove the route if it's a termination ack
579 unsigned _deliverAck(const Message& ackMessage_, unsigned ackType_, Field& commandId_)
580 {
581 Lock<Mutex> lock(_lock);
582 unsigned messagesDelivered = 0;
583 RouteMap::iterator it = _routes.find(commandId_);
584 if (it != _routes.end())
585 {
586 MessageRoute& route = it->second;
587 messagesDelivered += route.deliverAck(ackMessage_, ackType_);
588 if (route.isTerminationAck(ackType_))
589 {
590 _removeRoute(it);
591 ++messagesDelivered;
592 }
593 }
594 return messagesDelivered;
595 }
596 unsigned _processAckForRemoval(unsigned ackType_, Field& commandId_)
597 {
598 Lock<Mutex> lock(_lock);
599 RouteMap::iterator it = _routes.find(commandId_);
600 if (it != _routes.end())
601 {
602 MessageRoute& route = it->second;
603 if (route.isTerminationAck(ackType_))
604 {
605 _removeRoute(it);
606 return 1U;
607 }
608 }
609 return 0U;
610 }
611
612 // returns true if a route was removed.
613 bool _removeRoute(RouteMap::iterator& it_)
614 {
615 // Called with lock already held
616 AMPS_FETCH_ADD(&_generationCount, 1);
617 // Make a non-const copy of Field and clear it, which will clear i as well
618 Field f = it_->first;
619 void* routeData = it_->second.getMessageHandler().userData();
620 _routes.erase(it_);
621 f.clear();
622 if (routeData)
623 {
624 Unlock<Mutex> u(_lock);
625 amps_invoke_remove_route_function(routeData);
626 }
627 return true;
628 }
629
630 };
631
632
633 }
634
635 #endif
Message.hpp
Defines the AMPS::Message class and related classes.
AMPS::Field
Field represents the value of a single field in a Message.
Definition: Field.hpp:88
AMPS::Field::clear
void clear()
Deletes the data associated with this Field, should only be used on Fields that were created as deepC...
Definition: Field.hpp:266
AMPS::Field::empty
bool empty() const
Returns 'true' if empty, 'false' otherwise.
Definition: Field.hpp:129
AMPS::Field::deepCopy
void deepCopy(const Field &orig_)
Makes self a deep copy of the original field.
Definition: Field.hpp:219
AMPS::Field::len
size_t len() const
Returns the length of the data underlying this field.
Definition: Field.hpp:286
AMPS::Field::data
const char * data() const
Returns the (non-null-terminated) data underlying this field.
Definition: Field.hpp:279
AMPS::Handler
Wrapper for callback functions in AMPS.
Definition: MessageRouter.hpp:40
AMPS::Handler::Handler
Handler()
Null constructor – no function is wrapped.
Definition: MessageRouter.hpp:55
AMPS::Handler::Handler
Handler(const Handler &orig_)
Copy constructor.
Definition: MessageRouter.hpp:75
AMPS::Handler::Handler
Handler(const T &callback_)
Constructor for use with a standard c++ library function object.
Definition: MessageRouter.hpp:84
AMPS::Handler::Handler
Handler(Func func_, void *userData_)
Constructor for use with a bare function pointer.
Definition: MessageRouter.hpp:66
AMPS::MessageRouter
This class multiplexes messages from AMPS to multiple subscribers and uses the stream of acks from AM...
Definition: MessageRouter.hpp:136
AMPS::MessageStream
An iterable object representing the results of an AMPS subscription and/or query.
Definition: ampsplusplus.hpp:5208
AMPS::Message
Message encapsulates a single message sent to or received from an AMPS server, and provides methods f...
Definition: Message.hpp:520
AMPS::Message::getCommand
Field getCommand() const
Retrieves the value of the Command header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1237
AMPS::Message::getCommandId
Field getCommandId() const
Retrieves the value of the CommandId header of the Message as a Field which references the underlying...
Definition: Message.hpp:1344
AMPS::Message::getSubscriptionId
Field getSubscriptionId() const
Retrieves the value of the SubscriptionId header of the Message as a Field which references the under...
Definition: Message.hpp:1469
AMPS::Message::getQueryID
Field getQueryID() const
Retrieves the value of the QueryID header of the Message as a Field which references the underlying b...
Definition: Message.hpp:1459