adding compiler

.clangd

@@ -1,2 +1,2 @@
 CompileFlags:
-  Add: ['-std=gnu++23']
+  Add: ['-std=gnu++23', -I/opt/cling/include]

CMakeLists.txt

@@ -19,6 +19,14 @@ FetchContent_Declare(
 )
 
 FetchContent_MakeAvailable(inja)
+# FetchContent_Declare(
+#   json
+#   GIT_REPOSITORY https://github.com/nlohmann/json
+#   GIT_PROGRESS   TRUE
+#   GIT_TAG        main
+# )
+
+# FetchContent_MakeAvailable(json)
 
 FetchContent_Declare(
   libzmq

src/adapter.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "src/tuple.hpp"
 #include <boost/callable_traits.hpp>
 #include <boost/callable_traits/args.hpp>
 #include <boost/callable_traits/return_type.hpp>
@@ -16,23 +17,25 @@
 
 using namespace boost::callable_traits;
 
-template <typename Encoder, typename Decoder>
-  requires(
-      // Encoder return type is same as Decoder argument type, encoder and decoder have only 1 arg
-      std::tuple_size_v<args_t<Decoder>> == 1 && std::is_same_v<return_type_t<Encoder>, std::remove_cvref_t<std::tuple_element_t<0, args_t<Decoder>>>> &&
-      std::tuple_size_v<args_t<Encoder>> == 1 && std::is_same_v<return_type_t<Decoder>, std::remove_cvref_t<std::tuple_element_t<0, args_t<Encoder>>>>)
+template<typename ...> class Adapter;
+template <typename Encoder, typename Decoder, typename CallbackRetTypeTag, typename... CbkAargs>
+  requires(std::tuple_size_v<args_t<Decoder>> == 1 &&                                                               // Decoder has only 1 argument
+           std::is_same_v<return_type_t<Encoder>, std::remove_cvref_t<std::tuple_element_t<0, args_t<Decoder>>>> && // Encoder return type is same with decoder argument
+           std::tuple_size_v<args_t<Encoder>> == 1 &&                                                               // Encoder accepts only one argument
+           std::is_same_v<return_type_t<Decoder>, std::remove_cvref_t<std::tuple_element_t<0, args_t<Encoder>>>>    // Decoder return type is same with encoder argument
+           )
 
-class Adapter : public AdapterBase<std::tuple_element_t<0, boost::callable_traits::args_t<Encoder>>> {
+class Adapter<Encoder, Decoder, CallbackRetTypeTag, std::tuple<CbkAargs...>> : public AdapterBase<std::tuple_element_t<0, args_t<Encoder>>, CallbackRetTypeTag, CbkAargs...> {
 public:
   using encoder_type_t = Encoder;
   using decoder_type_t = Decoder;
-  using adapter_type_t = Adapter;
-  using callback_arg_type_t = std::tuple_element_t<0, boost::callable_traits::args_t<Encoder>>;
-  using base_t = AdapterBase<callback_arg_type_t>;
-  using callback_type_t = base_t::callback_type_t;
 
-  Adapter(const std::string &name, std::pair<Encoder, Decoder> &&fns)
-      : AdapterBase<callback_arg_type_t>(name), mc_enc_(std::forward<Encoder>(fns.first)), mc_dec_(std::forward<Decoder>(fns.second)) {}
+  using callback_arg_type_t = std::tuple_element_t<0, args_t<Encoder>>;
+  using base_t = AdapterBase<callback_arg_type_t, CallbackRetTypeTag, CbkAargs...>;
+
+  Adapter(const std::string &name, std::tuple<Encoder, Decoder, CallbackRetTypeTag> &&fns)
+      : AdapterBase<callback_arg_type_t, CallbackRetTypeTag, CbkAargs...>(name), mc_enc_(std::forward<Encoder>(std::get<0u>(fns))),
+        mc_dec_(std::forward<Decoder>(std::get<1u>(fns))) {}
 
   inline const auto &encoder() const { return mc_enc_; }
   inline const auto &decoder() const { return mc_dec_; }
@@ -42,5 +45,5 @@ public:
 private:
   const Encoder mc_enc_;
   const Decoder mc_dec_;
-  mutable callback_type_t m_callback_;
+  mutable typename base_t::callback_type_t m_callback_;
 };

src/adapter_base.hpp

@@ -1,18 +1,42 @@
 #pragma once
 
+#include "utils.hpp"
 #include <boost/signals2.hpp>
+#include <boost/signals2/variadic_signal.hpp>
+#include <type_traits>
+
+using namespace boost::signals2;
+
+template <typename InType, typename CallbackRetTypeTag, typename... CbkAargs> class AdapterBase {
+  using cbk_ret_type_t_ = typename CallbackRetTypeTag::type;
+
+  // If return type is not void we combine all callback ivocation results to std::vector
+  class CollectAllCombiner_ {
+  public:
+    using result_type = std::conditional_t<!std::is_void_v<cbk_ret_type_t_>, std::vector<cbk_ret_type_t_>, std::false_type>;
+
+    template <typename InputIterator> result_type operator()(InputIterator first, InputIterator last) const {
+      result_type results;
+      if constexpr (!std::is_same_v<result_type, std::false_type>) {
+        for (; first != last; ++first) {
+          results.push_back(*first);
+        }
+      }
+
+      return results;
+    }
+  };
 
-template <typename InType> class AdapterBase {
 public:
-  using callback_type_t = boost::signals2::signal<void(const InType &)>;
+  // 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>>;
 
   AdapterBase(const std::string &name) : mc_name_(name) {}
 
   virtual callback_type_t &callback() const = 0;
   inline const auto &name() const { return mc_name_; }
 
-protected:
 private:
   const std::string mc_name_;
 };
-

src/adapter_factory.hpp

@@ -1,7 +1,68 @@
 #pragma once
 
-#include "adapter.hpp"
+#include <boost/callable_traits/args.hpp>
+#include <boost/callable_traits/return_type.hpp>
+#include <tuple>
 
-template <typename Encoder, typename Decoder> auto makeAdapter(const std::string &name, std::pair<Encoder, Decoder> &&fns) {
-  return std::make_unique<Adapter<Encoder, Decoder>>(name, std::forward<std::pair<Encoder, Decoder>>(fns));
-}
+#include "adapter.hpp"
+#include "src/tuple.hpp"
+using namespace boost::callable_traits;
+
+template <typename...> class AdapterBuilder;
+template <typename...> class AdapterBuilderNamed;
+
+template <> class AdapterBuilder<> {
+public:
+  template <typename Encoder> AdapterBuilder<Encoder> encodeDataBy(Encoder &&encoder) { return AdapterBuilder<Encoder>(std::forward<Encoder>(encoder)); }
+};
+
+template <typename Encoder> class AdapterBuilder<Encoder> {
+public:
+  AdapterBuilder(Encoder &&encoder) : m_encoder_ref_(std::forward<Encoder>(encoder)) {}
+  template <typename Decoder> AdapterBuilder<Encoder, Decoder> decodeDataBy(Decoder &&decoder) {
+    return AdapterBuilder<Encoder, Decoder>(std::forward<Encoder>(m_encoder_ref_), std::forward<Decoder>(decoder));
+  }
+
+private:
+  Encoder &&m_encoder_ref_;
+};
+
+template <typename Encoder, typename Decoder> class AdapterBuilder<Encoder, Decoder> {
+public:
+  AdapterBuilder(Encoder &&encoder, Decoder &&decoder) : m_encoder_ref_(std::forward<Encoder>(encoder)), m_decoder_ref_(std::forward<Decoder>(decoder)) {}
+  template <typename CallbackSignature> AdapterBuilder<Encoder, Decoder, CallbackSignature> withCallbackSignature() {
+    return AdapterBuilder<Encoder, Decoder, CallbackSignature>(std::forward<Encoder>(m_encoder_ref_), std::forward<Decoder>(m_decoder_ref_));
+  }
+
+private:
+  Encoder &&m_encoder_ref_;
+  Decoder &&m_decoder_ref_;
+};
+
+template <typename Encoder, typename Decoder, typename CallbackSignature> class AdapterBuilder<Encoder, Decoder, CallbackSignature> {
+public:
+  AdapterBuilder(Encoder &&encoder, Decoder &&decoder) : m_encoder_ref_(std::forward<Encoder>(encoder)), m_decoder_ref_(std::forward<Decoder>(decoder)) {}
+  AdapterBuilderNamed<Encoder, Decoder, CallbackSignature> withName(const std::string &name) {
+    return AdapterBuilderNamed<Encoder, Decoder, CallbackSignature>(std::forward<Encoder>(m_encoder_ref_), std::forward<Decoder>(m_decoder_ref_), name);
+  }
+
+private:
+  Encoder &&m_encoder_ref_;
+  Decoder &&m_decoder_ref_;
+};
+
+template <typename Encoder, typename Decoder, typename CallbackSignature> class AdapterBuilderNamed<Encoder, Decoder, CallbackSignature> {
+public:
+  AdapterBuilderNamed(Encoder &&encoder, Decoder &&decoder, const std::string &name)
+      : m_encoder_ref_(std::forward<Encoder>(encoder)), m_decoder_ref_(std::forward<Decoder>(decoder)), mc_name_(name) {}
+
+  auto finalize() {
+    return std::make_unique<Adapter<Encoder, Decoder, tag_s<return_type_t<CallbackSignature>>, typename tp::tuple_tail<args_t<CallbackSignature>>::type>>(
+        mc_name_, std::make_tuple(std::forward<Encoder>(m_encoder_ref_), std::forward<Decoder>(m_decoder_ref_), tag_s<return_type_t<CallbackSignature>>{}));
+  }
+
+private:
+  const std::string &mc_name_;
+  Encoder &&m_encoder_ref_;
+  Decoder &&m_decoder_ref_;
+};

src/module.hpp

@@ -4,7 +4,6 @@
 #include <tuple>
 #include <unordered_map>
 #include <utility>
-#include <zmq.hpp>
 
 #include "port_base.hpp"
 #include "tuple.hpp"

src/port.hpp

@@ -5,60 +5,83 @@
 #include <boost/callable_traits/return_type.hpp>
 #include <cstddef>
 #include <stdexcept>
+#include <tuple>
+#include <type_traits>
+#include <utility>
 
 #include "port_base.hpp"
 #include "ports.hpp"
+#include "src/port_types.hpp"
 #include "tuple.hpp"
 
 using namespace boost::callable_traits;
 
 template <typename...> class Port;
 template <typename... Adapters, typename... Args>
+  requires(
+      (std::is_same_v<typename Adapters::base_t::callback_type_t::result_type, typename std::tuple_element_t<0, std::tuple<Adapters...>>::base_t::callback_type_t::result_type> &&
+       ...) &&
+      ((std::tuple_size_v<args_t<typename Adapters::base_t::callback_type_t::signature_type>> ==
+        std::tuple_size_v<args_t<typename std::tuple_element_t<0, std::tuple<Adapters...>>::callback_type_t::signature_type>>) &&
+       ...))
+
 class Port<std::tuple<Adapters...>, std::tuple<Args...>> : public PortBase<std::tuple_element_t<0, std::tuple<return_type_t<typename Adapters::encoder_type_t>...>>> {
 public:
+  static constexpr auto callback_args_num = std::tuple_size_v<args_t<typename std::tuple_element_t<0, std::tuple<Adapters...>>::callback_type_t::signature_type>>;
+  using this_t = Port<std::tuple<Adapters...>, std::tuple<Args...>>;
   using port_data_type_t = std::tuple_element_t<0, std::tuple<return_type_t<typename Adapters::encoder_type_t>...>>;
+  using callback_aargs_t = tp::tuple_tail<args_t<typename std::tuple_element_t<0, std::tuple<Adapters...>>::callback_type_t::signature_type>>::type;
 
-  Port(enum port_types_e pt, const std::string &name, const std::string &endpoint, zmq::context_t &zmq_ctx, std::tuple<std::unique_ptr<Adapters>...> &&adapters,
-       std::tuple<Args...> &&args)
-      : PortBase<port_data_type_t>(pt, name, endpoint, zmq_ctx),
+  Port(enum 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)
+      : PortBase<port_data_type_t>(pt, name, endpoints, zmq_ctx),
 
         // Init adapters
         mc_adapters_([&]<size_t... Ids>(std::index_sequence<Ids...>) {
           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 = return_type_t<typename adapter_type_t::decoder_type_t>;
+            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>;
 
-            return std::make_tuple(std::get<Idx>(adapters)->name(), std::hash<std::string>()(std::get<Idx>(adapters)->name()),
-                                   typeid(std::remove_cvref_t<adapter_input_type_t>).hash_code(), std::forward<std::unique_ptr<adapter_type_t>>(std::get<Idx>(adapters)));
+            // 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>());
+            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(),
+                                   typeid(adapter_callback_type_t).hash_code(), std::forward<std::unique_ptr<adapter_type_t>>(std::get<Idx>(adapters)));
           }.template operator()<Ids>()...);
         }(std::make_index_sequence<sizeof...(Adapters)>{})) {
-    std::apply([&, this](auto &&...args) { init_impl_(pt, zmq_ctx, std::move(endpoint), std::forward<Args>(args)...); }, std::forward<decltype(args)>(args));
+    std::apply([&, this](auto &&...args) { init_impl_(pt, zmq_ctx, std::move(endpoints), std::forward<Args>(args)...); }, std::forward<decltype(args)>(args));
   }
 
+  ~Port() override { stop(); }
+
   void stop() const override {
-    this->stop__();
-    mc_impl_->close();
+    stop__();
+    m_impl__->close();
   }
 
+  inline const auto &adapters() const { return mc_adapters_; }
+
 protected:
+  void stop__() const override { m_impl__->stop_source().request_stop(); }
   void send__(const std::string &addr, const void *data, size_t size, size_t hash) const override {
     const void *adapter_ptr = nullptr;
 
     tp::for_each(mc_adapters_, [&](const auto &e) {
-      const auto &[adapter_name, adapter_namehash, adapter_typehash, adapter] = e;
+      const auto &[adapter_name, adapter_namehash, adapter_typehash, adapter_cbk_typehash, adapter] = e;
 
       if (adapter_typehash == hash) {
         using adapter_in_type_t = std::remove_cvref_t<return_type_t<decltype(adapter->decoder())>>;
         adapter_ptr = &adapter;
 
-        typename PortImplBase<cbk_type_t_>::port_payload_s payload = {
+        typename PortImplBase<this_t, PortImplCallback_<callback_aargs_t>>::port_payload_s payload = {
             .typehash = hash,
-            .data = adapter->encoder()(*reinterpret_cast<const adapter_in_type_t *>(data)),
+            .data = {adapter->encoder()(*reinterpret_cast<const adapter_in_type_t *>(data))},
         };
 
         msgpack::sbuffer buf;
         msgpack::pack(buf, payload);
-        mc_impl_->send(addr, buf);
+        m_impl__->send(addr, buf);
       }
     });
 
@@ -67,20 +90,21 @@ protected:
     }
   }
 
-  void *get_adapter__(const std::string &name, size_t namehash, size_t typehash) const override final {
+  void *get_adapter__(const std::string &name, size_t namehash, size_t typehash, size_t cbk_typehash) const override final {
     void *ret = nullptr;
 
     tp::for_each(mc_adapters_, [&](auto &a) {
       if (!ret) {
-        auto &[adapter_name, adapter_namehash, adapter_typehash, adapter] = a;
-        if (adapter_typehash == typehash && adapter_namehash == namehash) {
+        auto &[adapter_name, adapter_namehash, adapter_typehash, adapter_cbk_typehash, adapter] = a;
+        if (adapter_typehash == typehash && adapter_namehash == namehash && adapter_cbk_typehash == cbk_typehash) {
           ret = reinterpret_cast<void *>(adapter.get());
         }
       }
     });
 
     if (!ret) {
-      throw std::runtime_error(fmt::format("No adapter '{}' in port '{}'\r\n", name, this->name()));
+      throw std::runtime_error(
+          fmt::format("No such callback in adapter '{}' in port '{}' (namehash: #{}, typehash: #{}, cbk_typehash: #{})\r\n", name, this->name(), namehash, typehash, cbk_typehash));
     }
 
     return ret;
@@ -88,20 +112,47 @@ protected:
 
 private:
   using base_t_ = PortBase<port_data_type_t>;
-  using cbk_type_t_ = std::function<void(const port_data_type_t &, size_t)>;
+  using cbk_return_type_t_ = typename std::tuple_element_t<0, std::tuple<Adapters...>>::base_t::callback_type_t::result_type;
 
-  mutable std::tuple<std::tuple<std::string, size_t, size_t, std::unique_ptr<Adapters>>...> mc_adapters_;
-  mutable std::unique_ptr<PortImplBase<cbk_type_t_>> mc_impl_;
+  template <typename...> class PortImplCallback_;
+  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 &&...)>;
 
-  void listen__(std::stop_token st) const override { mc_impl_->listen(st); }
+    cbk_return_type_t_ operator()(const port_data_type_t &data, size_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;
+
+      tp::for_each(mc_port_->mc_adapters_, [&](const auto &e) {
+        const auto &[adapter_name, adapter_namehash, adapter_typehash, adapter_cbk_typehash, adapter] = e;
+        if (adapter_typehash == hash) {
+          if constexpr (std::is_void_v<std::invoke_result_t<std::remove_cvref_t<decltype(adapter->callback())>, decltype(adapter->decoder()(data)), Aargs &&...>>) {
+            adapter->callback()(adapter->decoder()(data), std::forward<Aargs>(callback_args)...);
+          } else {
+            ret = adapter->callback()(adapter->decoder()(data), std::forward<Aargs>(callback_args)...);
+          }
+        }
+      });
+
+      if constexpr (!std::is_void_v<cbk_return_type_t_>) {
+        return ret;
+      }
+    }
+
+  private:
+    const Port *mc_port_;
+  };
+
+  mutable std::unique_ptr<PortImplBase<this_t, PortImplCallback_<callback_aargs_t>>> m_impl__;
+  mutable std::tuple<std::tuple<std::string, size_t, size_t, size_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;
     static constexpr auto make_null_impl_pair = []<enum port_types_e port_type>() consteval {
       if constexpr (port_type == UNKNOWN) {
-        return std::make_pair(port_type, static_cast<PortImplBase<cbk_type_t_> *>(nullptr));
+        return std::make_pair(port_type, static_cast<PortImplBase<this_t, PortImplCallback_<callback_aargs_t>> *>(nullptr));
       } else {
-        return std::make_pair(port_type, static_cast<PortImpl<port_type, cbk_type_t_> *>(nullptr));
+        return std::make_pair(port_type, static_cast<PortImpl<port_type, this_t, PortImplCallback_<callback_aargs_t>> *>(nullptr));
       }
     };
 
@@ -116,20 +167,8 @@ private:
       if (type == pt) {
         using impl_type_t = std::remove_pointer_t<decltype(null_pimpl)>;
 
-        if constexpr (std::is_constructible_v<impl_type_t, const ImplArgs &..., cbk_type_t_ &&>) {
-          mc_impl_ = std::make_unique<impl_type_t>(
-              // Args
-              std::forward<ImplArgs>(args)...,
-
-              // Callback
-              [this](const port_data_type_t &data, size_t hash) {
-                tp::for_each(mc_adapters_, [&](const auto &e) {
-                  const auto &[adapter_name, adapter_namehash, adapter_typehash, adapter] = e;
-                  if (adapter_typehash == hash) {
-                    adapter->callback()(adapter->decoder()(data));
-                  }
-                });
-              });
+        if constexpr (std::is_constructible_v<impl_type_t, const this_t *, const ImplArgs &..., PortImplCallback_<callback_aargs_t> &&>) {
+          m_impl__ = std::make_unique<impl_type_t>(this, std::forward<ImplArgs>(args)..., PortImplCallback_<callback_aargs_t>(this));
         }
       }
     });

src/port_base.hpp

@@ -1,8 +1,12 @@
 #pragma once
 
-#include <cstdint>
-#include <future>
+#include "src/tuple.hpp"
+#include <boost/callable_traits/args.hpp>
+#include <boost/callable_traits/return_type.hpp>
 #include <msgpack.hpp>
+#include <tuple>
+
+#define ZMQ_BUILD_DRAFT_API
 #include <zmq.hpp>
 
 #define FMT_HEADER_ONLY
@@ -13,6 +17,8 @@
 #include "port_types.hpp"
 #include "utils.hpp"
 
+using namespace boost::callable_traits;
+
 template <typename EncodedType> class PortBase {
   class AddressedPort_ {
   public:
@@ -28,14 +34,15 @@ template <typename EncodedType> class PortBase {
   };
 
 public:
-  PortBase(enum port_types_e port_type, const std::string &name, const std::string &endpoint, zmq::context_t &zmq_ctx)
-      : mc_type_(port_type), mc_name_(name), mc_endpoint_(endpoint), mc_name_hash_(std::hash<std::string>()(name)), mc_endpoint_hash_(std::hash<std::string>()(endpoint)) {}
+  PortBase(enum port_types_e port_type, const std::string &name, const std::map<std::string, std::string> &endpoints, zmq::context_t &zmq_ctx)
+      : mc_type_(port_type), mc_name_(name), mc_endpoints_(endpoints), mc_name_hash_(std::hash<std::string>()(name)) {}
+
+  virtual ~PortBase() = default;
 
   inline const auto &type() const { return mc_type_; }
   inline const auto &name() const { return mc_name_; }
-  inline const auto &endpoint() const { return mc_endpoint_; }
+  inline const auto &endpoints() const { return mc_endpoints_; }
   inline const auto &name_hash() const { return mc_name_hash_; }
-  inline const auto &endpoint_hash() const { return mc_endpoint_hash_; }
 
   const auto operator[](const std::string &address) const { return AddressedPort_(this, address); }
 
@@ -44,23 +51,37 @@ public:
     return *this;
   }
 
-  void listen() const { listen__(m_ss_.get_token()); };
   virtual void stop() const = 0;
-
-  template <typename AdapterInType> auto &callback(const std::string &name) const {
-    return (*static_cast<AdapterBase<AdapterInType> *>(get_adapter__(name, std::hash<std::string>()(name), typeid(std::remove_cvref_t<AdapterInType>).hash_code()))).callback();
+  template <typename Signature> auto &callback(const std::string &name) const {
+    return GetCallbackHelper_<typename tp::tuple_tail<args_t<Signature>>::type>(this).template operator()<Signature>(name);
   }
 
 protected:
-  void stop__() const { m_ss_.request_stop(); }
-
+  virtual void stop__() const = 0;
   virtual void send__(const std::string &addr, const void *data, size_t size, size_t type_hash) const = 0;
-  virtual void listen__(std::stop_token) const = 0;
-  virtual void *get_adapter__(const std::string &name, size_t namehash, size_t typehash) const = 0;
+  virtual void *get_adapter__(const std::string &name, size_t namehash, size_t typehash, size_t cbk_typehash) const = 0;
 
 private:
   const enum port_types_e mc_type_;
-  const std::string mc_name_, mc_endpoint_;
-  const size_t mc_name_hash_, mc_endpoint_hash_;
-  mutable std::stop_source m_ss_;
+  const std::string mc_name_;
+  const std::map<std::string, std::string> mc_endpoints_;
+  const size_t mc_name_hash_;
+
+  template <typename...> class GetCallbackHelper_;
+  template <typename... Aargs> class GetCallbackHelper_<std::tuple<Aargs...>> {
+  public:
+    GetCallbackHelper_(const PortBase *port) : mc_port_(port) {}
+
+    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>>>;
+      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(),
+                                          typeid(typename AdapterBase<arg_type_t, tag_s<ret_type_t>, Aargs...>::callback_type_t).hash_code())))
+          .callback();
+    }
+
+  private:
+    const PortBase *mc_port_;
+  };
 };

