add radio, dish, pair

src/adapter_base.hpp

@@ -4,8 +4,6 @@
 #include <boost/signals2/variadic_signal.hpp>
 #include <type_traits>
 
-using namespace boost::signals2;
-
 // Base adapter interface exposing a typed callback signal and name.
 template <typename InType, typename CallbackRetTypeTag, typename... CbkAargs> class AdapterBase {
   using cbk_ret_type_t_ = typename CallbackRetTypeTag::type;
@@ -30,7 +28,7 @@ template <typename InType, typename CallbackRetTypeTag, typename... CbkAargs> cl
 public:
   // Use combiner if return type is not void
   using signature_t = std::conditional_t<std::is_void_v<cbk_ret_type_t_>, void(const InType &, CbkAargs &&...), cbk_ret_type_t_(const InType &, CbkAargs &&...)>;
-  using callback_type_t = std::conditional_t<!std::is_void_v<cbk_ret_type_t_>, signal<signature_t, CollectAllCombiner_>, signal<signature_t>>;
+  using callback_type_t = std::conditional_t<!std::is_void_v<cbk_ret_type_t_>, boost::signals2::signal<signature_t, CollectAllCombiner_>, boost::signals2::signal<signature_t>>;
 
   AdapterBase(const std::string &name) : mc_name_(name) {}
 

src/codecs.cpp

@@ -0,0 +1,34 @@
+#include "codecs.hpp"
+
+template <> std::vector<uint8_t> codecs_s<std::vector<uint8_t>>::encoders_s::from_int(const int32_t &i) {
+  auto str = std::to_string(i);    // Convert to string first
+  return {str.begin(), str.end()}; // String to byte array
+}
+
+template <> auto codecs_s<std::vector<uint8_t>>::encoders_s::from_string(const std::string &s) -> std::vector<uint8_t> { return {s.begin(), s.end()}; };
+
+template <> auto codecs_s<std::vector<uint8_t>>::encoders_s::from_double(const double &d) -> std::vector<uint8_t> {
+  auto str = std::to_string(d);
+  return {str.begin(), str.end()};
+};
+
+template <> auto codecs_s<std::vector<uint8_t>>::decoders_s::to_int(const std::vector<uint8_t> &s) -> int32_t {
+  int32_t ret;
+  auto str = std::string(s.begin(), s.end());
+  if (std::from_chars(str.c_str(), str.c_str() + str.size(), ret).ec == std::errc{}) {
+    return ret;
+  }
+
+  throw std::runtime_error(fmt::format("Invalid convert from {} to integer type", str));
+};
+
+template <> auto codecs_s<std::vector<uint8_t>>::decoders_s::to_string(const std::vector<uint8_t> &i) -> std::string { return std::string(i.begin(), i.end()); };
+template <> auto codecs_s<std::vector<uint8_t>>::decoders_s::to_double(const std::vector<uint8_t> &s) -> double {
+  double ret;
+  auto str = std::string(s.begin(), s.end());
+  if (std::from_chars(str.c_str(), str.c_str() + str.size(), ret).ec == std::errc{}) {
+    return ret;
+  }
+
+  throw std::runtime_error(fmt::format("Invalid convert from {} to double type", str));
+};

src/codecs.hpp

@@ -0,0 +1,26 @@
+#pragma once
+
+#include <cstdint>
+#include <charconv>
+#include <stdexcept>
+#include <string>
+#include <functional>
+
+#define FMT_HEADER_ONLY
+#include <fmt/format.h>
+#include <fmt/ranges.h>
+
+// Simple encoder/decoder set for an environment payload type.
+template<typename EnvDataType> struct codecs_s {
+  struct encoders_s {
+    static auto from_int(const int32_t &) -> EnvDataType;
+    static auto from_string(const std::string &s) -> EnvDataType;
+    static auto from_double(const double &d) -> EnvDataType;
+  } encoders;
+
+  struct decoders_s {
+    static auto to_int(const EnvDataType &s) -> int32_t;
+    static auto to_string(const EnvDataType &i) -> std::string;
+    static auto to_double(const EnvDataType &s) -> double;
+  } decoders;
+};

src/module_factory.hpp

@@ -2,9 +2,59 @@
 
 #include "module.hpp"
 #include <memory>
+#include <string>
+#include <tuple>
+#include <utility>
 
 // Factory for building a Module from owned ports.
 template <typename... Ports>
 auto makeModule(int32_t argc, char **argv, char **envp, const std::string &name, zmq::context_t &zmq_ctx, std::tuple<std::unique_ptr<Ports>...> &&ports) {
   return std::make_unique<Module<Ports...>>(argc, argv, envp, name, zmq_ctx, std::forward<std::tuple<std::unique_ptr<Ports>...>>(ports));
 }
+
+class ModuleBuilderNamed;
+class ModuleBuilderContext;
+template <typename...> class ModuleBuilderPorts;
+
+// Fluent builder to assemble a Module from name/context/ports.
+class ModuleBuilder {
+public:
+  ModuleBuilderNamed withName(std::string name);
+};
+
+class ModuleBuilderNamed {
+public:
+  explicit ModuleBuilderNamed(std::string name) : name_(std::move(name)) {}
+  ModuleBuilderContext withContext(zmq::context_t &zmq_ctx);
+
+private:
+  std::string name_;
+};
+
+class ModuleBuilderContext {
+public:
+  ModuleBuilderContext(std::string name, zmq::context_t &zmq_ctx) : name_(std::move(name)), ctx_(&zmq_ctx) {}
+
+  template <typename... Ports> ModuleBuilderPorts<Ports...> withPorts(std::tuple<std::unique_ptr<Ports>...> &&ports) {
+    return ModuleBuilderPorts<Ports...>(name_, *ctx_, std::forward<std::tuple<std::unique_ptr<Ports>...>>(ports));
+  }
+
+private:
+  std::string name_;
+  zmq::context_t *ctx_;
+};
+
+template <typename... Ports> class ModuleBuilderPorts {
+public:
+  ModuleBuilderPorts(std::string name, zmq::context_t &zmq_ctx, std::tuple<std::unique_ptr<Ports>...> &&ports) : name_(std::move(name)), ctx_(&zmq_ctx), ports_(std::move(ports)) {}
+
+  auto finalize(int32_t argc, char **argv, char **envp) { return makeModule(argc, argv, envp, name_, *ctx_, std::move(ports_)); }
+
+private:
+  std::string name_;
+  zmq::context_t *ctx_;
+  std::tuple<std::unique_ptr<Ports>...> ports_;
+};
+
+inline ModuleBuilderNamed ModuleBuilder::withName(std::string name) { return ModuleBuilderNamed(std::move(name)); }
+inline ModuleBuilderContext ModuleBuilderNamed::withContext(zmq::context_t &zmq_ctx) { return ModuleBuilderContext(std::move(name_), zmq_ctx); }

src/port.hpp

@@ -29,9 +29,9 @@ template <typename... Adapters, typename... Args>
         std::tuple_size_v<args_t<typename std::tuple_element_t<0, std::tuple<Adapters...>>::callback_type_t::signature_type>>) &&
        ...) &&
 
