VLink  2.0.0
A high-performance communication middleware
abstract_factory.h
Go to the documentation of this file.
1 /*
2  * Copyright (C) 2026 by Thun Lu. All rights reserved.
3  * Author: Thun Lu <thun.lu@zohomail.cn>
4  * Repo: https://github.com/thun-res/vlink
5  * _ __ __ _ __
6  * | | / / / / (_) ____ / /__
7  * | | / / / / / / / __ \ / //_/
8  * | |/ / / /___ / / / / / / / ,<
9  * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10  *
11  * Licensed under the Apache License, Version 2.0 (the "License");
12  * you may not use this file except in compliance with the License.
13  * You may obtain a copy of the License at
14  *
15  * http://www.apache.org/licenses/LICENSE-2.0
16  *
17  * Unless required by applicable law or agreed to in writing, software
18  * distributed under the License is distributed on an "AS IS" BASIS,
19  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20  * See the License for the specific language governing permissions and
21  * limitations under the License.
22  */
23 
24 /**
25  * @file abstract_factory.h
26  * @brief Topic-scoped registration store that fans transport callbacks across @c NodeImpl peers.
27  *
28  * @details
29  * This is an internal implementation header used by the public VLink node templates
30  * (@c Publisher, @c Subscriber, @c Client, @c Server, @c Setter, @c Getter); it should not
31  * be included directly by application code. The header introduces two co-operating
32  * class templates that let several @c NodeImpl instances bound to the same logical
33  * topic share one registration record:
34  *
35  * - @c AbstractObject -- a per-topic record that owns the set of registered
36  * @c NodeImpl pointers together with six callback dictionaries (server connect,
37  * subscriber connect, request/response, serialised message, intra-process message
38  * and transport status).
39  * - @c AbstractFactory -- a thread-safe map keyed on @c FilterT (commonly the
40  * topic URL string) that lazily creates an @c AbstractObject for each key and
41  * reuses it via a cached @c std::weak_ptr until the last owner releases it.
42  *
43  * @par Registration flow
44  * @code
45  * +----------------------+
46  * | AbstractFactory<K> |
47  * | map<K, weak_ptr<O>> |
48  * +----------+-----------+
49  * | get_object<O>(key)
50  * v
51  * +----------------------+
52  * | AbstractObject<K> |
53  * | ImplList |
54  * | ConnectCallbackMap |
55  * | MsgCallbackMap ... |
56  * +----------+-----------+
57  * ^ | ^
58  * add_impl(impl) | | | register_msg_callback(impl, cb)
59  * remove_impl(impl)| | |
60  * | v
61  * +------+--+ +-+--------+ +----------+
62  * | NodeImpl| | NodeImpl | | NodeImpl |
63  * +---------+ +----------+ +----------+
64  * @endcode
65  *
66  * @par Registry keys
67  * | Map field | Callback signature |
68  * | ------------------------------- | ------------------------------------------------- |
69  * | @c server_connect_callback_map_ | @c void(bool) -- server side peer presence change |
70  * | @c sub_connect_callback_map_ | @c void(bool) -- subscriber side presence change |
71  * | @c req_resp_callback_map_ | @c void(uint64_t, const Bytes&, Bytes*) |
72  * | @c msg_callback_map_ | @c void(const Bytes&) |
73  * | @c intra_msg_callback_map_ | @c void(const IntraData&) |
74  * | @c status_callback_map_ | @c void(const Status::BasePtr&) |
75  *
76  * @par Example
77  * @code
78  * struct TopicObject final : vlink::AbstractObject<std::string> {
79  * using AbstractObject::AbstractObject;
80  * };
81  *
82  * vlink::AbstractFactory<std::string> factory;
83  *
84  * auto object = factory.get_object<TopicObject>("dds://my_topic");
85  * object->add_impl(impl);
86  * object->register_msg_callback(impl, [](const vlink::Bytes& bytes) {
87  * // forward each delivery to the owning node
88  * });
89  *
90  * object->traverse_msg_callback([](vlink::NodeImpl*, const vlink::NodeImpl::MsgCallback& cb) {
91  * cb(payload);
92  * });
93  * @endcode
94  *
95  * @note Every public @c AbstractObject method acquires the internal
96  * @c std::recursive_mutex; callbacks invoked through @c traverse_*() execute
97  * under that lock and may re-enter the same @c AbstractObject safely.
98  *
99  * @tparam FilterT Key type used to identify topics inside the factory map.
100  */
101 
102 #pragma once
103 
104 #include <algorithm>
105 #include <atomic>
106 #include <map>
107 #include <memory>
108 #include <mutex>
109 #include <string>
110 #include <unordered_map>
111 #include <unordered_set>
112 #include <utility>
113 
114 #include "../base/functional.h"
115 #include "../base/logger.h"
116 #include "./node_impl.h"
117 
118 namespace vlink {
119 
120 /**
121  * @class AbstractObject
122  * @brief Topic-scoped fan-out store of @c NodeImpl peers and their callbacks.
123  *
124  * @details
125  * Holds the active @c NodeImpl pointer set together with the six callback
126  * dictionaries documented at file scope and serialises mutation, traversal and
127  * accounting through a @c std::recursive_mutex. The traversal helpers honour
128  * the @c ignore_called() escape hatch so that individual callbacks can opt out
129  * of the "any callback was invoked" accounting tracked by @c has_called().
130  *
131  * @tparam FilterT Key type used by the owning @c AbstractFactory for this object.
132  */
133 template <typename FilterT>
134 class AbstractObject : public AbstractNode {
135  public:
136  using ImplList = std::unordered_set<NodeImpl*>; ///< Set of currently registered @c NodeImpl peers.
137 
138  using ConnectCallbackMap = std::unordered_map<NodeImpl*, NodeImpl::ConnectCallback>; ///< Connect handlers per impl.
140  std::unordered_map<NodeImpl*, NodeImpl::ReqRespCallback>; ///< Req/resp callbacks, keyed by impl.
141  using MsgCallbackMap = std::unordered_map<NodeImpl*, NodeImpl::MsgCallback>; ///< Message callbacks, keyed by impl.
142  using IntraMsgCallbackMap = std::unordered_map<NodeImpl*, NodeImpl::IntraMsgCallback>; ///< Intra-message callbacks.
143  using StatusCallbackMap = std::unordered_map<NodeImpl*, NodeImpl::StatusCallback>; ///< Status callbacks per impl.
144 
146  Function<void(NodeImpl*, const NodeImpl::ConnectCallback&)>; ///< Visitor invoked for each connect entry.
148  Function<void(NodeImpl*, const NodeImpl::ReqRespCallback&)>; ///< Visitor invoked for each req/resp entry.
150  Function<void(NodeImpl*, const NodeImpl::MsgCallback&)>; ///< Visitor invoked for each message entry.
152  Function<void(NodeImpl*, const NodeImpl::IntraMsgCallback&)>; ///< Visitor for each intra-message entry.
154  Function<void(NodeImpl*, const NodeImpl::StatusCallback&)>; ///< Visitor invoked for each status entry.
155 
156  /**
157  * @brief Registers @p impl as an active peer on this topic.
158  *
159  * @details
160  * Inserts @p impl into @c impl_list_ and refreshes the cached @c first_impl_
161  * pointer to the latest registrant. The operation is serialised against all
162  * other public methods.
163  *
164  * @param impl Non-owning peer pointer to track.
165  * @return @c true when the pointer was newly inserted; @c false when it was
166  * already present.
167  */
168  bool add_impl(NodeImpl* impl);
169 
170  /**
171  * @brief Removes @p impl from the peer set and forgets every associated callback.
172  *
173  * @details
174  * Erases the pointer from @c impl_list_, reassigns @c first_impl_ if needed
175  * and drops the entry from all six callback dictionaries. Thread-safe.
176  *
177  * @param impl Peer pointer previously passed to @c add_impl().
178  * @return @c true if @p impl was found and removed; @c false otherwise.
179  */
180  bool remove_impl(NodeImpl* impl);
181 
182  /**
183  * @brief Returns the most recently registered peer.
184  *
185  * @details
186  * The "first" pointer follows the latest successful @c add_impl() call. After
187  * a matching @c remove_impl() the cache is repopulated with an arbitrary
188  * remaining peer, or @c nullptr if the set has been drained.
189  *
190  * @return Pointer to the current cached peer; @c nullptr when no peer is registered.
191  */
192  [[nodiscard]] NodeImpl* get_first_impl() const;
193 
194  /**
195  * @brief Tests whether @p impl is currently part of the peer set.
196  *
197  * @param impl Pointer to query.
198  * @return @c true when @p impl is registered, @c false otherwise.
199  */
200  [[nodiscard]] bool is_contains_impl(NodeImpl* impl) const;
201 
202  /**
203  * @brief Indicates whether at least one peer has been registered.
204  *
205  * @return @c true when @c impl_list_ is non-empty.
206  */
207  [[nodiscard]] bool has_impl() const;
208 
209  /**
210  * @brief Stores @p callback as the server-side connect handler for @p impl.
211  *
212  * @param impl Peer that owns @p callback.
213  * @param callback Callable @c void(bool) invoked when a remote client comes or goes.
214  * @return @c true if the entry was inserted; @c false when one was already present.
215  */
217 
218  /**
219  * @brief Stores @p callback as the subscriber-side connect handler for @p impl.
220  *
221  * @param impl Peer that owns @p callback.
222  * @param callback Callable @c void(bool) invoked when a subscriber appears or disappears.
223  * @return @c true if the entry was inserted; @c false when one was already present.
224  */
226 
227  /**
228  * @brief Stores @p callback as the request/response handler for @p impl.
229  *
230  * @param impl Peer that owns @p callback.
231  * @param callback Callable invoked for every incoming RPC request.
232  * @return @c true on insertion; @c false when already registered.
233  */
235 
236  /**
237  * @brief Stores @p callback as the serialised-message handler for @p impl.
238  *
239  * @param impl Peer that owns @p callback.
240  * @param callback Callable @c void(const Bytes&) invoked for every received message.
241  * @return @c true on insertion; @c false when already registered.
242  */
243  bool register_msg_callback(NodeImpl* impl, NodeImpl::MsgCallback&& callback);
244 
245  /**
246  * @brief Stores @p callback as the intra-process message handler for @p impl.
247  *
248  * @param impl Peer that owns @p callback.
249  * @param callback Callable @c void(const IntraData&) invoked for each in-process delivery.
250  * @return @c true on insertion; @c false when already registered.
251  */
253 
254  /**
255  * @brief Stores @p callback as the transport-status handler for @p impl.
256  *
257  * @param impl Peer that owns @p callback.
258  * @param callback Callable invoked when the transport reports a status change.
259  * @return @c true on insertion; @c false when already registered.
260  */
262 
263  /**
264  * @brief Reports whether the server-connect dictionary is empty.
265  *
266  * @return @c true when no server-connect callbacks are registered.
267  */
268  [[nodiscard]] bool server_connect_map_is_empty() const;
269 
270  /**
271  * @brief Reports whether the subscriber-connect dictionary is empty.
272  *
273  * @return @c true when no subscriber-connect callbacks are registered.
274  */
275  [[nodiscard]] bool sub_connect_map_is_empty() const;
276 
277  /**
278  * @brief Reports whether the request/response dictionary is empty.
279  *
280  * @return @c true when no req/resp callbacks are registered.
281  */
282  [[nodiscard]] bool req_resp_map_is_empty() const;
283 
284  /**
285  * @brief Reports whether the serialised-message dictionary is empty.
286  *
287  * @return @c true when no message callbacks are registered.
288  */
289  [[nodiscard]] bool msg_map_is_empty() const;
290 
291  /**
292  * @brief Reports whether the intra-process message dictionary is empty.
293  *
294  * @return @c true when no intra-message callbacks are registered.
295  */
296  [[nodiscard]] bool intra_msg_map_is_empty() const;
297 
298  /**
299  * @brief Reports whether the transport-status dictionary is empty.
300  *
301  * @return @c true when no status callbacks are registered.
302  */
303  [[nodiscard]] bool status_map_is_empty() const;
304 
305  /**
306  * @brief Walks the server-connect dictionary and invokes @p callback for every entry.
307  *
308  * @details
309  * Iteration is performed while holding the recursive mutex. Individual visits
310  * may call @c ignore_called() to keep the current entry from setting the
311  * @c has_called() flag; iteration carries on regardless.
312  *
313  * @param callback Visitor receiving each peer pointer and its stored handler.
314  */
316 
317  /**
318  * @brief Walks the subscriber-connect dictionary and invokes @p callback for every entry.
319  *
320  * @param callback Visitor receiving each peer pointer and its stored handler.
321  */
323 
324  /**
325  * @brief Walks the request/response dictionary and invokes @p callback for every entry.
326  *
327  * @param callback Visitor receiving each peer pointer and its stored handler.
328  */
329  void traverse_req_resp_callback(const FindReqRespCallback& callback);
330 
331  /**
332  * @brief Walks the serialised-message dictionary and invokes @p callback for every entry.
333  *
334  * @param callback Visitor receiving each peer pointer and its stored handler.
335  */
336  void traverse_msg_callback(const FindMsgCallback& callback);
337 
338  /**
339  * @brief Walks the intra-process dictionary and invokes @p callback for every entry.
340  *
341  * @param callback Visitor receiving each peer pointer and its stored handler.
342  */
344 
345  /**
346  * @brief Walks the transport-status dictionary and invokes @p callback for every entry.
347  *
348  * @param callback Visitor receiving each peer pointer and its stored handler.
349  */
350  void traverse_status_callback(const FindStatusCallback& callback);
351 
352  protected:
354 
355  ~AbstractObject() override;
356 
357  [[nodiscard]] bool has_called() const;
358 
359  void ignore_called();
360 
361  private:
362  template <typename CallbackMapT, typename CallbackT>
363  void traverse_internal_callback(const CallbackMapT& map, const CallbackT& callback);
364 
365  bool has_called_{false};
366  bool ignore_called_{false};
367  ImplList impl_list_;
368  mutable std::recursive_mutex mtx_;
369  ConnectCallbackMap server_connect_callback_map_;
370  ConnectCallbackMap sub_connect_callback_map_;
371  ReqRespCallbackMap req_resp_callback_map_;
372  MsgCallbackMap msg_callback_map_;
373  IntraMsgCallbackMap intra_msg_callback_map_;
374  StatusCallbackMap status_callback_map_;
375  NodeImpl* first_impl_{nullptr};
376 
378 };
379 
380 /**
381  * @class AbstractFactory
382  * @brief Lazily allocates and caches @c AbstractObject instances keyed by @c FilterT.
383  *
384  * @details
385  * Holds a @c std::map<FilterT, std::weak_ptr<Object>> so that multiple node
386  * peers requesting the same key reuse the same registration record. The
387  * returned @c std::shared_ptr carries a custom deleter that erases the map
388  * entry when the last owner releases the object, which keeps the lookup map
389  * free of stale @c weak_ptr slots.
390  *
391  * @note The factory itself is non-copyable and non-movable; share it through a
392  * singleton or a per-transport static instance.
393  *
394  * @tparam FilterT Topic-key type (typically @c std::string).
395  */
396 template <typename FilterT>
399  using Map = std::map<FilterT, std::weak_ptr<Object>>;
400  using Set = std::unordered_set<Object*>;
401 
402  public:
403  /**
404  * @brief Tests whether @p ptr corresponds to a live object created by this factory.
405  *
406  * @param ptr Raw pointer to validate.
407  * @return @c true when the object is still alive in this factory; @c false otherwise.
408  */
409  [[nodiscard]] bool has_object(Object* ptr) const;
410 
411  /**
412  * @brief Looks up or creates the @c ObjectT registered against @p filter.
413  *
414  * @details
415  * If the cached @c weak_ptr is still valid, the existing instance is shared
416  * with the caller. Otherwise a new @c ObjectT is allocated (outside the
417  * factory lock so its constructor can re-enter VLink safely), the resulting
418  * @c shared_ptr is given a deleter that removes the cache entry on
419  * destruction, and the value is stored back into the map.
420  *
421  * @tparam ObjectT Concrete subclass of @c AbstractObject<FilterT> to allocate.
422  *
423  * @param filter Key identifying the topic.
424  * @return Shared ownership handle for the cached object.
425  */
426  template <typename ObjectT>
427  [[nodiscard]] std::shared_ptr<ObjectT> get_object(const FilterT& filter);
428 
429  protected:
430  /**
431  * @brief Constructs an empty factory.
432  */
434 
435  /**
436  * @brief Destroys the factory and releases the cache.
437  */
438  virtual ~AbstractFactory();
439 
440  private:
441  Set set_;
442  Map map_;
443  mutable std::mutex mtx_;
444 
446 };
447 
448 ////////////////////////////////////////////////////////////////
449 /// Details
450 ////////////////////////////////////////////////////////////////
451 
452 template <typename FilterT>
454  std::lock_guard lock(mtx_);
455 
456  first_impl_ = impl;
457 
458  return impl_list_.emplace(impl).second;
459 }
460 
461 template <typename FilterT>
463  std::lock_guard lock(mtx_);
464 
465  if VUNLIKELY (impl_list_.erase(impl) == 0) {
466  return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
467  }
468 
469  if (first_impl_ == impl) {
470  first_impl_ = impl_list_.empty() ? nullptr : *impl_list_.begin();
471  }
472 
473  server_connect_callback_map_.erase(impl);
474  sub_connect_callback_map_.erase(impl);
475  req_resp_callback_map_.erase(impl);
476  msg_callback_map_.erase(impl);
477  intra_msg_callback_map_.erase(impl);
478  status_callback_map_.erase(impl);
479  return true;
480 }
481 
482 template <typename FilterT>
484  std::lock_guard lock(mtx_);
485  return first_impl_;
486 }
487 
488 template <typename FilterT>
490  std::lock_guard lock(mtx_);
491  return impl_list_.find(impl) != impl_list_.end();
492 }
493 
494 template <typename FilterT>
496  std::lock_guard lock(mtx_);
497  return !impl_list_.empty();
498 }
499 
500 template <typename FilterT>
502  NodeImpl::ConnectCallback&& callback) {
503  std::lock_guard lock(this->mtx_);
504  return server_connect_callback_map_.try_emplace(impl, std::move(callback)).second;
505 }
506 
507 template <typename FilterT>
509  NodeImpl::ConnectCallback&& callback) {
510  std::lock_guard lock(this->mtx_);
511  return sub_connect_callback_map_.try_emplace(impl, std::move(callback)).second;
512 }
513 
514 template <typename FilterT>
516  std::lock_guard lock(this->mtx_);
517  return req_resp_callback_map_.try_emplace(impl, std::move(callback)).second;
518 }
519 
520 template <typename FilterT>
522  std::lock_guard lock(this->mtx_);
523  return msg_callback_map_.try_emplace(impl, std::move(callback)).second;
524 }
525 
526 template <typename FilterT>
528  NodeImpl::IntraMsgCallback&& callback) {
529  std::lock_guard lock(this->mtx_);
530  return intra_msg_callback_map_.try_emplace(impl, std::move(callback)).second;
531 }
532 
533 template <typename FilterT>
535  std::lock_guard lock(this->mtx_);
536  return status_callback_map_.try_emplace(impl, std::move(callback)).second;
537 }
538 
539 template <typename FilterT>
541  std::lock_guard lock(this->mtx_);
542  return server_connect_callback_map_.empty();
543 }
544 
545 template <typename FilterT>
547  std::lock_guard lock(this->mtx_);
548  return sub_connect_callback_map_.empty();
549 }
550 
551 template <typename FilterT>
553  std::lock_guard lock(this->mtx_);
554  return req_resp_callback_map_.empty();
555 }
556 
557 template <typename FilterT>
559  std::lock_guard lock(this->mtx_);
560  return msg_callback_map_.empty();
561 }
562 
563 template <typename FilterT>
565  std::lock_guard lock(this->mtx_);
566  return intra_msg_callback_map_.empty();
567 }
568 
569 template <typename FilterT>
571  std::lock_guard lock(this->mtx_);
572  return status_callback_map_.empty();
573 }
574 
575 template <typename FilterT>
577  this->traverse_internal_callback(server_connect_callback_map_, callback);
578 }
579 
580 template <typename FilterT>
582  this->traverse_internal_callback(sub_connect_callback_map_, callback);
583 }
584 
585 template <typename FilterT>
587  this->traverse_internal_callback(req_resp_callback_map_, callback);
588 }
589 
590 template <typename FilterT>
592  this->traverse_internal_callback(msg_callback_map_, callback);
593 }
594 
595 template <typename FilterT>
597  this->traverse_internal_callback(intra_msg_callback_map_, callback);
598 }
599 
600 template <typename FilterT>
602  this->traverse_internal_callback(status_callback_map_, callback);
603 }
604 
605 template <typename FilterT>
606 inline AbstractObject<FilterT>::AbstractObject() = default;
607 
608 template <typename FilterT>
610 
611 template <typename FilterT>
613  return has_called_;
614 }
615 
616 template <typename FilterT>
618  ignore_called_ = true;
619 }
620 
621 template <typename FilterT>
622 template <typename CallbackMapT, typename CallbackT>
623 inline void AbstractObject<FilterT>::traverse_internal_callback(const CallbackMapT& map, const CallbackT& callback) {
624  std::lock_guard lock(mtx_);
625 
626  this->ignore_called_ = false;
627  this->has_called_ = false;
628 
629  for (const auto& [impl, target_callback] : map) {
630  callback(impl, target_callback);
631 
632  if VUNLIKELY (this->ignore_called_) {
633  this->ignore_called_ = false;
634  } else {
635  this->has_called_ = true;
636  }
637  }
638 }
639 
640 template <typename FilterT>
642  std::lock_guard lock(mtx_);
643  return set_.count(ptr) > 0;
644 }
645 
646 template <typename FilterT>
647 template <typename ObjectT>
648 inline std::shared_ptr<ObjectT> AbstractFactory<FilterT>::get_object(const FilterT& filter) {
649  static_assert(std::is_base_of_v<Object, ObjectT>, "ObjectT must be derived from AbstractObject");
650  std::shared_ptr<ObjectT> obj;
651  {
652  std::unique_lock lock(mtx_);
653 
654  const auto& deleter = [this, filter](ObjectT* obj) {
655  {
656  std::lock_guard lock(mtx_);
657 
658  set_.erase(obj);
659 
660  auto iter = map_.find(filter);
661 
662  if VUNLIKELY (iter != map_.end() && iter->second.expired()) {
663  map_.erase(iter);
664  }
665  }
666 
667  delete obj;
668  };
669 
670  auto iter = map_.find(filter);
671 
672  if (iter != map_.end()) {
673  obj = std::static_pointer_cast<ObjectT>(iter->second.lock());
674 
675  if VLIKELY (obj) {
676  return obj;
677  }
678 
679  map_.erase(iter); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
680  }
681 
682  {
683  lock.unlock();
684  auto* obj_ptr = new ObjectT(filter);
685  lock.lock();
686 
687  auto [it, inserted] = map_.try_emplace(filter, std::weak_ptr<Object>());
688 
689  if (!inserted) {
690  obj = std::static_pointer_cast<ObjectT>(it->second.lock()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
691  }
692 
693  if (inserted || !obj) {
694  if (!inserted) {
695  map_.erase(it); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
696  it = map_.try_emplace(filter, std::weak_ptr<Object>()).first; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
697  }
698 
699  obj = std::shared_ptr<ObjectT>(obj_ptr, deleter);
700  it->second = obj;
701  set_.emplace(obj_ptr);
702  } else {
703  delete obj_ptr; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
704  }
705  }
706  return obj;
707  }
708 } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
709 
710 template <typename FilterT>
712 
713 template <typename FilterT>
715 
716 } // namespace vlink
#define VUNLIKELY(...)
Short alias for VLINK_UNLIKELY.
Definition: macros.h:289
#define VLIKELY(...)
Short alias for VLINK_LIKELY.
Definition: macros.h:284
#define VLINK_DISALLOW_COPY_AND_ASSIGN(classname)
Deletes the copy constructor and copy-assignment operator of classname.
Definition: macros.h:174
Foundational base classes shared by every transport-backed VLink node.