src/port_dealer_impl.hpp

@@ -7,16 +7,18 @@
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::DEALER, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::DEALER, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
+  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)) {}
+
+  // Send to socket depending on implementation
+  void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
+
+private:
+  void listen__(std::stop_token st) const override {
     while (!st.stop_requested()) {
       std::this_thread::sleep_for(std::chrono::milliseconds(100u));
     }
   }
-
-  // Send to socket depending on implementation
-  void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
 };
-

src/port_factory.hpp

@@ -3,8 +3,8 @@
 #include "port.hpp"
 
 template <typename... Adapters, typename... Args>
-auto makePort(enum port_types_e pt, const std::string &name, const std::string &endpoint, zmq::context_t &zmq_ctx, std::tuple<std::unique_ptr<Adapters>...> &&adapters,
-              std::tuple<Args...> &&args = {}) {
-  return std::make_unique<Port<std::tuple<Adapters...>, std::tuple<Args...>>>(pt, name, endpoint, zmq_ctx, std::forward<std::tuple<std::unique_ptr<Adapters>...>>(adapters),
+auto makePort(enum 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 = {}) {
+  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));
 }

src/port_impl_base.hpp

@@ -4,23 +4,24 @@
 #include <future>
 #include <msgpack.hpp>
 #include <tuple>