-      // Unique adapters check
+      // All adapters are unique
       ([]<typename... Ts>() consteval {
-        return []<std::size_t... Is>(std::index_sequence<Is...>) consteval {
+        return []<size_t... Is>(std::index_sequence<Is...>) consteval {
           using tuple_t = std::tuple<Ts...>;
           return ((tp::tuple_index<std::tuple_element_t<Is, tuple_t>, tuple_t>::value == Is) && ...);
         }(std::index_sequence_for<Ts...>{});
@@ -53,14 +53,11 @@ public:
           return std::make_tuple([&]<size_t Idx>() {
             using adapter_type_t = std::remove_cvref_t<decltype(*std::get<Idx>(adapters))>;
             using adapter_input_type_t = std::remove_cvref_t<return_type_t<typename adapter_type_t::decoder_type_t>>;
-            using adapter_callback_type_t = std::remove_cvref_t<typename adapter_type_t::base_t::callback_type_t>;
+            using adapter_callback_type_t = std::remove_cvref_t<typename adapter_type_t::base_t::callback_type_t::signature_type>;
 
-            // fmt::print("Adding callback: name: {}, namehash: {}, typehash: {}, cbk_typehash: {}, cbk_type: {}\r\n", std::get<Idx>(adapters)->name(),
-            // std::hash<std::string>()(std::get<Idx>(adapters)->name()), typeid(adapter_input_type_t).hash_code(), typeid(adapter_callback_type_t).hash_code(),
-            // type_name<adapter_callback_type_t>());
             // Cache name, name hash, input type hash, callback type hash, and adapter pointer.
-            return std::make_tuple(std::get<Idx>(adapters)->name(), std::hash<std::string>()(std::get<Idx>(adapters)->name()), typeid(adapter_input_type_t).hash_code(),
+            return std::make_tuple(std::get<Idx>(adapters)->name(), std::hash<std::string>()(std::get<Idx>(adapters)->name()), type_hash<adapter_input_type_t>(),
-                                   typeid(adapter_callback_type_t).hash_code(), std::forward<std::unique_ptr<adapter_type_t>>(std::get<Idx>(adapters)));
+                                   type_hash<adapter_callback_type_t>(), std::forward<std::unique_ptr<adapter_type_t>>(std::get<Idx>(adapters)));
           }.template operator()<Ids>()...);
         }(std::make_index_sequence<sizeof...(Adapters)>{})) {
     // Instantiate the port implementation with any extra args (e.g., SUB topics).
@@ -78,9 +75,9 @@ public:
 
 protected:
   void stop__() const override { m_impl__->stop_source().request_stop(); }
-  void send__(const void *data, size_t size, size_t hash, const std::string &addr = "") const override {
+  void send__(const void *data, size_t size, uint64_t hash, const std::string &addr = "") const override {
-    if (!addr.empty() && this->type() != port_types_e::PUB) {
+    if (!addr.empty() && this->type() != port_types_e::PUB && this->type() != port_types_e::RADIO) {
-      throw std::runtime_error("Addressed send is only supported for PUB ports");
+      throw std::runtime_error("Addressed send is only supported for PUB/RADIO ports");
     }
 
     const void *adapter_ptr = nullptr;
@@ -110,7 +107,7 @@ protected:
     }
   }
 
-  void *get_adapter__(const std::string &name, size_t namehash, size_t typehash, size_t cbk_typehash) const override final {
+  void *get_adapter__(const std::string &name, uint64_t namehash, uint64_t typehash, uint64_t cbk_typehash) const override final {
     void *ret = nullptr;
 
     tp::for_each(mc_adapters_, [&](auto &a) {
@@ -139,12 +136,14 @@ private:
   template <typename... Aargs> class PortImplCallback_<std::tuple<Aargs...>> {
   public:
     PortImplCallback_(const Port *port) : mc_port_(port) {}
-    using type_t = std::function<cbk_return_type_t_(const port_data_type_t &, size_t, Aargs &&...)>;
+    using type_t = std::function<cbk_return_type_t_(const port_data_type_t &, uint64_t, Aargs &&...)>;
 
-    cbk_return_type_t_ operator()(const port_data_type_t &data, size_t hash, Aargs &&...callback_args) const {
+    cbk_return_type_t_ operator()(const port_data_type_t &data, uint64_t hash, Aargs &&...callback_args) const {
       std::conditional_t<!std::is_void_v<cbk_return_type_t_>, cbk_return_type_t_, std::false_type> ret{};
+      bool found = false;
 
       tp::for_each(mc_port_->mc_adapters_, [&](const auto &e) {
+        if (!found) {
           const auto &[adapter_name, adapter_namehash, adapter_typehash, adapter_cbk_typehash, adapter] = e;
           if (adapter_typehash == hash) {
             // Decode payload and dispatch to the adapter's signal.
@@ -153,9 +152,16 @@ private:
             } else {
               ret = adapter->callback()(adapter->decoder()(data), std::forward<Aargs>(callback_args)...);
             }
+
+            found = true;
+          }
         }
       });
 
+      if (!found) {
+        throw std::runtime_error(fmt::format("Adapter with type hash: {} not found", hash));
+      }
+
       if constexpr (!std::is_void_v<cbk_return_type_t_>) {
         return ret;
       }
@@ -166,7 +172,7 @@ private:
   };
 
   mutable std::unique_ptr<PortImplBase<this_t, PortImplCallback_<callback_aargs_t>>> m_impl__{nullptr};
-  mutable std::tuple<std::tuple<std::string, size_t, size_t, size_t, std::unique_ptr<Adapters>>...> mc_adapters_;
+  mutable std::tuple<std::tuple<std::string, uint64_t, uint64_t, uint64_t, std::unique_ptr<Adapters>>...> mc_adapters_;
 
   template <typename... ImplArgs> void init_impl_(enum port_types_e pt, ImplArgs &&...args) const {
     using enum port_types_e;
@@ -182,7 +188,8 @@ private:
         std::make_tuple(make_null_impl_pair.template operator()<UNKNOWN>(), make_null_impl_pair.template operator()<PUB>(), make_null_impl_pair.template operator()<SUB>(),
                         make_null_impl_pair.template operator()<REQ>(), make_null_impl_pair.template operator()<REP>(), make_null_impl_pair.template operator()<ROUTER>(),
                         make_null_impl_pair.template operator()<DEALER>(), make_null_impl_pair.template operator()<PUSH>(), make_null_impl_pair.template operator()<PULL>(),
-                        make_null_impl_pair.template operator()<PAIR>());
+                        make_null_impl_pair.template operator()<PAIR_CLIENT>(), make_null_impl_pair.template operator()<PAIR_SERVER>(),
+                        make_null_impl_pair.template operator()<RADIO>(), make_null_impl_pair.template operator()<DISH>());
 
     tp::for_each(impl_map, [&](const auto &p) {
       const auto &[type, null_pimpl] = p;

src/port_base.hpp

@@ -3,8 +3,10 @@
 #include "src/tuple.hpp"
 #include <boost/callable_traits/args.hpp>
 #include <boost/callable_traits/return_type.hpp>
+#include <boost/signals2/signal.hpp>
 #include <msgpack.hpp>
 #include <tuple>
+#include <type_traits>
 
 #define ZMQ_BUILD_DRAFT_API
 #include <zmq.hpp>
@@ -26,7 +28,7 @@ template <typename EncodedType> class PortBase {
   public:
     AddressedPort_(const PortBase<EncodedType> *port, const std::string &address) : mc_addr_(address), mc_port_(port) {}
     template <typename InType> const auto &operator<<(const InType &in) const {
-      mc_port_->send__(&in, sizeof(InType), typeid(InType).hash_code(), mc_addr_);
+      mc_port_->send__(&in, sizeof(InType), type_hash<std::remove_cvref_t<InType>>(), mc_addr_);
       return *this;
     }
 
@@ -50,7 +52,7 @@ public:
 
   template <typename InType> const PortBase<EncodedType> &operator<<(const InType &in) const {
     // Use empty address for non-addressed sends.
-    send__(&in, sizeof(InType), typeid(InType).hash_code());
+    send__(&in, sizeof(InType), type_hash<std::remove_cvref_t<InType>>());
     return *this;
   }
 
@@ -62,14 +64,14 @@ public:
 
 protected:
   virtual void stop__() const = 0;
-  virtual void send__(const void *data, size_t size, size_t type_hash, const std::string &addr = "") const = 0;
+  virtual void send__(const void *data, size_t size, uint64_t type_hash, const std::string &addr = "") const = 0;
-  virtual void *get_adapter__(const std::string &name, size_t namehash, size_t typehash, size_t cbk_typehash) const = 0;
+  virtual void *get_adapter__(const std::string &name, uint64_t namehash, uint64_t typehash, uint64_t cbk_typehash) const = 0;
 
 private:
   const enum port_types_e mc_type_;
   const std::string mc_name_;
   const std::map<std::string, std::string> mc_endpoints_;
-  const size_t mc_name_hash_;
+  const uint64_t mc_name_hash_;
 
   // Type-safe callback lookup helper.
   template <typename...> class GetCallbackHelper_;
@@ -79,11 +81,13 @@ private:
 
     template <typename Signature> auto &operator()(const std::string &name) const {
       using ret_type_t = std::remove_cvref_t<return_type_t<std::function<Signature>>>;
-      using arg_type_t = std::tuple_element_t<0, args_t<std::function<Signature>>>;
+      using arg_type_t = std::remove_cvref_t<std::tuple_element_t<0, args_t<std::function<Signature>>>>;
+      using cbk_type_t = typename AdapterBase<arg_type_t, tag_s<ret_type_t>, Aargs...>::callback_type_t::signature_type;
+
       return (*static_cast<AdapterBase<arg_type_t, tag_s<ret_type_t>, Aargs...> *>(
-                  mc_port_->get_adapter__(name, std::hash<std::string>()(name), typeid(arg_type_t).hash_code(),
+                  mc_port_->get_adapter__(name, std::hash<std::string>()(name), type_hash<arg_type_t>(), type_hash<std::remove_cvref_t<cbk_type_t>>())))
-                                          typeid(typename AdapterBase<arg_type_t, tag_s<ret_type_t>, Aargs...>::callback_type_t).hash_code())))
           .callback();
+      // typeid(typename AdapterBase<arg_type_t, tag_s<ret_type_t>, Aargs...>::callback_type_t).hash_code()
     }
 
   private:

src/port_dealer_impl.hpp

@@ -2,6 +2,8 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <chrono>
+#include <thread>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>

src/port_dish_impl.hpp

@@ -0,0 +1,84 @@
+#pragma once
+
+#include "port_impl_base.hpp"
+#include "port_types.hpp"
+#include <chrono>
+#include <list>
+#include <optional>
+
+#define FMT_HEADER_ONLY
+#include <fmt/format.h>
+#include <fmt/ranges.h>
+
+// Dish transport: connects, joins groups, and dispatches payloads with group name.
+template <typename Port, typename Callback> class PortImpl<port_types_e::DISH, Port, Callback> : public PortImplBase<Port, Callback> {
+public:
+  using base_t = PortImplBase<Port, Callback>;
+  PortImpl(const Port *port, zmq::context_t &zmq_ctx, const std::map<std::string, std::string> &endpoints, const std::list<std::string> &groups, Callback &&callback)
+      : PortImplBase<Port, Callback>(port, zmq_ctx, endpoints, std::forward<Callback>(callback)), mc_groups_(groups) {
+    this->m_sock__ = zmq::socket_t(this->m_ctx__, zmq::socket_type::dish);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.connect(ep);
+    }
+
+    // Join each group.
+    for (const auto &group : mc_groups_) {
+      this->m_sock__.join(group.c_str());
+    }
+
+    // Start async listener loop.
+    listen__(this->stop_source().get_token());
+  }
+
+  // Send to socket depending on implementation
+  void send(const msgpack::sbuffer &data, const std::string &addr = "") const override { throw std::runtime_error("Can't send anything on DISH pattern socket"); };
+
+private:
+  const std::list<std::string> mc_groups_;
+
+  void listen__(std::stop_token st) const override {
+    this->m_listener_thread__ = std::async(
+        std::launch::async,
+        [this](std::stop_token st) {
+          try {
+            zmq::poller_t poller;
+            poller.add(this->m_sock__, zmq::event_flags::pollin);
+
+            while (!st.stop_requested()) {
+              std::vector<zmq::poller_event<zmq::no_user_data>> events(1u);
+              size_t num_events = poller.wait_all(events, std::chrono::milliseconds(base_t::sc_recv_timeout_ms__));
+
+              for (int32_t i = 0; i < num_events; ++i) {
+                zmq::message_t msg;
+
+                this->m_sock__.recv(msg, zmq::recv_flags::dontwait).and_then([&](const auto &res) {
+                  std::string group = std::string(msg.group());
+                  typename base_t::port_payload_s payload;
+
+                  msgpack::sbuffer buf;
+                  buf.write(reinterpret_cast<const char *>(msg.data()), msg.size());
+                  const auto &[typehash, batch] = msgpack::unpack(buf.data(), buf.size()).get().convert(payload);
+
+                  // Dispatch each decoded item.
+                  for (const auto &data : batch) {
+                    this->mc_cbk__(data, typehash, group);
+                  }
+
+                  return std::optional(res);
+                });
+              }
+            }
+          } catch (const zmq::error_t &) {
+            if (!st.stop_requested()) {
+              throw;
+            }
+          } catch (...) {
+            if (!st.stop_requested()) {
+              throw;
+            }
+          }
+        },
+        st);
+  }
+};

src/port_factory.hpp

@@ -1,6 +1,10 @@
 #pragma once
 
 #include "port.hpp"
+#include <map>
+#include <string>
+#include <tuple>
+#include <utility>
 
 // Factory for building a Port from adapters and extra args.
 template <typename... Adapters, typename... Args>
@@ -9,3 +13,107 @@ auto makePort(enum port_types_e pt, const std::string &name, const std::map<std:
   return std::make_unique<Port<std::tuple<Adapters...>, std::tuple<Args...>>>(pt, name, endpoints, zmq_ctx, std::forward<std::tuple<std::unique_ptr<Adapters>...>>(adapters),
                                                                               std::forward<std::tuple<Args...>>(args));
 }
+
+class PortBuilderNamed;
+class PortBuilderEndpoints;
+class PortBuilderContext;
+class PortBuilderWithContext;
+
+template <typename...> class PortBuilderAdapters;
+template <typename...> class PortBuilderArgs;
+
+class PortBuilderNamed {
+public:
+  explicit PortBuilderNamed(port_types_e pt) : m_pt_(pt) {}
+  PortBuilderEndpoints withName(const std::string &name);
+
+private:
+  port_types_e m_pt_;
+};
+
+// Fluent builder to assemble a Port from type/name/endpoints/context/adapters/args.
+class PortBuilder {
+public:
+  inline PortBuilderNamed withType(port_types_e pt) { return PortBuilderNamed(pt); }
+};
+
+class PortBuilderWithContext {
+public:
+  PortBuilderWithContext(port_types_e pt, const std::string &name, const std::map<std::string, std::string> &endpoints, zmq::context_t &zmq_ctx)
+      : m_pt_(pt), m_name_(name), m_endpoints_(endpoints), m_ctx_(&zmq_ctx) {}
+
+  template <typename... Adapters> inline PortBuilderAdapters<std::tuple<Adapters...>> withAdapters(std::tuple<std::unique_ptr<Adapters>...> &&adapters) {
+    return PortBuilderAdapters<std::tuple<Adapters...>>(m_pt_, m_name_, m_endpoints_, *m_ctx_, std::forward<std::tuple<std::unique_ptr<Adapters>...>>(adapters));
+  }
+
+private:
+  port_types_e m_pt_;
+  std::string m_name_;
+  std::map<std::string, std::string> m_endpoints_;
+  zmq::context_t *m_ctx_;
+};
+
+class PortBuilderContext {
+public:
+  PortBuilderContext(port_types_e pt, const std::string &name, const std::map<std::string, std::string> &endpoints) : m_pt_(pt), m_name_(name), m_endpoints_(endpoints) {}
+  PortBuilderWithContext withContext(zmq::context_t &zmq_ctx);
+
+private:
+  port_types_e m_pt_;
+  std::string m_name_;
+  std::map<std::string, std::string> m_endpoints_;
+};
+
+class PortBuilderEndpoints {
+public:
+  PortBuilderEndpoints(port_types_e pt, const std::string &name) : m_pt_(pt), m_name_(name) {}
+  PortBuilderContext withEndpoints(const std::map<std::string, std::string> &endpoints);
+
+private:
+  port_types_e m_pt_;
+  std::string m_name_;
+};
+
+template <typename... Adapters> class PortBuilderAdapters<std::tuple<Adapters...>> {
+public:
+  PortBuilderAdapters(port_types_e pt, const std::string &name, const std::map<std::string, std::string> &endpoints, zmq::context_t &zmq_ctx,
+                      std::tuple<std::unique_ptr<Adapters>...> &&adapters)
+      : m_pt_(pt), m_name_(name), m_endpoints_(endpoints), m_ctx_(&zmq_ctx), m_adapters_(&adapters) {}
+
+  template <typename... Args> PortBuilderArgs<std::tuple<Adapters...>, std::tuple<Args...>> withArgs(std::tuple<Args...> &&args) {
+    return PortBuilderArgs<std::tuple<Adapters...>, std::tuple<Args...>>(m_pt_, m_name_, m_endpoints_, *m_ctx_, std::forward<std::tuple<std::unique_ptr<Adapters>...>>(*m_adapters_),
+                                                 std::forward<std::tuple<Args...>>(args));
+  }
+
+  auto finalize() { return makePort(m_pt_, m_name_, m_endpoints_, *m_ctx_, std::forward<std::tuple<std::unique_ptr<Adapters>...>>(*m_adapters_)); }
+
+private:
+  port_types_e m_pt_;
+  std::string m_name_;
+  std::map<std::string, std::string> m_endpoints_;
+  zmq::context_t *m_ctx_;
+  std::tuple<std::unique_ptr<Adapters>...> *m_adapters_;
+};
+
+template <typename... Adapters, typename... Args> class PortBuilderArgs<std::tuple<Adapters...>, std::tuple<Args...>> {
+public:
+  PortBuilderArgs(port_types_e pt, const std::string &name, const std::map<std::string, std::string> &endpoints, zmq::context_t &zmq_ctx,
+                  std::tuple<std::unique_ptr<Adapters>...> &&adapters, std::tuple<Args...> &&args)
+      : m_pt_(pt), m_name_(name), m_endpoints_(endpoints), m_ctx_(&zmq_ctx), m_adapters_(&adapters), m_args_(std::forward<std::tuple<Args...>>(args)) {}
+
+  inline auto finalize() {
+    return makePort(m_pt_, m_name_, m_endpoints_, *m_ctx_, std::forward<std::tuple<std::unique_ptr<Adapters>...>>(*m_adapters_), std::forward<std::tuple<Args...>>(m_args_));
+  }
+
+private:
+  port_types_e m_pt_;
+  std::string m_name_;
+  std::map<std::string, std::string> m_endpoints_;
+  zmq::context_t *m_ctx_;
+  std::tuple<std::unique_ptr<Adapters>...> *m_adapters_;
+  std::tuple<Args...> m_args_;
+};
+
+inline PortBuilderEndpoints PortBuilderNamed::withName(const std::string &name) { return PortBuilderEndpoints(m_pt_, name); }
+inline PortBuilderWithContext PortBuilderContext::withContext(zmq::context_t &zmq_ctx) { return PortBuilderWithContext(m_pt_, m_name_, m_endpoints_, zmq_ctx); }
+inline PortBuilderContext PortBuilderEndpoints::withEndpoints(const std::map<std::string, std::string> &endpoints) { return PortBuilderContext(m_pt_, m_name_, endpoints); }

src/port_impl_base.hpp

@@ -16,7 +16,7 @@ template <typename Port, typename Callback> class PortImplBase {
 public:
   // Wire payload: type hash + batch of encoded values.
   struct port_payload_s {
-    size_t typehash;
+    uint64_t typehash;
     std::vector<port_data_type_t_> data;
     MSGPACK_DEFINE(typehash, data);
   };
@@ -27,14 +27,7 @@ public:
 
   virtual void send(const msgpack::sbuffer &data, const std::string &addr = "") const = 0;
   void close() {
-    // Close socket first to break any pending waits.
+    // Join listener thread before closing the socket to avoid polling on a closed fd.
-    try {
-      m_sock__.close();
-    } catch (...) {
-      // Ignore close errors during shutdown.
-    }
-
-    // Join listener thread if one was started.
     if (m_listener_thread__.valid()) {
       try {
         m_listener_thread__.get();
@@ -42,6 +35,12 @@ public:
         // Ignore listener exceptions during shutdown.
       }
     }
+
+    try {
+      m_sock__.close();
+    } catch (...) {
+      // Ignore close errors during shutdown.
+    }
   };
 
   inline auto &stop_source() { return m_ss_; }

src/port_pair_impl.hpp

@@ -2,14 +2,87 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <chrono>
+#include <optional>
+#include <vector>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-// Pair transport placeholder; currently no full send/recv support.
+// Pair transport (client): connects and listens for incoming payloads.
-template <typename Port, typename Callback> class PortImpl<port_types_e::PAIR, Port, Callback> : public PortImplBase<Port, Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::PAIR_CLIENT, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
+  using base_t = PortImplBase<Port, Callback>;
+  PortImpl(const Port *port, zmq::context_t &zmq_ctx, const std::map<std::string, std::string> &endpoints, Callback &&callback)
+      : PortImplBase<Port, Callback>(port, zmq_ctx, endpoints, std::forward<Callback>(callback)) {
+    this->m_sock__ = zmq::socket_t(this->m_ctx__, zmq::socket_type::pair);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.connect(ep);
+    }
+
+    listen__(this->stop_source().get_token());
+  }
+
+  // Send to socket depending on implementation
+  void send(const msgpack::sbuffer &data, const std::string &addr = "") const override {
+    try {
+      this->m_sock__.send(zmq::message_t(data.data(), data.size()), zmq::send_flags::none);
+    } catch (const zmq::error_t &err) {
+      fmt::print("ZMQ error: {1} ({0})\r\n", err.num(), err.what());
+    }
+  };
+
+private:
+  void listen__(std::stop_token st) const override {
+    this->m_listener_thread__ = std::async(
+        std::launch::async,
+        [this](std::stop_token st) {
+          try {
+            zmq::poller_t poller;
+            poller.add(this->m_sock__, zmq::event_flags::pollin);
+
+            while (!st.stop_requested()) {
+              std::vector<zmq::poller_event<zmq::no_user_data>> events(1u);
+              size_t num_events = poller.wait_all(events, std::chrono::milliseconds(base_t::sc_recv_timeout_ms__));
+
+              for (int32_t i = 0; i < num_events; ++i) {
+                zmq::message_t msg;
+
+                this->m_sock__.recv(msg, zmq::recv_flags::dontwait).and_then([&](const auto &res) {
+                  typename base_t::port_payload_s payload;
+
+                  msgpack::sbuffer buf;
+                  buf.write(reinterpret_cast<const char *>(msg.data()), msg.size());
+                  const auto &[typehash, batch] = msgpack::unpack(buf.data(), buf.size()).get().convert(payload);
+
+                  for (const auto &data : batch) {
+                    this->mc_cbk__(data, typehash);
+                  }
+
+                  return std::optional(res);
+                });
+              }
+            }
+          } catch (const zmq::error_t &) {
+            if (!st.stop_requested()) {
+              throw;
+            }
+          } catch (...) {
+            if (!st.stop_requested()) {
+              throw;
+            }
+          }
+        },
+        st);
+  }
+};
+
+// Pair transport (server): binds and listens for incoming payloads.
+template <typename Port, typename Callback> class PortImpl<port_types_e::PAIR_SERVER, Port, Callback> : public PortImplBase<Port, Callback> {
+public:
+  using base_t = PortImplBase<Port, Callback>;
   PortImpl(const Port *port, zmq::context_t &zmq_ctx, const std::map<std::string, std::string> &endpoints, Callback &&callback)
       : PortImplBase<Port, Callback>(port, zmq_ctx, endpoints, std::forward<Callback>(callback)) {
     this->m_sock__ = zmq::socket_t(this->m_ctx__, zmq::socket_type::pair);
@@ -17,15 +90,60 @@ public:
     for (const auto &[_, ep] : this->mc_endpoints__) {
       this->m_sock__.bind(ep);
     }
+
+    listen__(this->stop_source().get_token());
   }
 
   // Send to socket depending on implementation
-  void send(const msgpack::sbuffer &data, const std::string &addr = "") const override {};
+  void send(const msgpack::sbuffer &data, const std::string &addr = "") const override {
+    try {
+      this->m_sock__.send(zmq::message_t(data.data(), data.size()), zmq::send_flags::none);
+    } catch (const zmq::error_t &err) {
+      fmt::print("ZMQ error: {1} ({0})\r\n", err.num(), err.what());
+    }
+  };
 
 private:
   void listen__(std::stop_token st) const override {
+    this->m_listener_thread__ = std::async(
+        std::launch::async,
+        [this](std::stop_token st) {
+          try {
+            zmq::poller_t poller;
+            poller.add(this->m_sock__, zmq::event_flags::pollin);
+
             while (!st.stop_requested()) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+              std::vector<zmq::poller_event<zmq::no_user_data>> events(1u);
+              size_t num_events = poller.wait_all(events, std::chrono::milliseconds(base_t::sc_recv_timeout_ms__));
+
+              for (int32_t i = 0; i < num_events; ++i) {
+                zmq::message_t msg;
+
+                this->m_sock__.recv(msg, zmq::recv_flags::dontwait).and_then([&](const auto &res) {
+                  typename base_t::port_payload_s payload;
+
+                  msgpack::sbuffer buf;
+                  buf.write(reinterpret_cast<const char *>(msg.data()), msg.size());
+                  const auto &[typehash, batch] = msgpack::unpack(buf.data(), buf.size()).get().convert(payload);
+
+                  for (const auto &data : batch) {
+                    this->mc_cbk__(data, typehash);
                   }
+
+                  return std::optional(res);
+                });
+              }
+            }
+          } catch (const zmq::error_t &) {
+            if (!st.stop_requested()) {
+              throw;
+            }
+          } catch (...) {
+            if (!st.stop_requested()) {
+              throw;
+            }
+          }
+        },
+        st);
   }
 };

src/port_pull_impl.hpp

@@ -3,6 +3,7 @@
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
 #include <chrono>
+#include <optional>
 #include <zmq.hpp>
 
 #define FMT_HEADER_ONLY

src/port_radio_impl.hpp

@@ -0,0 +1,39 @@
+#pragma once
+
+#include "port_impl_base.hpp"
+#include "port_types.hpp"
+#include <string>
+
+#define FMT_HEADER_ONLY
+#include <fmt/format.h>
+#include <fmt/ranges.h>
+
+// Radio transport: binds endpoints and sends group-tagged payloads.
+template <typename Port, typename Callback> class PortImpl<port_types_e::RADIO, Port, Callback> final : public PortImplBase<Port, Callback> {
+public:
+  PortImpl(const Port *port, zmq::context_t &zmq_ctx, const std::map<std::string, std::string> &endpoints, Callback &&callback)
+      : PortImplBase<Port, Callback>(port, zmq_ctx, endpoints, std::forward<Callback>(callback)) {
+    this->m_sock__ = zmq::socket_t(this->m_ctx__, zmq::socket_type::radio);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.bind(ep);
+    }
+  }
+
+  // Send to socket depending on implementation
+  void send(const msgpack::sbuffer &data, const std::string &addr = "") const override {
+    try {
+      zmq::message_t msg(data.data(), data.size());
+      if (!addr.empty()) {
+        msg.set_group(addr.c_str());
+      }
+
+      this->m_sock__.send(msg, zmq::send_flags::none);
+    } catch (const zmq::error_t &err) {
+      fmt::print("ZMQ error: {1} ({0})\r\n", err.num(), err.what());
+    }
+  };
+
+private:
+  void listen__(std::stop_token st) const override { throw std::runtime_error("Can't listen on RADIO pattern socket"); }
+};

src/port_rep_impl.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <boost/callable_traits/return_type.hpp>
+#include <cstdint>
 #include <optional>
 #include <stdexcept>
 #include <tuple>
@@ -12,6 +13,7 @@
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
 #include "tuple.hpp"
+#include "utils.hpp"
 
 // Reply transport: listens for requests and sends reply payloads.
 template <typename Port, typename Callback> class PortImpl<port_types_e::REP, Port, Callback> : public PortImplBase<Port, Callback> {
@@ -24,6 +26,8 @@ public:
       this->m_sock__.bind(ep);
     }
 
+    this->m_sock__.set(zmq::sockopt::sndtimeo, static_cast<int32_t>(base_t::sc_recv_timeout_ms__));
+
     // Start async listener loop.
     listen__(this->stop_source().get_token());
   }
@@ -42,7 +46,7 @@ private:
 
             while (!st.stop_requested()) {
               std::vector<zmq::poller_event<zmq::no_user_data>> events(1u);
-              size_t num_events = poller.wait_all(events, std::chrono::milliseconds(base_t::sc_recv_timeout_ms__));
+              uint64_t num_events = poller.wait_all(events, std::chrono::milliseconds(base_t::sc_recv_timeout_ms__));
 
               for (int32_t i = 0; i < num_events; ++i) {
                 zmq::message_t msg;
@@ -59,13 +63,13 @@ private:
                     if (batch.size()) {
                       auto reply_data = this->mc_cbk__(batch.front(), typehash);
                       typename base_t::port_payload_s reply_payload = {
-                          .typehash = typeid(typename decltype(reply_data)::value_type).hash_code(),
+                          .typehash = type_hash<std::remove_cvref_t<typename decltype(reply_data)::value_type>>(),
                       };
 
                       for (const auto &d : reply_data) {
                         using adapter_in_type_t = std::remove_cvref_t<decltype(d)>;
 
-                        size_t typehash = typeid(adapter_in_type_t).hash_code();
+                        uint64_t typehash = type_hash<adapter_in_type_t>();
 
                         // Find matching encoder by type to build reply payload.
                         tp::for_each(this->mc_port__->adapters(), [&](const auto &e) {

src/port_req_impl.hpp

@@ -2,6 +2,8 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <cstdint>
+#include <optional>
 #include <stdexcept>
 
 #define FMT_HEADER_ONLY
@@ -21,7 +23,7 @@ public:
     }
 
     // Avoid blocking forever if no REP is available.
-    this->m_sock__.set(zmq::sockopt::rcvtimeo, static_cast<int>(base_t::sc_recv_timeout_ms__));
+    this->m_sock__.set(zmq::sockopt::rcvtimeo, static_cast<int32_t>(base_t::sc_recv_timeout_ms__));
   }
 
   void send(const msgpack::sbuffer &data, const std::string &addr = "") const override {
@@ -44,6 +46,7 @@ public:
       });
     } catch (const zmq::error_t &err) {
       fmt::print("ZMQ error: {1} ({0})\r\n", err.num(), err.what());
+      throw;
     }
   };
 

src/port_router_impl.hpp

@@ -2,6 +2,8 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <chrono>
+#include <thread>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>

src/port_subscriber_impl.hpp

@@ -3,6 +3,7 @@
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
 #include <chrono>
+#include <optional>
 #include <ranges>
 #include <stop_token>
 #include <sys/socket.h>

src/port_types.hpp

@@ -11,7 +11,8 @@ enum class port_types_e : uint32_t {
   REP,
   PUSH,
   PULL,
-  PAIR,
+  PAIR_CLIENT,
+  PAIR_SERVER,
   RADIO,
   DISH,
 
@@ -31,7 +32,8 @@ template<> struct endpoints_s<port_types_e::REQ> : endpoints_base_s {};
 template<> struct endpoints_s<port_types_e::REP> : endpoints_base_s {};
 template<> struct endpoints_s<port_types_e::PUSH> : endpoints_base_s {};
 template<> struct endpoints_s<port_types_e::PULL> : endpoints_base_s {};
-template<> struct endpoints_s<port_types_e::PAIR> : endpoints_base_s {};
+template<> struct endpoints_s<port_types_e::PAIR_CLIENT> : endpoints_base_s {};
+template<> struct endpoints_s<port_types_e::PAIR_SERVER> : endpoints_base_s {};
 template<> struct endpoints_s<port_types_e::RADIO> : endpoints_base_s {};
 template<> struct endpoints_s<port_types_e::DISH> : endpoints_base_s {};
 template<> struct endpoints_s<port_types_e::ROUTER> : endpoints_base_s {};

src/ports.hpp

@@ -10,4 +10,5 @@
 #include "port_router_impl.hpp"
 #include "port_dealer_impl.hpp"
 #include "port_pair_impl.hpp"
-
+#include "port_radio_impl.hpp"
+#include "port_dish_impl.hpp"

src/templates.cpp

@@ -2,9 +2,11 @@
 #include <boost/signals2/variadic_signal.hpp>
 #include <chrono>
 #include <cstdint>
+#include <list>
 #include <string>
 #include <thread>
 #include <tuple>
+#include <unordered_map>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
@@ -126,28 +128,85 @@ void rep_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<
   });
 }
 
+void pair_server_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  // Resolve port by name.
+  const auto &pair = *ports.at("pair_server_port");
+
+  auto &int_cbk = pair.callback<void(const int32_t &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = pair.callback<void(const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = pair.callback<void(const double &)>("double-vector<uint8_t>-double");
+
+  int_cbk.connect([](const int32_t &i) { fmt::print("PAIR server: got data: {} of {} type\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>()); });
+  string_cbk.connect([](const std::string &s) { fmt::print("PAIR server: got data: {} of {} type\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>()); });
+  double_cbk.connect([](const double &d) { fmt::print("PAIR server: got data: {} of {} type\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>()); });
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+  pair << 4 << 5 << double{6.f} << std::string("server->client");
+}
+
+void pair_client_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  // Resolve port by name.
+  const auto &pair = *ports.at("pair_client_port");
+
+  auto &int_cbk = pair.callback<void(const int32_t &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = pair.callback<void(const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = pair.callback<void(const double &)>("double-vector<uint8_t>-double");
+
+  int_cbk.connect([](const int32_t &i) { fmt::print("PAIR client: got data: {} of {} type\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>()); });
+  string_cbk.connect([](const std::string &s) { fmt::print("PAIR client: got data: {} of {} type\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>()); });
+  double_cbk.connect([](const double &d) { fmt::print("PAIR client: got data: {} of {} type\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>()); });
+
+  pair << 1 << 2 << double{3.f} << std::string("test");
+}
+
+void dish_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  // Resolve port by name.
+  const auto &dish = *ports.at("dish_port");
+
+  auto &int_cbk = dish.callback<void(const int32_t &, const std::string &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = dish.callback<void(const std::string &, const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = dish.callback<void(const double &, const std::string &)>("double-vector<uint8_t>-double");
+
+  int_cbk.connect([](const int32_t &i, const std::string &group) {
+    fmt::print("DISH socket: got data: {} of {} type from group: {}\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>(), group);
+  });
+
+  string_cbk.connect([](const std::string &s, const std::string &group) {
+    fmt::print("DISH socket: got data: {} of {} type from group: {}\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>(), group);
+  });
+
+  double_cbk.connect([](const double &d, const std::string &group) {
+    fmt::print("DISH socket: got data: {} of {} type from group: {}\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>(), group);
+  });
+}
+
+void radio_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  // Resolve port by name.
+  const auto &radio = *ports.at("radio_port");
+
+  radio["grp0"] << 1 << 2 << double{3.f};
+  radio["grp1"] << std::string("test");
+}
+
 int main(int argc, char *argv[], char *envp[]) {
   using enum port_types_e;
   codecs_s<env_data_type_t> codecs;
   // Shared ZMQ context for in-process transports.
   zmq::context_t zmq_ctx; // Use common context because both modules have in-process ports, but working in different threads
-  auto a = AdapterBuilder();
-  auto b = AdapterBuilder()
-               .encodeDataBy(&codecs.encoders.from_int)
-               .decodeDataBy(&codecs.decoders.to_int)
-               .withCallbackSignature<void(const int32_t &)>()
-               .withName("string-vector<uint8_t>-string");
 
   // Make module that contains only 1 port working on PUBLISHER pattern
-  // TODO: merge all builders to one complex "Module" builder
+  auto publisher_module = ModuleBuilder()
-  auto publisher_module = makeModule(argc, argv, envp, "publisher_module", zmq_ctx,
+                              .withName("publisher_module")
-                                     std::tuple{
+                              .withContext(zmq_ctx)
-                                         makePort(PUB, "publisher_port",
+                              .withPorts(std::tuple{
-                                                  {
+                                  PortBuilder()
+                                      .withType(PUB)
+                                      .withName("publisher_port")
+                                      .withEndpoints({
                                           {"test", "inproc://PUB-SUB"},
-                                                  } /* This port will publish messages here  */,
+                                      })
-                                                  zmq_ctx,
+                                      .withContext(zmq_ctx)
-                                                  std::tuple{
+                                      .withAdapters(std::tuple{
                                           AdapterBuilder()
                                               .encodeDataBy(&codecs.encoders.from_int)
                                               .decodeDataBy(&codecs.decoders.to_int)
@@ -158,7 +217,7 @@ int main(int argc, char *argv[], char *envp[]) {
                                           AdapterBuilder()
                                               .encodeDataBy(&codecs.encoders.from_string)
                                               .decodeDataBy(&codecs.decoders.to_string)
-                                                          .withCallbackSignature<void(const std::string)>()
+                                              .withCallbackSignature<void(const std::string &)>()
                                               .withName("string-vector<uint8_t>-string")
                                               .finalize(),
 
@@ -168,18 +227,24 @@ int main(int argc, char *argv[], char *envp[]) {
                                               .withCallbackSignature<void(const double &)>()
                                               .withName("double-vector<uint8_t>-double")
                                               .finalize(),
-                                                  }),
+                                      })
-                                     });
+                                      .finalize(),
+                              })
+                              .finalize(argc, argv, envp);
 
   // Make module that contains only 1 port working on SUBSCRIBER pattern
-  auto subscriber_module = makeModule(argc, argv, envp, "subscriber_module", zmq_ctx,
+  auto subscriber_module = ModuleBuilder()
-                                      std::tuple{
+                               .withName("subscriber_module")
-                                          makePort(SUB, "subscriber_port",
+                               .withContext(zmq_ctx)
-                                                   {
+                               .withPorts(std::tuple{
+                                   PortBuilder()
+                                       .withType(SUB)
+                                       .withName("subscriber_port")
+                                       .withEndpoints({
                                            {"test", "inproc://PUB-SUB"},
-                                                   } /* this port will read data here  */,
+                                       })
-                                                   zmq_ctx,
+                                       .withContext(zmq_ctx)
-                                                   std::tuple{
+                                       .withAdapters(std::tuple{
                                            AdapterBuilder()
                                                .encodeDataBy(&codecs.encoders.from_int)
                                                .decodeDataBy(&codecs.decoders.to_int)
@@ -200,23 +265,26 @@ int main(int argc, char *argv[], char *envp[]) {
                                                .withCallbackSignature<void(const double &, const std::string &)>()
                                                .withName("double-vector<uint8_t>-double")
                                                .finalize(),
-                                                   },
+                                       })
-
+                                       .withArgs(std::tuple{
-                                                   // This type of port requires arguments - topics to subscribe
-                                                   std::tuple{
-                                                       // Topics to subscribe
                                            std::list<std::string>{"topic0", "topic1", "topic2", "topic3"},
-                                                   }),
+                                       })
-                                      });
+                                       .finalize(),
+                               })
+                               .finalize(argc, argv, envp);
 
-  auto pusher_module = makeModule(argc, argv, envp, "pusher_module", zmq_ctx,
+  auto pusher_module = ModuleBuilder()
-                                  std::tuple{
+                           .withName("pusher_module")
-                                      makePort(PUSH, "push_port",
+                           .withContext(zmq_ctx)
-                                               {
+                           .withPorts(std::tuple{
+                               PortBuilder()
+                                   .withType(PUSH)
+                                   .withName("push_port")
+                                   .withEndpoints({
                                        {"test", "inproc://PUSH-PULL"},
-                                               } /* This port will publish messages here  */,
+                                   })
-                                               zmq_ctx,
+                                   .withContext(zmq_ctx)
-                                               std::tuple{
+                                   .withAdapters(std::tuple{
                                        AdapterBuilder()
                                            .encodeDataBy(&codecs.encoders.from_int)
                                            .decodeDataBy(&codecs.decoders.to_int)
@@ -227,7 +295,7 @@ int main(int argc, char *argv[], char *envp[]) {
                                        AdapterBuilder()
                                            .encodeDataBy(&codecs.encoders.from_string)
                                            .decodeDataBy(&codecs.decoders.to_string)
-                                                       .withCallbackSignature<void(const std::string)>()
+                                           .withCallbackSignature<void(const std::string &)>()
                                            .withName("string-vector<uint8_t>-string")
                                            .finalize(),
 
@@ -237,17 +305,23 @@ int main(int argc, char *argv[], char *envp[]) {
                                            .withCallbackSignature<void(const double &)>()
                                            .withName("double-vector<uint8_t>-double")
                                            .finalize(),
-                                               }),
+                                   })
-                                  });
+                                   .finalize(),
+                           })
+                           .finalize(argc, argv, envp);
 
-  auto puller_module = makeModule(argc, argv, envp, "puller_module", zmq_ctx,
+  auto puller_module = ModuleBuilder()
-                                  std::tuple{
+                           .withName("puller_module")
-                                      makePort(PULL, "pull_port",
+                           .withContext(zmq_ctx)
-                                               {
+                           .withPorts(std::tuple{
+                               PortBuilder()
+                                   .withType(PULL)
+                                   .withName("pull_port")
+                                   .withEndpoints({
                                        {"test", "inproc://PUSH-PULL"},
-                                               } /* This port will publish messages here  */,
+                                   })
-                                               zmq_ctx,
+                                   .withContext(zmq_ctx)
-                                               std::tuple{
+                                   .withAdapters(std::tuple{
                                        AdapterBuilder()
                                            .encodeDataBy(&codecs.encoders.from_int)
                                            .decodeDataBy(&codecs.decoders.to_int)
@@ -258,7 +332,7 @@ int main(int argc, char *argv[], char *envp[]) {
                                        AdapterBuilder()
                                            .encodeDataBy(&codecs.encoders.from_string)
                                            .decodeDataBy(&codecs.decoders.to_string)
-                                                       .withCallbackSignature<void(const std::string)>()
+                                           .withCallbackSignature<void(const std::string &)>()
                                            .withName("string-vector<uint8_t>-string")
                                            .finalize(),
 
@@ -268,17 +342,23 @@ int main(int argc, char *argv[], char *envp[]) {
                                            .withCallbackSignature<void(const double &)>()
                                            .withName("double-vector<uint8_t>-double")
                                            .finalize(),
-                                               }),
+                                   })
-                                  });
+                                   .finalize(),
+                           })
+                           .finalize(argc, argv, envp);
 
-  auto req_module = makeModule(argc, argv, envp, "req_module", zmq_ctx,
+  auto req_module = ModuleBuilder()
-                               std::tuple{
+                        .withName("req_module")
-                                   makePort(REQ, "req_port",
+                        .withContext(zmq_ctx)
-                                            {
+                        .withPorts(std::tuple{
+                            PortBuilder()
+                                .withType(REQ)
+                                .withName("req_port")
+                                .withEndpoints({
                                     {"test", "inproc://REQ-REP"},
-                                            } /* This port will publish messages here  */,
+                                })
-                                            zmq_ctx,
+                                .withContext(zmq_ctx)
-                                            std::tuple{
+                                .withAdapters(std::tuple{
                                     AdapterBuilder()
                                         .encodeDataBy(&codecs.encoders.from_int)
                                         .decodeDataBy(&codecs.decoders.to_int)
@@ -289,7 +369,7 @@ int main(int argc, char *argv[], char *envp[]) {
                                     AdapterBuilder()
                                         .encodeDataBy(&codecs.encoders.from_string)
                                         .decodeDataBy(&codecs.decoders.to_string)
-                                                    .withCallbackSignature<void(const std::string)>()
+                                        .withCallbackSignature<void(const std::string &)>()
                                         .withName("string-vector<uint8_t>-string")
                                         .finalize(),
 
@@ -299,17 +379,23 @@ int main(int argc, char *argv[], char *envp[]) {
                                         .withCallbackSignature<void(const double &)>()
                                         .withName("double-vector<uint8_t>-double")
                                         .finalize(),
-                                            }),
+                                })
-                               });
+                                .finalize(),
+                        })
+                        .finalize(argc, argv, envp);
 
-  auto rep_module = makeModule(argc, argv, envp, "rep_module", zmq_ctx,
+  auto rep_module = ModuleBuilder()
-                               std::tuple{
+                        .withName("rep_module")
-                                   makePort(REP, "rep_port",
+                        .withContext(zmq_ctx)
-                                            {
+                        .withPorts(std::tuple{
+                            PortBuilder()
+                                .withType(REP)
+                                .withName("rep_port")
+                                .withEndpoints({
                                     {"test", "inproc://REQ-REP"},
-                                            } /* This port will publish messages here  */,
+                                })
-                                            zmq_ctx,
+                                .withContext(zmq_ctx)
-                                            std::tuple{
+                                .withAdapters(std::tuple{
                                     AdapterBuilder()
                                         .encodeDataBy(&codecs.encoders.from_int)
                                         .decodeDataBy(&codecs.decoders.to_int)
@@ -330,8 +416,161 @@ int main(int argc, char *argv[], char *envp[]) {
                                         .withCallbackSignature<std::string(const double &)>()
                                         .withName("double-vector<uint8_t>-double")
                                         .finalize(),
-                                            }),
+                                })
-                               });
+                                .finalize(),
+                        })
+                        .finalize(argc, argv, envp);
+
+  auto pair_server_module = ModuleBuilder()
+                                .withName("pair_server_module")
+                                .withContext(zmq_ctx)
+                                .withPorts(std::tuple{
+                                    PortBuilder()
+                                        .withType(PAIR_SERVER)
+                                        .withName("pair_server_port")
+                                        .withEndpoints({
+                                            {"test", "inproc://PAIR"},
+                                        })
+                                        .withContext(zmq_ctx)
+                                        .withAdapters(std::tuple{
+                                            AdapterBuilder()
+                                                .encodeDataBy(&codecs.encoders.from_int)
+                                                .decodeDataBy(&codecs.decoders.to_int)
+                                                .withCallbackSignature<void(const int32_t &)>()
+                                                .withName("int-vector<uint8_t>-int")
+                                                .finalize(),
+
+                                            AdapterBuilder()
+                                                .encodeDataBy(&codecs.encoders.from_string)
+                                                .decodeDataBy(&codecs.decoders.to_string)
+                                                .withCallbackSignature<void(const std::string &)>()
+                                                .withName("string-vector<uint8_t>-string")
+                                                .finalize(),
+
+                                            AdapterBuilder()
+                                                .encodeDataBy(&codecs.encoders.from_double)
+                                                .decodeDataBy(&codecs.decoders.to_double)
+                                                .withCallbackSignature<void(const double &)>()
+                                                .withName("double-vector<uint8_t>-double")
+                                                .finalize(),
+                                        })
+                                        .finalize(),
+                                })
+                                .finalize(argc, argv, envp);
+
+  auto pair_client_module = ModuleBuilder()
+                                .withName("pair_client_module")
+                                .withContext(zmq_ctx)
+                                .withPorts(std::tuple{
+                                    PortBuilder()
+                                        .withType(PAIR_CLIENT)
+                                        .withName("pair_client_port")
+                                        .withEndpoints({
+                                            {"test", "inproc://PAIR"},
+                                        })
+                                        .withContext(zmq_ctx)
+                                        .withAdapters(std::tuple{
+                                            AdapterBuilder()
+                                                .encodeDataBy(&codecs.encoders.from_int)
+                                                .decodeDataBy(&codecs.decoders.to_int)
+                                                .withCallbackSignature<void(const int32_t &)>()
+                                                .withName("int-vector<uint8_t>-int")
+                                                .finalize(),
+
+                                            AdapterBuilder()
+                                                .encodeDataBy(&codecs.encoders.from_string)
+                                                .decodeDataBy(&codecs.decoders.to_string)
+                                                .withCallbackSignature<void(const std::string &)>()
+                                                .withName("string-vector<uint8_t>-string")
+                                                .finalize(),
+
+                                            AdapterBuilder()
+                                                .encodeDataBy(&codecs.encoders.from_double)
+                                                .decodeDataBy(&codecs.decoders.to_double)
+                                                .withCallbackSignature<void(const double &)>()
+                                                .withName("double-vector<uint8_t>-double")
+                                                .finalize(),
+                                        })
+                                        .finalize(),
+                                })
+                                .finalize(argc, argv, envp);
+
+  auto dish_module = ModuleBuilder()
+                         .withName("dish_module")
+                         .withContext(zmq_ctx)
+                         .withPorts(std::tuple{
+                             PortBuilder()
+                                 .withType(DISH)
+                                 .withName("dish_port")
+                                 .withEndpoints({
+                                     {"test", "inproc://RADIO-DISH"},
+                                 })
+                                 .withContext(zmq_ctx)
+                                 .withAdapters(std::tuple{
+                                     AdapterBuilder()
+                                         .encodeDataBy(&codecs.encoders.from_int)
+                                         .decodeDataBy(&codecs.decoders.to_int)
+                                         .withCallbackSignature<void(const int32_t &, const std::string &)>()
+                                         .withName("int-vector<uint8_t>-int")
+                                         .finalize(),
+
+                                     AdapterBuilder()
+                                         .encodeDataBy(&codecs.encoders.from_string)
+                                         .decodeDataBy(&codecs.decoders.to_string)
+                                         .withCallbackSignature<void(const std::string &, const std::string &)>()
+                                         .withName("string-vector<uint8_t>-string")
+                                         .finalize(),
+
+                                     AdapterBuilder()
+                                         .encodeDataBy(&codecs.encoders.from_double)
+                                         .decodeDataBy(&codecs.decoders.to_double)
+                                         .withCallbackSignature<void(const double &, const std::string &)>()
+                                         .withName("double-vector<uint8_t>-double")
+                                         .finalize(),
+                                 })
+                                 .withArgs(std::tuple{
+                                     std::list<std::string>{"grp0", "grp1"},
+                                 })
+                                 .finalize(),
+                         })
+                         .finalize(argc, argv, envp);
+
+  auto radio_module = ModuleBuilder()
+                          .withName("radio_module")
+                          .withContext(zmq_ctx)
+                          .withPorts(std::tuple{
+                              PortBuilder()
+                                  .withType(RADIO)
+                                  .withName("radio_port")
+                                  .withEndpoints({
+                                      {"test", "inproc://RADIO-DISH"},
+                                  })
+                                  .withContext(zmq_ctx)
+                                  .withAdapters(std::tuple{
+                                      AdapterBuilder()
+                                          .encodeDataBy(&codecs.encoders.from_int)
+                                          .decodeDataBy(&codecs.decoders.to_int)
+                                          .withCallbackSignature<void(const int32_t &)>()
+                                          .withName("int-vector<uint8_t>-int")
+                                          .finalize(),
+
+                                      AdapterBuilder()
+                                          .encodeDataBy(&codecs.encoders.from_string)
+                                          .decodeDataBy(&codecs.decoders.to_string)
+                                          .withCallbackSignature<void(const std::string &)>()
+                                          .withName("string-vector<uint8_t>-string")
+                                          .finalize(),
+
+                                      AdapterBuilder()
+                                          .encodeDataBy(&codecs.encoders.from_double)
+                                          .decodeDataBy(&codecs.decoders.to_double)
+                                          .withCallbackSignature<void(const double &)>()
+                                          .withName("double-vector<uint8_t>-double")
+                                          .finalize(),
+                                  })
+                                  .finalize(),
+                          })
+                          .finalize(argc, argv, envp);
 
   fmt::print("\r\nPUB-SUB test:\r\n");
   subscriber_module->run(subscriber_entry); // Subscribe and get data
@@ -347,6 +586,16 @@ int main(int argc, char *argv[], char *envp[]) {
   rep_module->run(rep_entry);
   req_module->run(req_entry);
   std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+  fmt::print("\r\nPAIR test:\r\n");
+  pair_server_module->run(pair_server_entry);
+  pair_client_module->run(pair_client_entry);
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+  fmt::print("\r\nRADIO-DISH test:\r\n");
+  dish_module->run(dish_entry);
+  radio_module->run(radio_entry);
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
   fmt::print("DONE!\r\n");
 
   return 0;

src/tuple.hpp

@@ -24,7 +24,7 @@ template <typename Tuple, typename Func, std::size_t... I> constexpr Func for_ea
 }
 
 template <typename... Ts, typename Function, size_t... Is> auto transform_impl(std::tuple<Ts...> const &inputs, Function function, std::index_sequence<Is...> is) {
-  return std::tuple<std::result_of_t<Function(Ts)>...>{function(std::get<Is>(inputs))...};
+  return std::tuple<std::invoke_result_t<Function(Ts)>...>{function(std::get<Is>(inputs))...};
 }
 
 template <typename... T, std::size_t... i> auto subtuple(const std::tuple<T...> &t, std::index_sequence<i...>) { return std::make_tuple(std::get<i>(t)...); }

src/utils.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <cstdint>
 #include <string_view>
 
 // Trait to detect template specializations.
@@ -25,6 +26,19 @@ template <class T> constexpr std::string_view type_name() {
 #endif
 }
 
+// Compile-time FNV-1a hash for stable type IDs across builds.
+constexpr uint64_t fnv1a_64(std::string_view sv) {
+  uint64_t hash = 14695981039346656037ull;
+  for (const char &c : sv) {
+    hash ^= static_cast<uint8_t>(c);
+    hash *= 1099511628211ull;
+  }
+
+  return hash;
+}
+
+template <typename T> constexpr uint64_t type_hash() { return fnv1a_64(type_name<T>()); }
+
 // Type tag wrapper used to pass types through templates.
 template <typename T> struct tag_s {
   using type = T;