+
+#define ZMQ_BUILD_DRAFT_API
 #include <zmq.hpp>
 
-template <typename Callback> class PortImplBase {
+template <typename Port, typename Callback> class PortImplBase {
   using port_data_type_t_ = std::remove_cvref_t<std::tuple_element_t<0, boost::callable_traits::args_t<Callback>>>;
 
 public:
   struct port_payload_s {
     size_t typehash;
-    port_data_type_t_ data;
+    std::vector<port_data_type_t_> data;
     MSGPACK_DEFINE(typehash, data);
   };
 
-  PortImplBase(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : mc_endpoint__(endpoint), m_ctx__(zmq_ctx), mc_cbk__(callback) {}
-
+  PortImplBase(const Port *port, zmq::context_t &zmq_ctx, const std::map<std::string, std::string> &endpoints, Callback &&callback)
+      : mc_endpoints__(endpoints), m_ctx__(zmq_ctx), mc_cbk__(callback), mc_port__(port) {}
   virtual ~PortImplBase() = default;
 
-  virtual void listen(std::stop_token st) const = 0;
   virtual void send(const std::string &addr, const msgpack::sbuffer &data) const = 0;
   void close() {
     m_sock__.close();
@@ -30,11 +31,18 @@ public:
     }
   };
 
+  inline auto &stop_source() { return m_ss_; }
+
 protected:
+  static constexpr auto sc_recv_timeout_ms__ = 1'000u;
   mutable std::future<void> m_listener_thread__;
   mutable zmq::socket_t m_sock__;
   zmq::context_t &m_ctx__;
-  const std::string mc_endpoint__;
+  const std::map<std::string, std::string> mc_endpoints__;
   const Callback mc_cbk__;
+  const Port *mc_port__;
+  virtual void listen__(std::stop_token st) const = 0;
 
+private:
+  mutable std::stop_source m_ss_;
 };

src/port_pair_impl.hpp

@@ -7,15 +7,24 @@
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::PAIR, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::PAIR, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
-    while (!st.stop_requested()) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+  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::push);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.bind(ep);
     }
   }
 
   // Send to socket depending on implementation
   void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
+
+private:
+  void listen__(std::stop_token st) const override {
+    while (!st.stop_requested()) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+    }
+  }
 };

src/port_publisher_impl.hpp

@@ -7,17 +7,19 @@
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::PUB, Callback> final : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::PUB, Port, Callback> final : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(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::pub);
-    this->m_sock__.bind(this->mc_endpoint__);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.bind(ep);
+    }
   }
 
   ~PortImpl() override {}
 
-  void listen(std::stop_token st) const override { throw std::runtime_error("Can't listen on PUBLISHER pattern socket"); }
-
   // Send to socket depending on implementation
   void send(const std::string &addr, const msgpack::sbuffer &data) const override {
     try {
@@ -27,4 +29,7 @@ public:
       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 PUBLISHER pattern socket"); }
 };

src/port_pull_impl.hpp

@@ -2,20 +2,61 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <chrono>
+#include <zmq.hpp>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::PULL, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::PULL, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
-    while (!st.stop_requested()) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+  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::pull);
+
+    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 std::string &addr, const msgpack::sbuffer &data) const override {};
+  void send(const std::string &addr, const msgpack::sbuffer &data) const override { throw std::runtime_error("Can't send anything on PULL pattern socket"); };
+
+private:
+  void listen__(std::stop_token st) const override {
+    this->m_listener_thread__ = std::async(
+        std::launch::async,
+        [this](std::stop_token st) {
+          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);
+              });
+            }
+          }
+        },
+        st);
+  }
 };

src/port_push_impl.hpp

@@ -7,15 +7,26 @@
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::PUSH, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::PUSH, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
-    while (!st.stop_requested()) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+  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::push);
+    
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.bind(ep);
     }
   }
 
   // Send to socket depending on implementation
-  void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
+  void send(const std::string &addr, const msgpack::sbuffer &data) 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 { throw std::runtime_error("Can't listen on PUSH pattern socket"); }
 };

src/port_rep_impl.hpp

@@ -1,22 +1,92 @@
 #pragma once
 
-#include "port_impl_base.hpp"
-#include "port_types.hpp"
+#include <boost/callable_traits/return_type.hpp>
+#include <optional>
+#include <stdexcept>
+#include <tuple>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::REP, Callback> : public PortImplBase<Callback> {
+#include "port_impl_base.hpp"
+#include "port_types.hpp"
+#include "tuple.hpp"
+
+template <typename Port, typename Callback> class PortImpl<port_types_e::REP, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
-    while (!st.stop_requested()) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+  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::rep);
+    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 std::string &addr, const msgpack::sbuffer &data) const override {};
-};
+  void send(const std::string &addr, const msgpack::sbuffer &data) const override { throw std::runtime_error("Can't send anything on REPLY pattern socket"); };
 
+private:
+  void listen__(std::stop_token st) const override {
+    this->m_listener_thread__ = std::async(
+        std::launch::async,
+        [this](std::stop_token st) {
+          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) {
+                  if constexpr (!std::is_void_v<boost::callable_traits::return_type_t<std::remove_cvref_t<decltype(this->mc_cbk__)>>>) {
+                    auto reply_data = this->mc_cbk__(data, typehash);
+                    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();
+
+                      tp::for_each(this->mc_port__->adapters(), [&](const auto &e) {
+                        const auto &[adapter_name, adapter_namehash, adapter_typehash, adapter_cbk_typehash, adapter] = e;
+                        if constexpr (std::is_same_v<std::remove_cvref_t<std::tuple_element_t<0, boost::callable_traits::args_t<decltype(adapter->encoder())>>>,
+                                                     adapter_in_type_t>) {
+                          if (adapter_typehash == typehash) {
+
+                            typename base_t::port_payload_s payload = {
+                                .typehash = typeid(d).hash_code(),
+                                .data = {adapter->encoder()(d)},
+                            };
+
+                            msgpack::sbuffer buf;
+                            msgpack::pack(buf, payload);
+                            this->m_sock__.send(zmq::message_t(buf.data(), buf.size()), zmq::send_flags::none);
+                          }
+                        }
+                      });
+                    }
+                  } else {
+                    std::runtime_error("Callback of REPLY pattern socket should return non-void value");
+                  }
+                }
+
+                return std::optional(res);
+              });
+            }
+          }
+        },
+        st);
+  }
+};

src/port_req_impl.hpp

@@ -2,21 +2,47 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <stdexcept>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::REQ, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::REQ, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
-    while (!st.stop_requested()) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(100u));
+  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::req);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.connect(ep);
     }
   }
 
-  // Send to socket depending on implementation
-  void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
+  void send(const std::string &addr, const msgpack::sbuffer &data) const override {
+    try {
+      zmq::message_t reply;
+      this->m_sock__.send(zmq::message_t(data.data(), data.size()), zmq::send_flags::none);
+
+      this->m_sock__.recv(reply, zmq::recv_flags::none).and_then([&](const auto &res) {
+        typename base_t::port_payload_s payload;
+
+        msgpack::sbuffer buf;
+        buf.write(reinterpret_cast<const char *>(reply.data()), reply.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 &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 REQUEST pattern socket"); }
 };
-

src/port_router_impl.hpp

@@ -7,16 +7,18 @@
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::ROUTER, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::ROUTER, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, Callback &&callback) : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)) {}
-  void listen(std::stop_token st) const override {
+  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)) {}
+
+  // Send to socket depending on implementation
+  void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
+
+private:
+  void listen__(std::stop_token st) const override {
     while (!st.stop_requested()) {
       std::this_thread::sleep_for(std::chrono::milliseconds(100u));
     }
   }
-
-  // Send to socket depending on implementation
-  void send(const std::string &addr, const msgpack::sbuffer &data) const override {};
 };
-

src/port_subscriber_impl.hpp

@@ -2,49 +2,33 @@
 
 #include "port_impl_base.hpp"
 #include "port_types.hpp"
+#include <chrono>
+#include <ranges>
+#include <stop_token>
+#include <sys/socket.h>
+#include <utility>
+#include <zmq.hpp>
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
-template <typename Callback> class PortImpl<port_types_e::SUB, Callback> : public PortImplBase<Callback> {
+template <typename Port, typename Callback> class PortImpl<port_types_e::SUB, Port, Callback> : public PortImplBase<Port, Callback> {
 public:
-  using base_t = PortImplBase<Callback>;
-  PortImpl(zmq::context_t &zmq_ctx, const std::string &endpoint, const std::list<std::string> &topics, Callback &&callback)
-      : PortImplBase<Callback>(zmq_ctx, endpoint, std::forward<Callback>(callback)), mc_topics_(topics) {
+  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> &topics, Callback &&callback)
+      : PortImplBase<Port, Callback>(port, zmq_ctx, endpoints, std::forward<Callback>(callback)), mc_topics_(topics) {
     this->m_sock__ = zmq::socket_t(this->m_ctx__, zmq::socket_type::sub);
-    this->m_sock__.connect(this->mc_endpoint__);
+
+    for (const auto &[_, ep] : this->mc_endpoints__) {
+      this->m_sock__.connect(ep);
+    }
 
     for (const auto &topic : mc_topics_) {
       this->m_sock__.set(zmq::sockopt::subscribe, topic);
     }
-  }
 
-  void listen(std::stop_token st) const override {
-    this->m_listener_thread__ = std::async(
-        std::launch::async,
-        [this](std::stop_token st) {
-          while (!st.stop_requested()) {
-            zmq::message_t msg;
-
-            this->m_sock__.recv(msg, zmq::recv_flags::dontwait).and_then([&](const auto &res) {
-              fmt::print("Received envelope: {}\r\n", std::string(static_cast<const char *>(msg.data()), msg.size()));
-              this->m_sock__.recv(msg).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, data] = msgpack::unpack(buf.data(), buf.size()).get().convert(payload);
-
-                this->mc_cbk__(data, typehash);
-                return std::optional(res);
-              });
-
-              return std::optional(res);
-            });
-          }
-        },
-        st);
+    listen__(this->stop_source().get_token());
   }
 
   // Send to socket depending on implementation
@@ -52,4 +36,43 @@ public:
 
 private:
   const std::list<std::string> mc_topics_;
+
+  void listen__(std::stop_token st) const override {
+    this->m_listener_thread__ = std::async(
+        std::launch::async,
+        [this](std::stop_token st) {
+          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 envelope = std::string(static_cast<const char *>(msg.data()), msg.size());
+
+                this->m_sock__.recv(msg).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, envelope);
+                  }
+
+                  return std::optional(res);
+                });
+
+                return std::optional(res);
+              });
+            }
+          }
+        },
+        st);
+  }
 };

src/port_types.hpp

@@ -8,11 +8,29 @@ enum class port_types_e : uint32_t {
   SUB,
   REQ,
   REP,
-  ROUTER,
-  DEALER,
   PUSH,
   PULL,
   PAIR,
+  RADIO,
+  DISH,
+
+  ROUTER,
+  DEALER,
 };
 
 template <enum port_types_e, typename...> class PortImpl;
+
+struct endpoints_base_s {};
+template <auto> struct endpoints_s;
+
+template<> struct endpoints_s<port_types_e::PUB> : endpoints_base_s {};
+template<> struct endpoints_s<port_types_e::SUB> : endpoints_base_s {};
+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::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 {};
+template<> struct endpoints_s<port_types_e::DEALER> : endpoints_base_s {};

src/templates.cpp

@@ -1,116 +1,330 @@
 #include <boost/signals2/signal_type.hpp>
+#include <boost/signals2/variadic_signal.hpp>
 #include <chrono>
 #include <cstdint>
 #include <string>
 #include <thread>
 #include <tuple>
 
-#include <zmq.hpp>
-
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
 #include <fmt/ranges.h>
 
 #include "adapter_factory.hpp"
+#include "codecs.hpp"
 #include "module_factory.hpp"
 #include "port_factory.hpp"
 
-void entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<std::vector<uint8_t>> *> &ports) {
-  const auto &subscriber = *ports.at("subscriber");
-  const auto &publisher = *ports.at("publisher");
+using env_data_type_t = std::vector<uint8_t>;
 
-  auto &a = subscriber.callback<int32_t>("int-string-int");
-  auto &b = subscriber.callback<std::string>("string-string-string");
+void subscriber_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  const auto &subscriber = *ports.at("subscriber_port"); // Get subscriber port
 
-  a.connect([](const int32_t &i) -> void { fmt::print("Echo: {}, typename: {}\r\n", i, type_name<decltype(i)>()); });
-  b.connect([](const std::string &i) -> void { fmt::print("Echo: {}, typename: {}\r\n", i, type_name<decltype(i)>()); });
+  auto &int_cbk = subscriber.callback<void(const int32_t &, const std::string &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = subscriber.callback<void(const std::string &, const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = subscriber.callback<void(const double &, const std::string &)>("double-vector<uint8_t>-double");
 
-  subscriber.listen();
+  // Connect callbacks
+  int_cbk.connect([](const int32_t &i, const std::string &addr) {
+    fmt::print("SUBSCRIBER socket: got data: {} of {} type from address: {}\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>(), addr);
+  });
+
+  string_cbk.connect([](const std::string &s, const std::string &addr) {
+    fmt::print("SUBSCRIBER socket: got data: {} of {} type from address: {}\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>(), addr);
+  });
+
+  double_cbk.connect([](const double &d, const std::string &addr) {
+    fmt::print("SUBSCRIBER socket: got data: {} of {} type from address: {}\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>(), addr);
+  });
+}
+
+// Publisher module entrypoint
+void publisher_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  const auto &publisher = *ports.at("publisher_port"); // Get publisher port
+
+  // Publish data
   publisher["topic0"] << 1 << 2 << double{3.f} << std::string("test");
   publisher["topic1"] << 1 << 2 << double{3.f} << std::string("test");
   publisher["topic2"] << 1 << 2 << double{3.f} << std::string("test");
   publisher["topic3"] << 1 << 2 << double{3.f} << std::string("test");
+}
 
-  std::this_thread::sleep_for(std::chrono::milliseconds(1000u));
-  subscriber.stop();
+// Publisher module entrypoint
+void pusher_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  const auto &publisher = *ports.at("push_port"); // Get publisher port
+  publisher << 1 << 2 << double{3.f} << std::string("test");
+}
+
+void puller_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  const auto &puller = *ports.at("pull_port"); // Get subscriber port
+
+  auto &int_cbk = puller.callback<void(const int32_t &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = puller.callback<void(const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = puller.callback<void(const double &)>("double-vector<uint8_t>-double");
+
+  // Connect callbacks
+  int_cbk.connect([](const int32_t &i) { fmt::print("PULL socket: got data: {} of {} type\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>()); });
+  string_cbk.connect([](const std::string &s) { fmt::print("PULL socket: got data: {} of {} type\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>()); });
+  double_cbk.connect([](const double &d) { fmt::print("PULL socket: got data: {} of {} type\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>()); });
+}
+
+void req_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  const auto &req = *ports.at("req_port"); // Get publisher port
+  // static_assert(std::is_same_v<typename std::remove_cvref_t<decltype(req)>, void>, "");
+  auto &int_cbk = req.callback<void(const int32_t &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = req.callback<void(const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = req.callback<void(const double &)>("double-vector<uint8_t>-double");
+
+  // Connect callbacks
+  int_cbk.connect([](const int32_t &i) { fmt::print("REQUEST socket: got data: {} of {} type\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>()); });
+  string_cbk.connect([](const std::string &s) { fmt::print("REQUEST socket: got data: {} of {} type\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>()); });
+  double_cbk.connect([](const double &d) { fmt::print("REQUEST socket: got data: {} of {} type\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>()); });
+
+  req << 1 << 2 << double{3.f} << std::string("test");
+}
+
+void rep_entry(int32_t argc, char **argv, char **envp, const std::unordered_map<std::string, const PortBase<env_data_type_t> *> &ports) {
+  const auto &rep = *ports.at("rep_port"); // Get subscriber port
+
+  auto &int_cbk = rep.callback<std::string(const int32_t &)>("int-vector<uint8_t>-int");
+  auto &string_cbk = rep.callback<std::string(const std::string &)>("string-vector<uint8_t>-string");
+  auto &double_cbk = rep.callback<std::string(const double &)>("double-vector<uint8_t>-double");
+
+  // Connect callbacks
+  int_cbk.connect([](const int32_t &i) -> std::string {
+    fmt::print("REPLY socket: got data: {} of {} type\r\n", i, type_name<std::remove_cvref_t<decltype(i)>>());
+    // Handle data ...
+    return fmt::format("'Handle request: {} of type: {}'", i, type_name<std::remove_cvref_t<decltype(i)>>());
+  });
+
+  string_cbk.connect([](const std::string &s) -> std::string {
+    fmt::print("REPLY socket: got data: {} of {} type\r\n", s, type_name<std::remove_cvref_t<decltype(s)>>());
+    // Handle data ...
+    return fmt::format("'Handle request: {} of type: {}'", s, type_name<std::remove_cvref_t<decltype(s)>>());
+  });
+
+  double_cbk.connect([](const double &d) -> std::string {
+    fmt::print("REPLY socket: got data: {} of {} type\r\n", d, type_name<std::remove_cvref_t<decltype(d)>>());
+    // Handle data ...
+    return fmt::format("'Handle request: {} of type: {}'", d, type_name<std::remove_cvref_t<decltype(d)>>());
+  });
 }
 
 int main(int argc, char *argv[], char *envp[]) {
   using enum port_types_e;
-  zmq::context_t zmq_ctx;
+  codecs_s<env_data_type_t> codecs;
+  zmq::context_t zmq_ctx; // Use common context because both modules have in-process ports, but working in different threads
 
-  auto mod = makeModule(argc, argv, envp, "test_mod", zmq_ctx,
+  // Make module that contains only 1 port working on PUBLISHER pattern
+  // TODO: merge all builders to one complex "Module" builder
+  auto publisher_module = makeModule(argc, argv, envp, "publisher_module", zmq_ctx,
                                      std::tuple{
-                            makePort(PUB, "publisher", "inproc://publisher_port", zmq_ctx,
+                                         makePort(PUB, "publisher_port",
+                                                  {
+                                                      {"test", "inproc://PUB-SUB"},
+                                                  } /* This port will publish messages here  */,
+                                                  zmq_ctx,
                                                   std::tuple{
-                                         makeAdapter("int-string-int", std::pair{
-                                                                           [](const int32_t &i) -> std::vector<uint8_t> {
-                                                                             auto str = std::to_string(i + 5);
-                                                                             return {str.begin(), str.end()};
-                                                                           },
+                                                      AdapterBuilder()
+                                                          .encodeDataBy(&codecs.encoders.from_int)
+                                                          .decodeDataBy(&codecs.decoders.to_int)
+                                                          .withCallbackSignature<void(const int32_t &)>()
+                                                          .withName("int-vector<uint8_t>-int")
+                                                          .finalize(),
 
-                                                                           [](const std::vector<uint8_t> &s) -> int32_t { return 5; },
-                                                                       }),
+                                                      AdapterBuilder()
+                                                          .encodeDataBy(&codecs.encoders.from_string)
+                                                          .decodeDataBy(&codecs.decoders.to_string)
+                                                          .withCallbackSignature<void(const std::string)>()
+                                                          .withName("string-vector<uint8_t>-string")
+                                                          .finalize(),
 
-                                         makeAdapter("string-string-string", std::pair{
-                                                                                 [](const std::string &i) -> std::vector<uint8_t> {
-                                                                                   auto str = i + "_test";
-                                                                                   return {str.begin(), str.end()};
-                                                                                 },
-
-                                                                                 [](const std::vector<uint8_t> &i) -> std::string { return "works!"; },
-                                                                             }),
-
-                                         makeAdapter("double-string-double", std::pair{
-                                                                                 [](const double &i) -> std::vector<uint8_t> {
-                                                                                   auto str = std::to_string(i / 2.f);
-                                                                                   return {str.begin(), str.end()};
-                                                                                 },
-
-                                                                                 [](const std::vector<uint8_t> &s) -> double { return .1f; },
-                                                                             }),
-                                     }),
-
-                            makePort(SUB, "subscriber", "inproc://publisher_port", zmq_ctx,
-                                     std::tuple{
-                                         makeAdapter("int-string-int", std::pair{
-                                                                           [](const int32_t &i) -> std::vector<uint8_t> {
-                                                                             auto str = std::to_string(i + 5);
-                                                                             return {str.begin(), str.end()};
-                                                                           },
-
-                                                                           [](const std::vector<uint8_t> &s) -> int32_t { return 5; },
-                                                                       }),
-
-                                         makeAdapter("string-string-string", std::pair{
-                                                                                 [](const std::string &i) -> std::vector<uint8_t> {
-                                                                                   auto str = i + "_test";
-                                                                                   return {str.begin(), str.end()};
-                                                                                 },
-
-                                                                                 [](const std::vector<uint8_t> &i) -> std::string { return "works!"; },
-                                                                             }),
-
-                                         makeAdapter("double-string-double", std::pair{
-                                                                                 [](const double &i) -> std::vector<uint8_t> {
-                                                                                   auto str = std::to_string(i / 2.f);
-                                                                                   return {str.begin(), str.end()};
-                                                                                 },
-
-                                                                                 [](const std::vector<uint8_t> &s) -> double { return .1f; },
-                                                                             }),
-                                     },
-
-                                     std::tuple{
-                                         std::list<std::string>{
-                                             "topic0",
-                                             "topic1",
-                                             "topic2",
-                                             "topic3",
-                                         },
+                                                      AdapterBuilder()
+                                                          .encodeDataBy(&codecs.encoders.from_double)
+                                                          .decodeDataBy(&codecs.decoders.to_double)
+                                                          .withCallbackSignature<void(const double &)>()
+                                                          .withName("double-vector<uint8_t>-double")
+                                                          .finalize(),
                                                   }),
                                      });
 
-  mod->run(entry);
+  // Make module that contains only 1 port working on SUBSCRIBER pattern
+  auto subscriber_module = makeModule(argc, argv, envp, "subscriber_module", zmq_ctx,
+                                      std::tuple{
+                                          makePort(SUB, "subscriber_port",
+                                                   {
+                                                       {"test", "inproc://PUB-SUB"},
+                                                   } /* this port will read data here  */,
+                                                   zmq_ctx,
+                                                   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(),
+                                                   },
+
+                                                   // This type of port requires arguments - topics to subscribe
+                                                   std::tuple{
+                                                       // Topics to subscribe
+                                                       std::list<std::string>{"topic0", "topic1", "topic2", "topic3"},
+                                                   }),
+                                      });
+
+  auto pusher_module = makeModule(argc, argv, envp, "pusher_module", zmq_ctx,
+                                  std::tuple{
+                                      makePort(PUSH, "push_port",
+                                               {
+                                                   {"test", "inproc://PUSH-PULL"},
+                                               } /* This port will publish messages here  */,
+                                               zmq_ctx,
+                                               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(),
+                                               }),
+                                  });
+
+  auto puller_module = makeModule(argc, argv, envp, "puller_module", zmq_ctx,
+                                  std::tuple{
+                                      makePort(PULL, "pull_port",
+                                               {
+                                                   {"test", "inproc://PUSH-PULL"},
+                                               } /* This port will publish messages here  */,
+                                               zmq_ctx,
+                                               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(),
+                                               }),
+                                  });
+
+  auto req_module = makeModule(argc, argv, envp, "req_module", zmq_ctx,
+                               std::tuple{
+                                   makePort(REQ, "req_port",
+                                            {
+                                                {"test", "inproc://REQ-REP"},
+                                            } /* This port will publish messages here  */,
+                                            zmq_ctx,
+                                            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(),
+                                            }),
+                               });
+
+  auto rep_module = makeModule(argc, argv, envp, "rep_module", zmq_ctx,
+                               std::tuple{
+                                   makePort(REP, "rep_port",
+                                            {
+                                                {"test", "inproc://REQ-REP"},
+                                            } /* This port will publish messages here  */,
+                                            zmq_ctx,
+                                            std::tuple{
+                                                AdapterBuilder()
+                                                    .encodeDataBy(&codecs.encoders.from_int)
+                                                    .decodeDataBy(&codecs.decoders.to_int)
+                                                    .withCallbackSignature<std::string(const int32_t &)>()
+                                                    .withName("int-vector<uint8_t>-int")
+                                                    .finalize(),
+
+                                                AdapterBuilder()
+                                                    .encodeDataBy(&codecs.encoders.from_string)
+                                                    .decodeDataBy(&codecs.decoders.to_string)
+                                                    .withCallbackSignature<std::string(const std::string)>()
+                                                    .withName("string-vector<uint8_t>-string")
+                                                    .finalize(),
+
+                                                AdapterBuilder()
+                                                    .encodeDataBy(&codecs.encoders.from_double)
+                                                    .decodeDataBy(&codecs.decoders.to_double)
+                                                    .withCallbackSignature<std::string(const double &)>()
+                                                    .withName("double-vector<uint8_t>-double")
+                                                    .finalize(),
+                                            }),
+                               });
+
+  fmt::print("\r\nPUB-SUB test:\r\n");
+  subscriber_module->run(subscriber_entry); // Subscribe and get data
+  publisher_module->run(publisher_entry);   // Publish data
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+  fmt::print("\r\nPUSH-PULL test:\r\n");
+  pusher_module->run(pusher_entry);
+  puller_module->run(puller_entry);
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+  fmt::print("\r\nREQ-REP test:\r\n");
+  rep_module->run(rep_entry);
+  req_module->run(req_entry);
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+  fmt::print("DONE!\r\n");
+
   return 0;
 }

src/tuple.hpp

@@ -0,0 +1,126 @@
+#pragma once
+
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
+#include <functional>
+#include <optional>
+#include <stdexcept>
+#include <tuple>
+#include <utility>
+
+namespace tp {
+namespace detail {
+
+template <typename Func, typename... Args> constexpr Func for_each_arg(Func f, Args &&...args) {
+  (f(std::forward<Args>(args)), ...);
+  return f;
+}
+
+template <typename Tuple, typename Func, std::size_t... I> constexpr Func for_each_impl(Tuple &&t, Func &&f, std::index_sequence<I...> is) {
+  (std::forward<Func>(f)(std::get<I>(std::forward<Tuple>(t))), ...);
+  return f;
+}
+
+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))...};
+}
+
+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)...); }
+
+// ZIP utilities
+template <std::size_t I, typename... Tuples> using zip_tuple_at_index_t = std::tuple<std::tuple_element_t<I, std::decay_t<Tuples>>...>;
+template <std::size_t I, typename... Tuples> zip_tuple_at_index_t<I, Tuples...> zip_tuple_at_index(Tuples &&...tuples) { return {std::get<I>(std::forward<Tuples>(tuples))...}; }
+template <typename... Tuples, std::size_t... I> std::tuple<zip_tuple_at_index_t<I, Tuples...>...> tuple_zip_impl(Tuples &&...tuples, std::index_sequence<I...>) {
+  return {zip_tuple_at_index<I>(std::forward<Tuples>(tuples)...)...};
+}
+}; // namespace detail
+
+template <class Tuple, class F> constexpr decltype(auto) for_each(Tuple &&tuple, F &&f) {
+  return []<std::size_t... I>(Tuple &&tuple, F &&f, std::index_sequence<I...>) {
+    (f(std::get<I>(tuple)), ...);
+    return f;
+  }(std::forward<Tuple>(tuple), std::forward<F>(f), std::make_index_sequence<std::tuple_size<std::remove_reference_t<Tuple>>::value>{});
+}
+
+template <typename... Ts, typename Function> auto transform(std::tuple<Ts...> const &inputs, Function function) {
+  return detail::transform_impl(inputs, function, std::make_index_sequence<sizeof...(Ts)>{});
+}
+
+template <typename Tuple, typename Predicate> constexpr size_t find_if(Tuple &&tuple, Predicate pred) {
+  size_t index = std::tuple_size<std::remove_reference_t<Tuple>>::value;
+  size_t currentIndex = 0;
+  bool found = false;
+
+  for_each(tuple, [&](auto &&value) {
+    if (!found && pred(value)) {
+      index = currentIndex;
+      found = true;
+    }
+
+    ++currentIndex;
+  });
+
+  return index;
+}
+
+template <typename Tuple, typename Action> void perform(Tuple &&tuple, size_t index, Action action) {
+  size_t currentIndex = 0;
+  for_each(tuple, [&action, index, &currentIndex](auto &&value) {
+    if (currentIndex == index) {
+
+      action(std::forward<decltype(value)>(value));
+    }
+
+    ++currentIndex;
+  });
+}
+
+template <typename Tuple, typename Predicate> bool all_of(Tuple &&tuple, Predicate pred) {
+  return find_if(tuple, std::not_fn(pred)) == std::tuple_size<std::decay_t<Tuple>>::value;
+}
+
+template <typename Tuple, typename Predicate> bool none_of(Tuple &&tuple, Predicate pred) { return find_if(tuple, pred) == std::tuple_size<std::decay_t<Tuple>>::value; }
+template <typename Tuple, typename Predicate> bool any_of(Tuple &&tuple, Predicate pred) { return !none_of(tuple, pred); }
+
+template <typename Tuple, typename Function> Tuple &operator|(Tuple &&tuple, Function func) {
+  for_each(tuple, func);
+  return tuple;
+}
+
+template <int trim, typename... T> auto subtuple(const std::tuple<T...> &t) { return detail::subtuple(t, std::make_index_sequence<sizeof...(T) - trim>()); }
+
+template <size_t starting, size_t elems, class Tuple, class Seq = decltype(std::make_index_sequence<elems>())> struct sub_range;
+
+template <size_t starting, size_t elems, class... Args, size_t... indx> struct sub_range<starting, elems, std::tuple<Args...>, std::index_sequence<indx...>> {
+  static_assert(elems <= sizeof...(Args) - starting, "sub range is out of bounds!");
+  using tuple = std::tuple<std::tuple_element_t<indx + starting, std::tuple<Args...>>...>;
+};
+
+template <typename Tuple, std::size_t... Ints> auto select_tuple(Tuple &&tuple, std::index_sequence<Ints...>) {
+  return std::tuple<std::tuple_element_t<Ints, Tuple>...>(std::get<Ints>(std::forward<Tuple>(tuple))...);
+}
+
+template <class T, class Tuple> struct tuple_index;
+
+template <class T, class... Types> struct tuple_index<T, std::tuple<T, Types...>> {
+  static const std::size_t value = 0;
+};
+
+template <class T, class U, class... Types> struct tuple_index<T, std::tuple<U, Types...>> {
+  static const std::size_t value = 1 + tuple_index<T, std::tuple<Types...>>::value;
+};
+
+// ZIP
+template <typename Head, typename... Tail>
+  requires((std::tuple_size_v<std::decay_t<Tail>> == std::tuple_size_v<std::decay_t<Head>>) && ...)
+auto tuple_zip(Head &&head, Tail &&...tail) {
+  return detail::tuple_zip_impl<Head, Tail...>(std::forward<Head>(head), std::forward<Tail>(tail)..., std::make_index_sequence<std::tuple_size_v<std::decay_t<Head>>>());
+}
+
+// TAIL
+template <typename T> struct tuple_tail;
+template <typename Head, typename... Tail> struct tuple_tail<std::tuple<Head, Tail...>> {
+  using type = std::tuple<Tail...>;
+}; //
+}; // namespace tp

src/utils.hpp

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <string_view>
+
+template <class T, template <class...> class Template> struct is_specialization : std::false_type {};
+template <template <class...> class Template, class... Args> struct is_specialization<Template<Args...>, Template> : std::true_type {};
+
+template <class T> constexpr std::string_view type_name() {
+  using namespace std;
+#ifdef __clang__
+  string_view p = __PRETTY_FUNCTION__;
+  return string_view(p.data() + 34, p.size() - 34 - 1);
+#elif defined(__GNUC__)
+  string_view p = __PRETTY_FUNCTION__;
+#  if __cplusplus < 201402
+  return string_view(p.data() + 36, p.size() - 36 - 1);
+#  else
+  return string_view(p.data() + 49, p.find(';', 49) - 49);
+#  endif
+#elif defined(_MSC_VER)
+  string_view p = __FUNCSIG__;
+  return string_view(p.data() + 84, p.size() - 84 - 7);
+#endif
+}
+
+template <typename T> struct tag_s {
+  using type = T;
+};