diff --git a/.cspell.json b/.cspell.json
index 4c14cb9f..f305225f 100644
--- a/.cspell.json
+++ b/.cspell.json
@@ -32,6 +32,7 @@
     "nproc",
     "nsec",
     "ntoa",
+    "OVFL",
     "pandar",
     "PANDAR",
     "PANDARAT",
diff --git a/nebula_common/include/nebula_common/loggers/console_logger.hpp b/nebula_common/include/nebula_common/loggers/console_logger.hpp
new file mode 100644
index 00000000..fb2a6127
--- /dev/null
+++ b/nebula_common/include/nebula_common/loggers/console_logger.hpp
@@ -0,0 +1,55 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "nebula_common/loggers/logger.hpp"
+
+#include <cstdio>
+#include <iostream>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <utility>
+
+namespace nebula::drivers::loggers
+{
+
+class ConsoleLogger : public Logger
+{
+public:
+  explicit ConsoleLogger(std::string name) : name_(std::move(name)) {}
+
+  void debug(const std::string & message) override { print_tagged(std::cout, "DEBUG", message); }
+  void info(const std::string & message) override { print_tagged(std::cout, "INFO", message); }
+  void warn(const std::string & message) override { print_tagged(std::cerr, "WARN", message); }
+  void error(const std::string & message) override { print_tagged(std::cerr, "ERROR", message); }
+
+  std::shared_ptr<Logger> child(const std::string & name) override
+  {
+    return std::make_shared<ConsoleLogger>(name_ + "." + name);
+  }
+
+private:
+  std::string name_;
+
+  void print_tagged(std::ostream & os, const std::string & severity, const std::string & message)
+  {
+    // In multithreaded logging, building the string first (... + ...) and then shifting to the
+    // stream will ensure that no other logger outputs between string fragments
+    os << ("[" + name_ + "][" + severity + "] " + message) << std::endl;
+  }
+};
+
+}  // namespace nebula::drivers::loggers
diff --git a/nebula_common/include/nebula_common/loggers/logger.hpp b/nebula_common/include/nebula_common/loggers/logger.hpp
new file mode 100644
index 00000000..b416e252
--- /dev/null
+++ b/nebula_common/include/nebula_common/loggers/logger.hpp
@@ -0,0 +1,49 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <memory>
+#include <sstream>
+#include <string>
+
+#define NEBULA_LOG_STREAM(log_func, stream_args) \
+  {                                              \
+    std::stringstream ss{};                      \
+    ss << stream_args;                           \
+    log_func(ss.str());                          \
+  }
+
+namespace nebula::drivers::loggers
+{
+
+class Logger
+{
+public:
+  Logger() = default;
+  Logger(const Logger &) = default;
+  Logger(Logger &&) = delete;
+  Logger & operator=(const Logger &) = default;
+  Logger & operator=(Logger &&) = delete;
+  virtual ~Logger() = default;
+
+  virtual void debug(const std::string & message) = 0;
+  virtual void info(const std::string & message) = 0;
+  virtual void warn(const std::string & message) = 0;
+  virtual void error(const std::string & message) = 0;
+
+  virtual std::shared_ptr<Logger> child(const std::string & name) = 0;
+};
+
+}  // namespace nebula::drivers::loggers
diff --git a/nebula_common/include/nebula_common/util/expected.hpp b/nebula_common/include/nebula_common/util/expected.hpp
index 1d433344..45ed0b59 100644
--- a/nebula_common/include/nebula_common/util/expected.hpp
+++ b/nebula_common/include/nebula_common/util/expected.hpp
@@ -15,6 +15,7 @@
 #pragma once
 
 #include <exception>
+#include <stdexcept>
 #include <string>
 #include <variant>
 
diff --git a/nebula_hw_interfaces/CMakeLists.txt b/nebula_hw_interfaces/CMakeLists.txt
index 2577ea9a..13efaaa1 100644
--- a/nebula_hw_interfaces/CMakeLists.txt
+++ b/nebula_hw_interfaces/CMakeLists.txt
@@ -2,13 +2,13 @@ cmake_minimum_required(VERSION 3.14)
 project(nebula_hw_interfaces)
 
 # Default to C++17
-if (NOT CMAKE_CXX_STANDARD)
+if(NOT CMAKE_CXX_STANDARD)
     set(CMAKE_CXX_STANDARD 17)
-endif ()
+endif()
 
-if (CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
     add_compile_options(-Wall -Wextra -Wpedantic -Wunused-function)
-endif ()
+endif()
 
 find_package(ament_cmake_auto REQUIRED)
 find_package(boost_tcp_driver)
@@ -53,7 +53,6 @@ target_link_libraries(nebula_hw_interfaces_velodyne PUBLIC
     ${boost_tcp_driver_LIBRARIES}
     ${boost_udp_driver_LIBRARIES}
     ${velodyne_msgs_TARGETS}
-
 )
 target_include_directories(nebula_hw_interfaces_velodyne PUBLIC
     ${boost_udp_driver_INCLUDE_DIRS}
@@ -68,7 +67,6 @@ target_link_libraries(nebula_hw_interfaces_robosense PUBLIC
     ${boost_tcp_driver_LIBRARIES}
     ${boost_udp_driver_LIBRARIES}
     ${robosense_msgs_TARGETS}
-
 )
 target_include_directories(nebula_hw_interfaces_robosense PUBLIC
     ${boost_udp_driver_INCLUDE_DIRS}
@@ -100,6 +98,34 @@ install(DIRECTORY include/ DESTINATION include/${PROJECT_NAME})
 if(BUILD_TESTING)
     find_package(ament_lint_auto REQUIRED)
     ament_lint_auto_find_test_dependencies()
+
+    find_package(ament_cmake_gtest REQUIRED)
+    find_package(ament_cmake_gmock REQUIRED)
+
+    ament_add_gtest(test_udp
+        test/common/test_udp.cpp
+    )
+
+    target_include_directories(test_udp PUBLIC
+        ${nebula_common_INCLUDE_DIRS}
+        include
+        test)
+
+    ament_add_gmock(hesai_test_ptc
+        test/hesai/test_ptc.cpp
+    )
+
+    target_include_directories(hesai_test_ptc PUBLIC
+        ${nebula_common_INCLUDE_DIRS}
+        ${nebula_hw_interfaces_hesai_INCLUDE_DIRS}
+        ${boost_tcp_driver_INCLUDE_DIRS}
+        ${boost_udp_driver_INCLUDE_DIRS}
+        include
+        test)
+
+    target_link_libraries(hesai_test_ptc
+        nebula_hw_interfaces_hesai
+    )
 endif()
 
 ament_export_include_directories("include/${PROJECT_NAME}")
diff --git a/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp b/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp
new file mode 100644
index 00000000..23cb4b92
--- /dev/null
+++ b/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp
@@ -0,0 +1,383 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <nebula_common/util/expected.hpp>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/poll.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <algorithm>
+#include <atomic>
+#include <cassert>
+#include <cerrno>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <exception>
+#include <functional>
+#include <optional>
+#include <stdexcept>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+namespace nebula::drivers::connections
+{
+
+class SocketError : public std::exception
+{
+public:
+  explicit SocketError(int err_no) : what_(strerror(err_no)) {}
+
+  explicit SocketError(const char * msg) : what_(msg) {}
+
+  const char * what() const noexcept override { return what_.c_str(); }
+
+private:
+  std::string what_;
+};
+
+class UsageError : public std::runtime_error
+{
+public:
+  explicit UsageError(const std::string & msg) : std::runtime_error(msg) {}
+};
+
+class UdpSocket
+{
+private:
+  struct Endpoint
+  {
+    std::string ip;
+    uint16_t port;
+  };
+
+  struct MsgBuffers
+  {
+    msghdr msg{};
+    iovec iov{};
+    std::array<std::byte, 1024> control;
+    sockaddr_in sender_addr;
+  };
+
+  class DropMonitor
+  {
+    uint32_t last_drop_counter_{0};
+
+  public:
+    uint32_t get_drops_since_last_receive(uint32_t current_drop_counter)
+    {
+      uint32_t last = last_drop_counter_;
+      last_drop_counter_ = current_drop_counter;
+
+      bool counter_did_wrap = current_drop_counter < last;
+      if (counter_did_wrap) {
+        return (UINT32_MAX - last) + current_drop_counter;
+      }
+
+      return current_drop_counter - last;
+    }
+  };
+
+  enum class State { UNINITIALIZED, INITIALIZED, BOUND, ACTIVE };
+
+  static const int g_poll_timeout_ms = 10;
+
+public:
+  struct ReceiveMetadata
+  {
+    std::optional<uint64_t> timestamp_ns;
+    uint64_t drops_since_last_receive{0};
+    bool truncated;
+  };
+
+  using callback_t = std::function<void(const std::vector<uint8_t> &, const ReceiveMetadata &)>;
+
+  /**
+   * @brief Construct a UDP socket with timestamp measuring enabled. The minimal way to start
+   * receiving on the socket is `UdpSocket().init(...).bind().subscribe(...);`.
+   */
+  UdpSocket()
+  {
+    int sock_fd = socket(AF_INET, SOCK_DGRAM, 0);
+    if (sock_fd == -1) throw SocketError(errno);
+
+    int enable = 1;
+    int result = setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
+    if (result < 0) throw SocketError(errno);
+
+    // Enable kernel-space receive time measurement
+    result = setsockopt(sock_fd, SOL_SOCKET, SO_TIMESTAMP, &enable, sizeof(enable));
+    if (result < 0) throw SocketError(errno);
+
+    // Enable reporting on packets dropped due to full UDP receive buffer
+    result = setsockopt(sock_fd, SOL_SOCKET, SO_RXQ_OVFL, &enable, sizeof(enable));
+    if (result < 0) throw SocketError(errno);
+
+    poll_fd_ = {sock_fd, POLLIN, 0};
+    sock_fd_ = sock_fd;
+  }
+
+  /**
+   * @brief Specify the host address and port for this socket to be bound to. To bind the socket,
+   * call the `bind()` function.
+   *
+   * @param host_ip The address to bind to.
+   * @param host_port The port to bind to.
+   */
+  UdpSocket & init(const std::string & host_ip, uint16_t host_port)
+  {
+    if (state_ > State::INITIALIZED) throw UsageError("Socket must be initialized before binding");
+
+    host_ = {host_ip, host_port};
+    state_ = State::INITIALIZED;
+    return *this;
+  }
+
+  /**
+   * @brief Set the socket to drop all packets not coming from `sender_ip` and `sender_port`.
+   *
+   * @param sender_ip The only allowed sender IP. Cannot be a multicast or broadcast address.
+   * @param sender_port The only allowed sender port.
+   */
+  UdpSocket & limit_to_sender(const std::string & sender_ip, uint16_t sender_port)
+  {
+    if (state_ >= State::ACTIVE) throw UsageError("Sender has to be set before subscribing");
+
+    sender_.emplace(Endpoint{sender_ip, sender_port});
+    return *this;
+  }
+
+  /**
+   * @brief Set the MTU this socket supports. While this can be set arbitrarily, it is best set to
+   * the MTU of the network interface, or to the maximum expected packet length.
+   *
+   * @param bytes The MTU size. The default value is 1500.
+   */
+  UdpSocket & set_mtu(size_t bytes)
+  {
+    if (state_ >= State::ACTIVE) throw UsageError("MTU size has to be set before subscribing");
+
+    buffer_size_ = bytes;
+    return *this;
+  }
+
+  UdpSocket & set_socket_buffer_size(size_t bytes)
+  {
+    if (state_ > State::INITIALIZED) throw UsageError("Buffer size has to be set before binding");
+    if (bytes > static_cast<size_t>(INT32_MAX))
+      throw UsageError("The maximum value supported (0x7FFFFFF) has been exceeded");
+
+    auto buf_size = static_cast<int>(bytes);
+    int result = setsockopt(sock_fd_, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof(buf_size));
+    if (result < 0) throw SocketError(errno);
+    return *this;
+  }
+
+  /**
+   * @brief Join an IP multicast group. Only one group can be joined by the socket.
+   *
+   * @param group_ip The multicast IP. It has to be in the multicast range `224.0.0.0/4` (between
+   * `224.0.0.0` and `239.255.255.255`).
+   */
+  UdpSocket & join_multicast_group(const std::string & group_ip)
+  {
+    if (state_ < State::INITIALIZED) throw UsageError("Socket has to be initialized first");
+    if (state_ >= State::BOUND)
+      throw UsageError("Multicast groups have to be joined before binding");
+
+    ip_mreq mreq;
+    mreq.imr_multiaddr.s_addr = inet_addr(group_ip.c_str());  // Multicast group address
+    mreq.imr_interface.s_addr = inet_addr(host_.ip.c_str());
+
+    int result = setsockopt(sock_fd_, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
+    if (result < 0) throw SocketError(errno);
+
+    multicast_ip_.emplace(group_ip);
+    return *this;
+  }
+
+  /**
+   * @brief Bind the socket to host IP and port given in `init()`. If `join_multicast_group()` was
+   * called before this function, the socket will be bound to `group_ip` instead. At least `init()`
+   * has to have been called before.
+   */
+  UdpSocket & bind()
+  {
+    if (state_ < State::INITIALIZED) throw UsageError("Socket has to be initialized first");
+    if (state_ >= State::BOUND) throw UsageError("Re-binding already bound socket");
+
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(host_.port);
+    addr.sin_addr.s_addr = multicast_ip_ ? inet_addr(multicast_ip_->c_str()) : INADDR_ANY;
+
+    int result = ::bind(sock_fd_, (struct sockaddr *)&addr, sizeof(addr));
+    if (result == -1) throw SocketError(errno);
+
+    state_ = State::BOUND;
+    return *this;
+  }
+
+  /**
+   * @brief Register a callback for processing received packets and start the receiver thread. The
+   * callback will be called for each received packet, and will be executed in the receive thread.
+   * Has to be called on a bound socket (`bind()` has to have been called before).
+   *
+   * @param callback The function to be executed for each received packet.
+   */
+  UdpSocket & subscribe(callback_t && callback)
+  {
+    if (state_ < State::BOUND) throw UsageError("Socket has to be bound first");
+    if (state_ > State::BOUND) throw UsageError("Cannot re-subscribe to socket");
+
+    callback_ = std::move(callback);
+    launch_receiver();
+    return *this;
+  }
+
+  /**
+   * @brief Gracefully stops the active receiver thread (if any) but keeps the socket alive. The
+   * same socket can later be subscribed again.
+   */
+  UdpSocket & unsubscribe()
+  {
+    if (state_ == State::ACTIVE) state_ = State::BOUND;
+    if (receive_thread_.joinable()) receive_thread_.join();
+    return *this;
+  }
+
+  ~UdpSocket()
+  {
+    unsubscribe();
+    close(sock_fd_);
+  }
+
+private:
+  void launch_receiver()
+  {
+    assert(state_ == State::BOUND);
+    assert(callback_);
+
+    state_ = State::ACTIVE;
+    receive_thread_ = std::thread([this]() {
+      std::vector<uint8_t> buffer;
+      while (state_ == State::ACTIVE) {
+        auto data_available = is_data_available();
+        if (!data_available.has_value()) throw SocketError(data_available.error());
+        if (!data_available.value()) continue;
+
+        buffer.resize(buffer_size_);
+        auto msg_header = make_msg_header(buffer);
+
+        ssize_t recv_result = recvmsg(sock_fd_, &msg_header.msg, MSG_TRUNC);
+        if (recv_result < 0) throw SocketError(errno);
+        size_t untruncated_packet_length = recv_result;
+
+        if (!is_accepted_sender(msg_header.sender_addr)) continue;
+
+        ReceiveMetadata metadata;
+        get_receive_metadata(msg_header.msg, metadata);
+        metadata.truncated = untruncated_packet_length > buffer_size_;
+
+        buffer.resize(std::min(buffer_size_, untruncated_packet_length));
+        callback_(buffer, metadata);
+      }
+    });
+  }
+
+  void get_receive_metadata(msghdr & msg, ReceiveMetadata & inout_metadata)
+  {
+    for (cmsghdr * cmsg = CMSG_FIRSTHDR(&msg); cmsg != nullptr; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
+      if (cmsg->cmsg_level != SOL_SOCKET) continue;
+
+      switch (cmsg->cmsg_type) {
+        case SO_TIMESTAMP: {
+          auto tv = (timeval const *)CMSG_DATA(cmsg);
+          uint64_t timestamp_ns = tv->tv_sec * 1'000'000'000 + tv->tv_usec * 1000;
+          inout_metadata.timestamp_ns.emplace(timestamp_ns);
+          break;
+        }
+        case SO_RXQ_OVFL: {
+          auto drops = (uint32_t const *)CMSG_DATA(cmsg);
+          inout_metadata.drops_since_last_receive =
+            drop_monitor_.get_drops_since_last_receive(*drops);
+          break;
+        }
+        default:
+          continue;
+      }
+    }
+  }
+
+  util::expected<bool, int> is_data_available()
+  {
+    int status = poll(&poll_fd_, 1, g_poll_timeout_ms);
+    if (status < 0) return errno;
+    return (poll_fd_.revents & POLLIN) && (status > 0);
+  }
+
+  bool is_accepted_sender(const sockaddr_in & sender_addr)
+  {
+    if (!sender_) return true;
+
+    std::array<char, INET_ADDRSTRLEN> sender_name;
+    inet_ntop(AF_INET, &sender_addr.sin_addr, sender_name.data(), INET_ADDRSTRLEN);
+    return std::strncmp(sender_->ip.c_str(), sender_name.data(), INET_ADDRSTRLEN) == 0;
+  }
+
+  MsgBuffers make_msg_header(std::vector<uint8_t> & receive_buffer) const
+  {
+    msghdr msg{};
+    iovec iov{};
+    std::array<std::byte, 1024> control;
+
+    sockaddr_in sender_addr;
+    socklen_t sender_addr_len = sizeof(sender_addr);
+
+    iov.iov_base = receive_buffer.data();
+    iov.iov_len = receive_buffer.size();
+
+    msg.msg_iov = &iov;
+    msg.msg_iovlen = 1;
+    msg.msg_control = control.data();
+    msg.msg_controllen = control.size();
+    msg.msg_name = &sender_addr;
+    msg.msg_namelen = sender_addr_len;
+
+    return MsgBuffers{msg, iov, control, sender_addr};
+  }
+
+  std::atomic<State> state_{State::UNINITIALIZED};
+
+  int sock_fd_;
+  pollfd poll_fd_;
+
+  size_t buffer_size_ = 1500;
+  Endpoint host_;
+  std::optional<std::string> multicast_ip_;
+  std::optional<Endpoint> sender_;
+  std::thread receive_thread_;
+  callback_t callback_;
+
+  DropMonitor drop_monitor_;
+};
+
+}  // namespace nebula::drivers::connections
diff --git a/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/connections/tcp.hpp b/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/connections/tcp.hpp
new file mode 100644
index 00000000..aa5f4598
--- /dev/null
+++ b/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/connections/tcp.hpp
@@ -0,0 +1,103 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <boost_tcp_driver/tcp_driver.hpp>
+
+#include <boost/asio/io_context.hpp>
+
+#include <cstdint>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+namespace nebula::drivers::connections
+{
+
+class TcpError : public std::runtime_error
+{
+public:
+  explicit TcpError(const std::string & msg) : std::runtime_error(msg) {}
+};
+
+class AbstractTcpSocket
+{
+public:
+  using header_callback_t = std::function<void(const std::vector<uint8_t> &)>;
+  using payload_callback_t = std::function<void(const std::vector<uint8_t> &)>;
+  using completion_callback_t = std::function<void()>;
+
+  virtual ~AbstractTcpSocket() = default;
+
+  virtual void init(
+    const std::string & host_ip, uint16_t host_port, const std::string & remote_ip,
+    uint16_t remote_port) = 0;
+
+  virtual void bind() = 0;
+
+  virtual void close() = 0;
+
+  virtual void async_ptc_request(
+    std::vector<uint8_t> & ptc_packet, header_callback_t cb_header, payload_callback_t cb_payload,
+    completion_callback_t cb_completion) = 0;
+};
+
+class TcpSocket : public AbstractTcpSocket
+{
+public:
+  using callback_t = std::function<void(const std::vector<uint8_t> &)>;
+
+  void init(
+    const std::string & host_ip, uint16_t host_port, const std::string & remote_ip,
+    uint16_t remote_port) override
+  {
+    tcp_driver_.init_socket(remote_ip, remote_port, host_ip, host_port);
+  }
+
+  void bind() override
+  {
+    if (!tcp_driver_.isOpen() && !tcp_driver_.open()) {
+      throw TcpError("could not open TCP socket for an unknown reason");
+    }
+  }
+
+  void close() override
+  {
+    if (tcp_driver_.isOpen()) {
+      tcp_driver_.close();
+    }
+  }
+
+  void async_ptc_request(
+    std::vector<uint8_t> & ptc_packet, header_callback_t cb_header, payload_callback_t cb_payload,
+    completion_callback_t cb_completion) override
+  {
+    if (tcp_driver_.GetIOContext()->stopped()) {
+      tcp_driver_.GetIOContext()->restart();
+    }
+    bool success =
+      tcp_driver_.asyncSendReceiveHeaderPayload(ptc_packet, cb_header, cb_payload, cb_completion);
+    if (!success) {
+      throw TcpError("sending the PTC command failed for an unknown reason");
+    }
+    tcp_driver_.GetIOContext()->run();
+  }
+
+private:
+  ::drivers::tcp_driver::TcpDriver tcp_driver_{std::make_shared<boost::asio::io_context>(1)};
+};
+
+}  // namespace nebula::drivers::connections
diff --git a/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp b/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp
index d37a33e0..d5b58e42 100644
--- a/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp
+++ b/nebula_hw_interfaces/include/nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp
@@ -17,7 +17,10 @@
 // Have to define macros to silence warnings about deprecated headers being used by
 // boost/property_tree/ in some versions of boost.
 // See: https://github.com/boostorg/property_tree/issues/51
-#include "nebula_common/nebula_status.hpp"
+#include "nebula_hw_interfaces/nebula_hw_interfaces_hesai/connections/tcp.hpp"
+#include "nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp"
+
+#include <nebula_common/nebula_status.hpp>
 
 #include <boost/version.hpp>
 
@@ -28,15 +31,13 @@
 #if (BOOST_VERSION / 100 == 1074)  // Boost 1.74
 #define BOOST_ALLOW_DEPRECATED_HEADERS
 #endif
-#include "boost_tcp_driver/http_client_driver.hpp"
-#include "boost_tcp_driver/tcp_driver.hpp"
-#include "boost_udp_driver/udp_driver.hpp"
-#include "nebula_common/hesai/hesai_common.hpp"
-#include "nebula_common/hesai/hesai_status.hpp"
-#include "nebula_common/util/expected.hpp"
 #include "nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_cmd_response.hpp"
 
-#include <rclcpp/rclcpp.hpp>
+#include <boost_tcp_driver/http_client_driver.hpp>
+#include <nebula_common/hesai/hesai_common.hpp>
+#include <nebula_common/hesai/hesai_status.hpp>
+#include <nebula_common/loggers/logger.hpp>
+#include <nebula_common/util/expected.hpp>
 
 #include <boost/algorithm/string.hpp>
 #include <boost/property_tree/json_parser.hpp>
@@ -136,12 +137,11 @@ class HesaiHwInterface
 
   using ptc_cmd_result_t = nebula::util::expected<std::vector<uint8_t>, ptc_error_t>;
 
-  std::unique_ptr<::drivers::common::IoContext> cloud_io_context_;
-  std::shared_ptr<boost::asio::io_context> m_owned_ctx;
-  std::unique_ptr<::drivers::udp_driver::UdpDriver> cloud_udp_driver_;
-  std::shared_ptr<::drivers::tcp_driver::TcpDriver> tcp_driver_;
+  std::shared_ptr<loggers::Logger> logger_;
+  connections::UdpSocket udp_socket_{};
+  std::shared_ptr<connections::AbstractTcpSocket> tcp_socket_;
   std::shared_ptr<const HesaiSensorConfiguration> sensor_configuration_;
-  std::function<void(std::vector<uint8_t> & buffer)>
+  std::function<void(const std::vector<uint8_t> & buffer)>
     cloud_packet_callback_; /**This function pointer is called when the scan is complete*/
 
   std::mutex mtx_inflight_tcp_request_;
@@ -165,20 +165,6 @@ class HesaiHwInterface
   /// @param str Received string
   void str_cb(const std::string & str);
 
-  std::shared_ptr<rclcpp::Logger> parent_node_logger;
-  /// @brief Printing the string to RCLCPP_INFO_STREAM
-  /// @param info Target string
-  void PrintInfo(std::string info);
-  /// @brief Printing the string to RCLCPP_ERROR_STREAM
-  /// @param error Target string
-  void PrintError(std::string error);
-  /// @brief Printing the string to RCLCPP_DEBUG_STREAM
-  /// @param debug Target string
-  void PrintDebug(std::string debug);
-  /// @brief Printing the bytes to RCLCPP_DEBUG_STREAM
-  /// @param bytes Target byte vector
-  void PrintDebug(const std::vector<uint8_t> & bytes);
-
   /// @brief Convert an error code to a human-readable string
   /// @param error_code The error code, containing the sensor's error code (if any), along with
   /// flags such as TCP_ERROR_UNRELATED_RESPONSE etc.
@@ -202,7 +188,9 @@ class HesaiHwInterface
 
 public:
   /// @brief Constructor
-  HesaiHwInterface();
+  HesaiHwInterface(
+    std::shared_ptr<loggers::Logger> logger,
+    std::shared_ptr<connections::AbstractTcpSocket> tcp_socket);
   /// @brief Destructor
   ~HesaiHwInterface();
   /// @brief Initializing tcp_driver for TCP communication
@@ -219,7 +207,7 @@ class HesaiHwInterface
 
   /// @brief Callback function to receive the Cloud Packet data from the UDP Driver
   /// @param buffer Buffer containing the data received from the UDP socket
-  void ReceiveSensorPacketCallback(std::vector<uint8_t> & buffer);
+  void ReceiveSensorPacketCallback(const std::vector<uint8_t> & buffer);
   /// @brief Starting the interface that handles UDP streams
   /// @return Resulting status
   Status SensorInterfaceStart();
@@ -242,7 +230,7 @@ class HesaiHwInterface
   /// @brief Registering callback for PandarScan
   /// @param scan_callback Callback function
   /// @return Resulting status
-  Status RegisterScanCallback(std::function<void(std::vector<uint8_t> &)> scan_callback);
+  Status RegisterScanCallback(std::function<void(const std::vector<uint8_t> &)> scan_callback);
   /// @brief Getting data with PTC_COMMAND_GET_LIDAR_CALIBRATION
   /// @return Resulting status
   std::string GetLidarCalibrationString();
@@ -375,12 +363,6 @@ class HesaiHwInterface
   /// @return Resulting status
   HesaiLidarMonitor GetLidarMonitor();
 
-  /// @brief Call run() of IO Context
-  void IOContextRun();
-  /// @brief GetIO Context
-  /// @return IO Context
-  std::shared_ptr<boost::asio::io_context> GetIOContext();
-
   /// @brief Setting spin_speed via HTTP API
   /// @param ctx IO Context used
   /// @param rpm spin_speed (300, 600, 1200)
@@ -460,10 +442,6 @@ class HesaiHwInterface
   /// @brief Whether to use HTTP for getting LidarMonitor
   /// @return Use HTTP
   bool UseHttpGetLidarMonitor();
-
-  /// @brief Setting rclcpp::Logger
-  /// @param node Logger
-  void SetLogger(std::shared_ptr<rclcpp::Logger> node);
 };
 }  // namespace nebula::drivers
 
diff --git a/nebula_hw_interfaces/package.xml b/nebula_hw_interfaces/package.xml
index f77e8876..049df32b 100644
--- a/nebula_hw_interfaces/package.xml
+++ b/nebula_hw_interfaces/package.xml
@@ -20,6 +20,7 @@
   <depend>ros2_socketcan</depend>
   <depend>velodyne_msgs</depend>
 
+  <test_depend>ament_cmake_gmock</test_depend>
   <test_depend>ament_cmake_gtest</test_depend>
   <test_depend>ament_lint_auto</test_depend>
 
diff --git a/nebula_hw_interfaces/src/nebula_hesai_hw_interfaces/hesai_hw_interface.cpp b/nebula_hw_interfaces/src/nebula_hesai_hw_interfaces/hesai_hw_interface.cpp
index 0a08dd93..bae31975 100644
--- a/nebula_hw_interfaces/src/nebula_hesai_hw_interfaces/hesai_hw_interface.cpp
+++ b/nebula_hw_interfaces/src/nebula_hesai_hw_interfaces/hesai_hw_interface.cpp
@@ -6,6 +6,7 @@
 #include "nebula_common/hesai/hesai_status.hpp"
 #include "nebula_common/nebula_common.hpp"
 #include "nebula_common/nebula_status.hpp"
+#include "nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp"
 #include "nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_cmd_response.hpp"
 
 #include <nlohmann/json.hpp>
@@ -14,9 +15,11 @@
 #include <boost/asio/socket_base.hpp>
 
 #include <cassert>
+#include <cstddef>
 #include <sstream>
 #include <stdexcept>
 #include <string>
+#include <thread>
 
 // #define WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
 
@@ -27,17 +30,20 @@
 
 #include <boost/asio.hpp>
 
+#include <utility>
+
 namespace nebula::drivers
 {
 
 using std::string_literals::operator""s;
 using nlohmann::json;
 
-HesaiHwInterface::HesaiHwInterface()
-: cloud_io_context_{new ::drivers::common::IoContext(1)},
-  m_owned_ctx{new boost::asio::io_context(1)},
-  cloud_udp_driver_{new ::drivers::udp_driver::UdpDriver(*cloud_io_context_)},
-  tcp_driver_{new ::drivers::tcp_driver::TcpDriver(m_owned_ctx)}
+HesaiHwInterface::HesaiHwInterface(
+  std::shared_ptr<loggers::Logger> logger,
+  std::shared_ptr<connections::AbstractTcpSocket> tcp_socket)
+: logger_(std::move(logger)),
+  tcp_socket_{std::move(tcp_socket)},
+  target_model_no(NebulaModelToHesaiModelNo(SensorModel::UNKNOWN))
 {
 }
 
@@ -76,18 +82,13 @@ HesaiHwInterface::ptc_cmd_result_t HesaiHwInterface::SendReceive(
      << " (" << len << ") ";
   std::string log_tag = ss.str();
 
-  PrintDebug(log_tag + "Entering lock");
+  logger_->debug(log_tag + "Entering lock");
 
   std::timed_mutex tm;
   tm.lock();
 
-  if (tcp_driver_->GetIOContext()->stopped()) {
-    PrintDebug(log_tag + "IOContext was stopped");
-    tcp_driver_->GetIOContext()->restart();
-  }
-
-  PrintDebug(log_tag + "Sending payload");
-  tcp_driver_->asyncSendReceiveHeaderPayload(
+  logger_->debug(log_tag + "Sending payload");
+  tcp_socket_->async_ptc_request(
     send_buf,
     [this, log_tag, command_id, response_complete,
      error_code](const std::vector<uint8_t> & header_bytes) {
@@ -95,7 +96,7 @@ HesaiHwInterface::ptc_cmd_result_t HesaiHwInterface::SendReceive(
 
       size_t payload_len = (header_bytes[4] << 24) | (header_bytes[5] << 16) |
                            (header_bytes[6] << 8) | header_bytes[7];
-      PrintDebug(
+      logger_->debug(
         log_tag + "Received header (expecting " + std::to_string(payload_len) + "B payload)");
       // If command_id in the response does not match, we got a response for another command (or
       // rubbish), probably as a result of too many simultaneous TCP connections to the sensor (e.g.
@@ -109,11 +110,11 @@ HesaiHwInterface::ptc_cmd_result_t HesaiHwInterface::SendReceive(
     },
     [this, log_tag, recv_buf, response_complete,
      error_code](const std::vector<uint8_t> & payload_bytes) {
-      PrintDebug(log_tag + "Received payload");
+      logger_->debug(log_tag + "Received payload");
 
       // Header had payload length 0 (thus, header callback processed request successfully already),
       // but we still received a payload: invalid state
-      if (*response_complete == true) {
+      if (*response_complete) {
         error_code->error_flags |= TCP_ERROR_UNEXPECTED_PAYLOAD;
         return;
       }
@@ -123,19 +124,18 @@ HesaiHwInterface::ptc_cmd_result_t HesaiHwInterface::SendReceive(
       *response_complete = true;
     },
     [this, log_tag, &tm]() {
-      PrintDebug(log_tag + "Unlocking mutex");
+      logger_->debug(log_tag + "Unlocking mutex");
       tm.unlock();
-      PrintDebug(log_tag + "Unlocked mutex");
+      logger_->debug(log_tag + "Unlocked mutex");
     });
-  this->IOContextRun();
   if (!tm.try_lock_for(std::chrono::seconds(1))) {
-    PrintError(log_tag + "Request did not finish within 1s");
+    logger_->error(log_tag + "Request did not finish within 1s");
     error_code->error_flags |= TCP_ERROR_TIMEOUT;
     return *error_code;
   }
 
   if (!response_complete) {
-    PrintError(log_tag + "Did not receive response");
+    logger_->error(log_tag + "Did not receive response");
     error_code->error_flags |= TCP_ERROR_INCOMPLETE_RESPONSE;
     return *error_code;
   }
@@ -144,7 +144,7 @@ HesaiHwInterface::ptc_cmd_result_t HesaiHwInterface::SendReceive(
     return *error_code;
   }
 
-  PrintDebug(log_tag + "Received response");
+  logger_->debug(log_tag + "Received response");
 
   return *recv_buf;
 }
@@ -159,66 +159,47 @@ Status HesaiHwInterface::SetSensorConfiguration(
 
 Status HesaiHwInterface::SensorInterfaceStart()
 {
-  try {
-    PrintInfo("Starting UDP receiver");
-    if (sensor_configuration_->multicast_ip.empty()) {
-      cloud_udp_driver_->init_receiver(
-        sensor_configuration_->host_ip, sensor_configuration_->data_port);
-    } else {
-      cloud_udp_driver_->init_receiver(
-        sensor_configuration_->multicast_ip, sensor_configuration_->data_port,
-        sensor_configuration_->host_ip, sensor_configuration_->data_port);
-      cloud_udp_driver_->receiver()->setMulticast(true);
-    }
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-    PrintError("init ok");
-#endif
-    cloud_udp_driver_->receiver()->open();
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-    PrintError("open ok");
-#endif
-
-    bool success = cloud_udp_driver_->receiver()->setKernelBufferSize(UDP_SOCKET_BUFFER_SIZE);
-    if (!success) {
-      PrintError(
-        "Could not set receive buffer size. Try increasing net.core.rmem_max to " +
-        std::to_string(UDP_SOCKET_BUFFER_SIZE) + " B.");
-      return Status::ERROR_1;
-    }
+  udp_socket_.init(sensor_configuration_->host_ip, sensor_configuration_->data_port);
+  if (!sensor_configuration_->multicast_ip.empty()) {
+    udp_socket_.join_multicast_group(sensor_configuration_->multicast_ip);
+  }
 
-    cloud_udp_driver_->receiver()->bind();
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-    PrintError("bind ok");
-#endif
+  udp_socket_.set_mtu(MTU_SIZE);
 
-    cloud_udp_driver_->receiver()->asyncReceive(
-      std::bind(&HesaiHwInterface::ReceiveSensorPacketCallback, this, std::placeholders::_1));
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-    PrintError("async receive set");
-#endif
-  } catch (const std::exception & ex) {
-    Status status = Status::UDP_CONNECTION_ERROR;
-    std::cerr << status << " for " << sensor_configuration_->sensor_ip << ":"
-              << sensor_configuration_->data_port << " - " << ex.what() << std::endl;
-    return status;
+  try {
+    udp_socket_.set_socket_buffer_size(UDP_SOCKET_BUFFER_SIZE);
+  } catch (const connections::SocketError & e) {
+    throw std::runtime_error(
+      "Could not set socket receive buffer size to " + std::to_string(UDP_SOCKET_BUFFER_SIZE) +
+      ". Try increasing net.core.rmem_max.");
   }
+
+  udp_socket_.bind().subscribe([&](
+                                 const std::vector<uint8_t> & packet,
+                                 const connections::UdpSocket::ReceiveMetadata & metadata) {
+    if (metadata.drops_since_last_receive) std::cout << metadata.drops_since_last_receive << '\n';
+    ReceiveSensorPacketCallback(packet);
+  });
+
   return Status::OK;
 }
 
 Status HesaiHwInterface::RegisterScanCallback(
-  std::function<void(std::vector<uint8_t> &)> scan_callback)
+  std::function<void(const std::vector<uint8_t> &)> scan_callback)
 {
   cloud_packet_callback_ = std::move(scan_callback);
   return Status::OK;
 }
 
-void HesaiHwInterface::ReceiveSensorPacketCallback(std::vector<uint8_t> & buffer)
+void HesaiHwInterface::ReceiveSensorPacketCallback(const std::vector<uint8_t> & buffer)
 {
   cloud_packet_callback_(buffer);
 }
+
 Status HesaiHwInterface::SensorInterfaceStop()
 {
-  return Status::ERROR_1;
+  udp_socket_.unsubscribe();
+  return Status::OK;
 }
 
 Status HesaiHwInterface::GetSensorConfiguration(
@@ -226,53 +207,29 @@ Status HesaiHwInterface::GetSensorConfiguration(
 {
   std::stringstream ss;
   ss << sensor_configuration;
-  PrintDebug(ss.str());
+  logger_->debug(ss.str());
   return Status::ERROR_1;
 }
 
 Status HesaiHwInterface::GetCalibrationConfiguration(
   CalibrationConfigurationBase & calibration_configuration)
 {
-  PrintDebug(calibration_configuration.calibration_file);
+  logger_->debug(calibration_configuration.calibration_file);
   return Status::ERROR_1;
 }
 
 Status HesaiHwInterface::InitializeTcpDriver()
 {
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-  std::cout << "HesaiHwInterface::InitializeTcpDriver" << std::endl;
-  std::cout << "st: tcp_driver_->init_socket" << std::endl;
-  std::cout << "sensor_configuration_->sensor_ip=" << sensor_configuration_->sensor_ip << std::endl;
-  std::cout << "sensor_configuration_->host_ip=" << sensor_configuration_->host_ip << std::endl;
-  std::cout << "PandarTcpCommandPort=" << PandarTcpCommandPort << std::endl;
-#endif
-  tcp_driver_->init_socket(
-    sensor_configuration_->sensor_ip, PandarTcpCommandPort, sensor_configuration_->host_ip,
+  tcp_socket_->init(
+    sensor_configuration_->host_ip, PandarTcpCommandPort, sensor_configuration_->sensor_ip,
     PandarTcpCommandPort);
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-  std::cout << "ed: tcp_driver_->init_socket" << std::endl;
-#endif
-  if (!tcp_driver_->open()) {
-#ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
-    std::cout << "!tcp_driver_->open()" << std::endl;
-#endif
-    //    tcp_driver_->close();
-    tcp_driver_->closeSync();
-    return Status::ERROR_1;
-  }
+  tcp_socket_->bind();
   return Status::OK;
 }
 
 Status HesaiHwInterface::FinalizeTcpDriver()
 {
-  try {
-    if (tcp_driver_) {
-      tcp_driver_->close();
-    }
-  } catch (std::exception & e) {
-    PrintError("Error while finalizing the TcpDriver");
-    return Status::UDP_CONNECTION_ERROR;
-  }
+  tcp_socket_->close();
   return Status::OK;
 }
 
@@ -672,7 +629,7 @@ HesaiPtpConfig HesaiHwInterface::GetPtpConfig()
   if (response.size() < sizeof(HesaiPtpConfig)) {
     throw std::runtime_error("HesaiPtpConfig has unexpected payload size");
   } else if (response.size() > sizeof(HesaiPtpConfig)) {
-    PrintError("HesaiPtpConfig from Sensor has unknown format. Will parse anyway.");
+    logger_->error("HesaiPtpConfig from Sensor has unknown format. Will parse anyway.");
   }
 
   HesaiPtpConfig hesai_ptp_config;
@@ -712,16 +669,6 @@ HesaiLidarMonitor HesaiHwInterface::GetLidarMonitor()
   return CheckSizeAndParse<HesaiLidarMonitor>(response);
 }
 
-void HesaiHwInterface::IOContextRun()
-{
-  m_owned_ctx->run();
-}
-
-std::shared_ptr<boost::asio::io_context> HesaiHwInterface::GetIOContext()
-{
-  return m_owned_ctx;
-}
-
 HesaiStatus HesaiHwInterface::GetHttpClientDriverOnce(
   std::shared_ptr<boost::asio::io_context> ctx,
   std::unique_ptr<::drivers::tcp_driver::HttpClientDriver> & hcd)
@@ -735,7 +682,7 @@ HesaiStatus HesaiHwInterface::GetHttpClientDriverOnce(
     std::stringstream ss;
     ss << "HesaiHwInterface::GetHttpClientDriverOnce: " << status
        << sensor_configuration_->sensor_ip << "," << 80 << std::endl;
-    PrintError(ss.str());
+    logger_->error(ss.str());
     return Status::HTTP_CONNECTION_ERROR;
   }
   return Status::OK;
@@ -752,7 +699,7 @@ HesaiStatus HesaiHwInterface::GetHttpClientDriverOnce(
 
 void HesaiHwInterface::str_cb(const std::string & str)
 {
-  PrintInfo(str);
+  logger_->info(str);
 }
 
 std::pair<HesaiStatus, std::string> HesaiHwInterface::unwrap_http_response(
@@ -881,7 +828,7 @@ HesaiStatus HesaiHwInterface::GetLidarMonitorAsyncHttp(
   std::unique_ptr<::drivers::tcp_driver::HttpClientDriver> hcd;
   auto st = GetHttpClientDriverOnce(ctx, hcd);
   if (st != Status::OK) {
-    PrintError("HesaiHwInterface::GetLidarMonitorAsyncHttp: cannot GetHttpClientDriverOnce");
+    logger_->error("HesaiHwInterface::GetLidarMonitorAsyncHttp: cannot GetHttpClientDriverOnce");
     return st;
   }
 
@@ -891,7 +838,7 @@ HesaiStatus HesaiHwInterface::GetLidarMonitorAsyncHttp(
   boost::system::error_code ec;
   ctx->run(ec);
   if (ec) {
-    PrintError("HesaiHwInterface::GetLidarMonitorAsyncHttp: " + ec.message());
+    logger_->error("HesaiHwInterface::GetLidarMonitorAsyncHttp: " + ec.message());
   }
   return Status::WAITING_FOR_SENSOR_RESPONSE;
 }
@@ -918,15 +865,15 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
   if (sensor_configuration->return_mode != current_return_mode) {
     std::stringstream ss;
     ss << current_return_mode;
-    PrintInfo("Current LiDAR return_mode: " + ss.str());
+    logger_->info("Current LiDAR return_mode: " + ss.str());
     std::stringstream ss2;
     ss2 << sensor_configuration->return_mode;
-    PrintInfo("Current Configuration return_mode: " + ss2.str());
+    logger_->info("Current Configuration return_mode: " + ss2.str());
     std::thread t([this, sensor_configuration] {
       auto return_mode_int = nebula::drivers::int_from_return_mode_hesai(
         sensor_configuration->return_mode, sensor_configuration->sensor_model);
       if (return_mode_int < 0) {
-        PrintError(
+        logger_->error(
           "Invalid Return Mode for this sensor. Please check your settings. Falling back to Dual "
           "mode.");
         return_mode_int = 2;
@@ -939,16 +886,16 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
 
   auto current_rotation_speed = hesai_config.spin_rate;
   if (sensor_configuration->rotation_speed != current_rotation_speed.value()) {
-    PrintInfo(
+    logger_->info(
       "current lidar rotation_speed: " +
       std::to_string(static_cast<int>(current_rotation_speed.value())));
-    PrintInfo(
+    logger_->info(
       "current configuration rotation_speed: " +
       std::to_string(sensor_configuration->rotation_speed));
     if (UseHttpSetSpinRate()) {
       SetSpinSpeedAsyncHttp(sensor_configuration->rotation_speed);
     } else {
-      PrintInfo(
+      logger_->info(
         "Setting up spin rate via TCP." + std::to_string(sensor_configuration->rotation_speed));
       std::thread t(
         [this, sensor_configuration] { SetSpinRate(sensor_configuration->rotation_speed); });
@@ -969,27 +916,27 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
                              : sensor_configuration->multicast_ip;
   if (desired_host_addr != current_host_addr) {
     set_flg = true;
-    PrintInfo("current lidar dest_ipaddr: " + current_host_addr);
-    PrintInfo("current configuration host_ip: " + desired_host_addr);
+    logger_->info("current lidar dest_ipaddr: " + current_host_addr);
+    logger_->info("current configuration host_ip: " + desired_host_addr);
   }
 
   auto current_host_dport = hesai_config.dest_LiDAR_udp_port;
   if (sensor_configuration->data_port != current_host_dport.value()) {
     set_flg = true;
-    PrintInfo(
+    logger_->info(
       "current lidar dest_LiDAR_udp_port: " +
       std::to_string(static_cast<int>(current_host_dport.value())));
-    PrintInfo(
+    logger_->info(
       "current configuration data_port: " + std::to_string(sensor_configuration->data_port));
   }
 
   auto current_host_tport = hesai_config.dest_gps_udp_port;
   if (sensor_configuration->gnss_port != current_host_tport.value()) {
     set_flg = true;
-    PrintInfo(
+    logger_->info(
       "current lidar dest_gps_udp_port: " +
       std::to_string(static_cast<int>(current_host_tport.value())));
-    PrintInfo(
+    logger_->info(
       "current configuration gnss_port: " + std::to_string(sensor_configuration->gnss_port));
   }
 
@@ -1015,9 +962,9 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
       set_flg = true;
     }
     if (sync_flg && set_flg) {
-      PrintInfo("current lidar sync: " + std::to_string(hesai_config.sync));
-      PrintInfo("current lidar sync_angle: " + std::to_string(sensor_sync_angle));
-      PrintInfo("current configuration sync_angle: " + std::to_string(config_sync_angle));
+      logger_->info("current lidar sync: " + std::to_string(hesai_config.sync));
+      logger_->info("current lidar sync_angle: " + std::to_string(sensor_sync_angle));
+      logger_->info("current configuration sync_angle: " + std::to_string(config_sync_angle));
       std::thread t(
         [this, sync_flg, config_sync_angle] { SetSyncAngle(sync_flg, config_sync_angle); });
       t.join();
@@ -1031,7 +978,7 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
         sensor_configuration->sensor_model == SensorModel::HESAI_PANDARQT64 ||
         sensor_configuration->sensor_model == SensorModel::HESAI_PANDARXT32 ||
         sensor_configuration->sensor_model == SensorModel::HESAI_PANDARXT32M) {
-        PrintInfo("Trying to set Clock source to PTP");
+        logger_->info("Trying to set Clock source to PTP");
         SetClockSource(HESAI_LIDAR_PTP_CLOCK_SOURCE);
       }
       std::ostringstream tmp_ostringstream;
@@ -1039,28 +986,28 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
                         << ", Domain: " << std::to_string(sensor_configuration->ptp_domain)
                         << ", Transport: " << sensor_configuration->ptp_transport_type
                         << ", Switch Type: " << sensor_configuration->ptp_switch_type << " via TCP";
-      PrintInfo(tmp_ostringstream.str());
+      logger_->info(tmp_ostringstream.str());
       SetPtpConfig(
         static_cast<int>(sensor_configuration->ptp_profile), sensor_configuration->ptp_domain,
         static_cast<int>(sensor_configuration->ptp_transport_type),
         static_cast<int>(sensor_configuration->ptp_switch_type), PTP_LOG_ANNOUNCE_INTERVAL,
         PTP_SYNC_INTERVAL, PTP_LOG_MIN_DELAY_INTERVAL);
-      PrintDebug("Setting properties done");
+      logger_->debug("Setting properties done");
     });
-    PrintDebug("Waiting for thread to finish");
+    logger_->debug("Waiting for thread to finish");
 
     t.join();
-    PrintDebug("Thread finished");
+    logger_->debug("Thread finished");
 
     std::this_thread::sleep_for(wait_time);
   } else {  // AT128 only supports PTP setup via HTTP
-    PrintInfo("Trying to set SyncAngle via HTTP");
+    logger_->info("Trying to set SyncAngle via HTTP");
     SetSyncAngleSyncHttp(1, sensor_configuration->sync_angle);
     std::ostringstream tmp_ostringstream;
     tmp_ostringstream << "Trying to set PTP Config: " << sensor_configuration->ptp_profile
                       << ", Domain: " << sensor_configuration->ptp_domain
                       << ", Transport: " << sensor_configuration->ptp_transport_type << " via HTTP";
-    PrintInfo(tmp_ostringstream.str());
+    logger_->info(tmp_ostringstream.str());
     SetPtpConfigSyncHttp(
       static_cast<int>(sensor_configuration->ptp_profile), sensor_configuration->ptp_domain,
       static_cast<int>(sensor_configuration->ptp_transport_type), PTP_LOG_ANNOUNCE_INTERVAL,
@@ -1070,7 +1017,7 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
 #ifdef WITH_DEBUG_STDOUT_HESAI_HW_INTERFACE
   std::cout << "End CheckAndSetConfig(HesaiConfig)!!" << std::endl;
 #endif
-  PrintDebug("GetAndCheckConfig(HesaiConfig) finished");
+  logger_->debug("GetAndCheckConfig(HesaiConfig) finished");
 
   return Status::OK;
 }
@@ -1097,10 +1044,10 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
       static_cast<int>(sensor_configuration->cloud_min_angle * 10) !=
       current_cloud_min_angle_ddeg.value()) {
       set_flg = true;
-      PrintInfo(
+      logger_->info(
         "current lidar range.start: " +
         std::to_string(static_cast<int>(current_cloud_min_angle_ddeg.value())));
-      PrintInfo(
+      logger_->info(
         "current configuration cloud_min_angle: " +
         std::to_string(sensor_configuration->cloud_min_angle));
     }
@@ -1110,10 +1057,10 @@ HesaiStatus HesaiHwInterface::CheckAndSetConfig(
       static_cast<int>(sensor_configuration->cloud_max_angle * 10) !=
       current_cloud_max_angle_ddeg.value()) {
       set_flg = true;
-      PrintInfo(
+      logger_->info(
         "current lidar range.end: " +
         std::to_string(static_cast<int>(current_cloud_max_angle_ddeg.value())));
-      PrintInfo(
+      logger_->info(
         "current configuration cloud_max_angle: " +
         std::to_string(sensor_configuration->cloud_max_angle));
     }
@@ -1307,48 +1254,6 @@ bool HesaiHwInterface::UseHttpGetLidarMonitor()
   return UseHttpGetLidarMonitor(target_model_no);
 }
 
-void HesaiHwInterface::SetLogger(std::shared_ptr<rclcpp::Logger> logger)
-{
-  parent_node_logger = logger;
-}
-
-void HesaiHwInterface::PrintInfo(std::string info)
-{
-  if (parent_node_logger) {
-    RCLCPP_INFO_STREAM((*parent_node_logger), info);
-  } else {
-    std::cout << info << std::endl;
-  }
-}
-
-void HesaiHwInterface::PrintError(std::string error)
-{
-  if (parent_node_logger) {
-    RCLCPP_ERROR_STREAM((*parent_node_logger), error);
-  } else {
-    std::cerr << error << std::endl;
-  }
-}
-
-void HesaiHwInterface::PrintDebug(std::string debug)
-{
-  if (parent_node_logger) {
-    RCLCPP_DEBUG_STREAM((*parent_node_logger), debug);
-  } else {
-    std::cout << debug << std::endl;
-  }
-}
-
-void HesaiHwInterface::PrintDebug(const std::vector<uint8_t> & bytes)
-{
-  std::stringstream ss;
-  for (const auto & b : bytes) {
-    ss << static_cast<int>(b) << ", ";
-  }
-  ss << std::endl;
-  PrintDebug(ss.str());
-}
-
 std::string HesaiHwInterface::PrettyPrintPTCError(ptc_error_t error_code)
 {
   if (error_code.ok()) {
@@ -1430,9 +1335,13 @@ T HesaiHwInterface::CheckSizeAndParse(const std::vector<uint8_t> & data)
   }
 
   if (data.size() > sizeof(T)) {
-    RCLCPP_WARN_ONCE(
-      *parent_node_logger,
-      "Sensor returned longer payload than expected. Truncating and parsing anyway.");
+    // TODO(mojomex): having  a static variable for this is not optimal, but the loggers::Logger
+    // class does not support things like _ONCE macros yet
+    static bool already_warned_for_this_type = false;
+    if (!already_warned_for_this_type) {
+      logger_->warn("Sensor returned longer payload than expected. Truncating and parsing anyway.");
+      already_warned_for_this_type = true;
+    }
   }
 
   T parsed;
diff --git a/nebula_hw_interfaces/test/common/test_udp.cpp b/nebula_hw_interfaces/test/common/test_udp.cpp
new file mode 100644
index 00000000..67407724
--- /dev/null
+++ b/nebula_hw_interfaces/test/common/test_udp.cpp
@@ -0,0 +1,198 @@
+// Copyright 2024 TIER IV, Inc.
+
+#include "common/test_udp/utils.hpp"
+#include "nebula_common/util/expected.hpp"
+#include "nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp"
+
+#include <gtest/gtest.h>
+#include <gtest/internal/gtest-internal.h>
+
+#include <algorithm>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <fstream>
+#include <optional>
+#include <string>
+#include <vector>
+
+namespace nebula::drivers::connections
+{
+
+using std::chrono_literals::operator""ms;
+
+static const char localhost_ip[] = "127.0.0.1";
+static const char broadcast_ip[] = "255.255.255.255";
+static const char any_ip[] = "0.0.0.0";
+static const char multicast_group[] = "230.1.2.3";
+
+static const char sender_ip[] = "192.168.201";
+static const uint16_t sender_port = 7373;
+static const uint16_t host_port = 6262;
+
+static const std::chrono::duration send_receive_timeout = 100ms;
+
+UdpSocket::callback_t empty_cb()
+{
+  return [](const auto &, const auto &) {};
+}
+
+util::expected<uint64_t, std::string> read_sys_param(const std::string & param_fqn)
+{
+  std::string path = "/proc/sys/" + param_fqn;
+  std::replace(path.begin(), path.end(), '.', '/');
+  std::ifstream ifs{path};
+  if (!ifs) return "could not read " + param_fqn;
+
+  size_t param{};
+  if (!(ifs >> param)) return param_fqn + " has unrecognized format";
+  return param;
+}
+
+TEST(test_udp, test_basic_lifecycle)
+{
+  ASSERT_NO_THROW(
+    UdpSocket().init(localhost_ip, host_port).bind().subscribe(empty_cb()).unsubscribe());
+}
+
+TEST(test_udp, test_special_addresses_bind)
+{
+  ASSERT_NO_THROW(UdpSocket().init(broadcast_ip, host_port).bind());
+  ASSERT_NO_THROW(UdpSocket().init(any_ip, host_port).bind());
+}
+
+TEST(test_udp, test_wildcard_multicast_join_error)
+{
+  ASSERT_THROW(
+    UdpSocket().init(broadcast_ip, host_port).join_multicast_group(multicast_group).bind(),
+    SocketError);
+}
+
+TEST(test_udp, test_buffer_resize)
+{
+  auto rmem_max_maybe = read_sys_param("net.core.rmem_max");
+  if (!rmem_max_maybe.has_value()) GTEST_SKIP() << rmem_max_maybe.error();
+  size_t rmem_max = rmem_max_maybe.value();
+
+  // Setting buffer sizes up to and including rmem_max shall succeed
+  ASSERT_NO_THROW(
+    UdpSocket().init(localhost_ip, host_port).set_socket_buffer_size(rmem_max).bind());
+
+  // Linux only supports sizes up to INT32_MAX
+  ASSERT_THROW(
+    UdpSocket()
+      .init(localhost_ip, host_port)
+      .set_socket_buffer_size(static_cast<size_t>(INT32_MAX) + 1)
+      .bind(),
+    UsageError);
+}
+
+TEST(test_udp, test_correct_usage_is_enforced)
+{
+  // These functions require other functions (e.g. `init()`) to be called beforehand
+  ASSERT_THROW(UdpSocket().bind(), UsageError);
+  ASSERT_THROW(UdpSocket().join_multicast_group(multicast_group), UsageError);
+  ASSERT_THROW(UdpSocket().subscribe(empty_cb()), UsageError);
+
+  // The following functions can be called in any order, any number of times
+  ASSERT_NO_THROW(UdpSocket().limit_to_sender(sender_ip, sender_port));
+  ASSERT_NO_THROW(UdpSocket().init(localhost_ip, host_port));
+  ASSERT_NO_THROW(UdpSocket()
+                    .limit_to_sender(sender_ip, sender_port)
+                    .init(localhost_ip, host_port)
+                    .limit_to_sender(sender_ip, sender_port)
+                    .init(localhost_ip, host_port)
+                    .bind());
+
+  // Sockets cannot be re-bound
+  ASSERT_THROW(UdpSocket().init(localhost_ip, host_port).bind().bind(), UsageError);
+
+  ASSERT_THROW(
+    UdpSocket().init(localhost_ip, host_port).bind().init(localhost_ip, host_port), UsageError);
+  ASSERT_NO_THROW(
+    UdpSocket().init(localhost_ip, host_port).bind().limit_to_sender(sender_ip, sender_port));
+
+  // Only bound sockets can be subscribed
+  ASSERT_THROW(UdpSocket().init(localhost_ip, host_port).subscribe(empty_cb()), UsageError);
+  ASSERT_NO_THROW(UdpSocket().init(localhost_ip, host_port).bind().subscribe(empty_cb()));
+
+  // Only one callback can exist
+  ASSERT_THROW(
+    UdpSocket().init(localhost_ip, host_port).bind().subscribe(empty_cb()).subscribe(empty_cb()),
+    UsageError);
+
+  // But un- and re-subscribing shall be supported
+  ASSERT_NO_THROW(UdpSocket()
+                    .init(localhost_ip, host_port)
+                    .bind()
+                    .subscribe(empty_cb())
+                    .unsubscribe()
+                    .subscribe(empty_cb()));
+
+  // Unsubscribing on a non-subscribed socket shall also be supported
+  ASSERT_NO_THROW(UdpSocket().unsubscribe());
+}
+
+TEST(test_udp, test_receiving)
+{
+  const std::vector<uint8_t> payload{1, 2, 3};
+  UdpSocket sock{};
+  sock.init(localhost_ip, host_port).bind();
+
+  auto err_no_opt = udp_send(localhost_ip, host_port, payload);
+  if (err_no_opt.has_value()) GTEST_SKIP() << strerror(err_no_opt.value());
+
+  auto result_opt = receive_once(sock, send_receive_timeout);
+
+  ASSERT_TRUE(result_opt.has_value());
+  auto const & [recv_payload, metadata] = result_opt.value();
+  ASSERT_EQ(recv_payload, payload);
+  ASSERT_FALSE(metadata.truncated);
+  ASSERT_EQ(metadata.drops_since_last_receive, 0);
+
+  // TODO(mojomex): currently cannot test timestamping on loopback interface (no timestamp produced)
+}
+
+TEST(test_udp, test_receiving_oversized)
+{
+  const size_t mtu = 1500;
+  std::vector<uint8_t> payload;
+  payload.resize(mtu + 1, 0x42);
+  UdpSocket sock{};
+  sock.init(localhost_ip, host_port).set_mtu(mtu).bind();
+
+  auto err_no_opt = udp_send(localhost_ip, host_port, payload);
+  if (err_no_opt.has_value()) GTEST_SKIP() << strerror(err_no_opt.value());
+
+  auto result_opt = receive_once(sock, send_receive_timeout);
+
+  ASSERT_TRUE(result_opt.has_value());
+  auto const & [recv_payload, metadata] = result_opt.value();
+  ASSERT_EQ(recv_payload.size(), mtu);
+  ASSERT_TRUE(std::equal(recv_payload.begin(), recv_payload.end(), payload.begin()));
+  ASSERT_TRUE(metadata.truncated);
+  ASSERT_EQ(metadata.drops_since_last_receive, 0);
+}
+
+TEST(test_udp, test_filtering_sender)
+{
+  std::vector<uint8_t> payload{1, 2, 3};
+  UdpSocket sock{};
+  sock.init(localhost_ip, host_port).bind().limit_to_sender(sender_ip, sender_port);
+
+  auto err_no_opt = udp_send(localhost_ip, host_port, payload);
+  if (err_no_opt.has_value()) GTEST_SKIP() << strerror(err_no_opt.value());
+
+  auto result_opt = receive_once(sock, send_receive_timeout);
+  ASSERT_FALSE(result_opt.has_value());
+}
+
+}  // namespace nebula::drivers::connections
+
+int main(int argc, char * argv[])
+{
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+};
diff --git a/nebula_hw_interfaces/test/common/test_udp/utils.hpp b/nebula_hw_interfaces/test/common/test_udp/utils.hpp
new file mode 100644
index 00000000..dc9a08c0
--- /dev/null
+++ b/nebula_hw_interfaces/test/common/test_udp/utils.hpp
@@ -0,0 +1,78 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "nebula_hw_interfaces/nebula_hw_interfaces_common/connections/udp.hpp"
+
+#include <arpa/inet.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <cstring>
+#include <optional>
+#include <utility>
+#include <vector>
+
+std::optional<int> udp_send(
+  const char * to_ip, uint16_t to_port, const std::vector<uint8_t> & bytes)
+{
+  int sock_fd = socket(AF_INET, SOCK_DGRAM, 0);
+  if (sock_fd < 0) return errno;
+
+  int enable = 1;
+  int result = setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable));
+  if (result < 0) return errno;
+
+  sockaddr_in receiver_addr{};
+  memset(&receiver_addr, 0, sizeof(receiver_addr));
+  receiver_addr.sin_family = AF_INET;
+  receiver_addr.sin_port = htons(to_port);
+  receiver_addr.sin_addr.s_addr = inet_addr(to_ip);
+
+  result = sendto(
+    sock_fd, bytes.data(), bytes.size(), 0, reinterpret_cast<const sockaddr *>(&receiver_addr),
+    sizeof(receiver_addr));
+  if (result < 0) return errno;
+  result = close(sock_fd);
+  if (result < 0) return errno;
+  return {};
+}
+
+template <typename _T, typename _R>
+std::optional<
+  std::pair<std::vector<uint8_t>, nebula::drivers::connections::UdpSocket::ReceiveMetadata>>
+receive_once(nebula::drivers::connections::UdpSocket & sock, std::chrono::duration<_T, _R> timeout)
+{
+  std::condition_variable cv_received_result;
+  std::mutex mtx_result;
+  std::optional<
+    std::pair<std::vector<uint8_t>, nebula::drivers::connections::UdpSocket::ReceiveMetadata>>
+    result;
+
+  sock.subscribe([&](const auto & data, const auto & metadata) {
+    std::lock_guard lock(mtx_result);
+    result.emplace(data, metadata);
+    cv_received_result.notify_one();
+  });
+
+  std::unique_lock lock(mtx_result);
+  cv_received_result.wait_for(lock, timeout, [&result]() { return result.has_value(); });
+  sock.unsubscribe();
+  return result;
+}
diff --git a/nebula_hw_interfaces/test/hesai/test_ptc.cpp b/nebula_hw_interfaces/test/hesai/test_ptc.cpp
new file mode 100644
index 00000000..34e3544e
--- /dev/null
+++ b/nebula_hw_interfaces/test/hesai/test_ptc.cpp
@@ -0,0 +1,115 @@
+// Copyright 2024 TIER IV, Inc.
+
+#include "hesai/test_ptc/ptc_test.hpp"
+#include "nebula_common/hesai/hesai_common.hpp"
+#include "nebula_common/nebula_common.hpp"
+
+#include <gmock/gmock-cardinalities.h>
+#include <gmock/gmock-spec-builders.h>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <cmath>
+
+namespace nebula::drivers
+{
+
+using testing::_;
+using testing::AtLeast;
+using testing::Exactly;
+using testing::InSequence;
+
+const SensorModel g_models_under_test[] = {
+  SensorModel::HESAI_PANDAR64,      SensorModel::HESAI_PANDAR40P,  SensorModel::HESAI_PANDARQT64,
+  SensorModel::HESAI_PANDARQT128,   SensorModel::HESAI_PANDARXT32, SensorModel::HESAI_PANDARAT128,
+  SensorModel::HESAI_PANDAR128_E4X,
+};
+
+const uint16_t g_u16_invalid = 0x4242;
+const uint16_t g_ptc_port = 9347;
+const char g_host_ip[] = "192.168.42.42";
+const char g_sensor_ip[] = "192.168.84.84";
+
+auto make_sensor_config(SensorModel model)
+{
+  uint16_t rotation_speed = 600;
+  uint16_t sync_angle = 0;
+  double cut_angle = 0.0;
+  uint16_t cloud_min_angle = 0;
+  uint16_t cloud_max_angle = 360;
+
+  if (model == SensorModel::HESAI_PANDARAT128) {
+    rotation_speed = 200;
+    sync_angle = 30;
+    cut_angle = 150.0;
+    cloud_min_angle = 30;
+    cloud_max_angle = 150;
+  }
+
+  HesaiSensorConfiguration config{
+    LidarConfigurationBase{
+      EthernetSensorConfigurationBase{
+        SensorConfigurationBase{model, "test"}, g_host_ip, g_sensor_ip, g_u16_invalid},
+      ReturnMode::UNKNOWN,
+      g_u16_invalid,
+      g_u16_invalid,
+      CoordinateMode::UNKNOWN,
+      NAN,
+      NAN,
+      false,
+      {},
+      false},
+    "",
+    g_u16_invalid,
+    sync_angle,
+    cut_angle,
+    0.1,
+    "",
+    rotation_speed,
+    cloud_min_angle,
+    cloud_max_angle,
+    PtpProfile::IEEE_802_1AS_AUTO,
+    0,
+    PtpTransportType::L2,
+    PtpSwitchType::NON_TSN};
+
+  return std::make_shared<HesaiSensorConfiguration>(config);
+}
+
+TEST_P(PtcTest, ConnectionLifecycle)
+{
+  /* Constructor does not immediately connect, destructor closes socket */ {
+    auto tcp_sock_ptr = make_mock_tcp_socket();
+    auto & tcp_sock = *tcp_sock_ptr;
+
+    EXPECT_CALL(tcp_sock, close()).Times(AtLeast(1));
+    auto hw_interface = make_hw_interface(tcp_sock_ptr);
+  }
+
+  /* Full lifecycle without sending/receiving */ {
+    auto tcp_sock_ptr = make_mock_tcp_socket();
+    auto & tcp_sock = *tcp_sock_ptr;
+
+    InSequence seq;
+    EXPECT_CALL(tcp_sock, init(g_host_ip, _, g_sensor_ip, g_ptc_port)).Times(Exactly(1));
+    EXPECT_CALL(tcp_sock, bind()).Times(Exactly(1));
+    EXPECT_CALL(tcp_sock, close()).Times(AtLeast(1));
+
+    auto cfg = make_sensor_config(GetParam());
+
+    auto hw_interface = make_hw_interface(tcp_sock_ptr);
+    hw_interface->SetSensorConfiguration(cfg);
+    hw_interface->InitializeTcpDriver();
+    hw_interface->FinalizeTcpDriver();
+  }
+}
+
+INSTANTIATE_TEST_SUITE_P(TestMain, PtcTest, testing::ValuesIn(g_models_under_test));
+
+}  // namespace nebula::drivers
+
+int main(int argc, char * argv[])
+{
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+};
diff --git a/nebula_hw_interfaces/test/hesai/test_ptc/ptc_test.hpp b/nebula_hw_interfaces/test/hesai/test_ptc/ptc_test.hpp
new file mode 100644
index 00000000..0cbe4f96
--- /dev/null
+++ b/nebula_hw_interfaces/test/hesai/test_ptc/ptc_test.hpp
@@ -0,0 +1,50 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "hesai/test_ptc/tcp_mock.hpp"
+#include "nebula_common/loggers/console_logger.hpp"
+#include "nebula_common/nebula_common.hpp"
+#include "nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <utility>
+
+namespace nebula::drivers
+{
+class PtcTest : public ::testing::TestWithParam<SensorModel>
+{
+protected:
+  void SetUp() override {}
+
+  void TearDown() override {}
+
+  static auto make_mock_tcp_socket() { return std::make_shared<connections::MockTcpSocket>(); }
+
+  static auto make_hw_interface(std::shared_ptr<connections::MockTcpSocket> tcp_socket)
+  {
+    auto model = GetParam();
+
+    auto logger = std::make_shared<loggers::ConsoleLogger>("HwInterface");
+
+    auto hw_interface = std::make_unique<HesaiHwInterface>(logger, std::move(tcp_socket));
+    hw_interface->SetTargetModel(hw_interface->NebulaModelToHesaiModelNo(model));
+    return hw_interface;
+  }
+};
+
+}  // namespace nebula::drivers
diff --git a/nebula_hw_interfaces/test/hesai/test_ptc/tcp_mock.hpp b/nebula_hw_interfaces/test/hesai/test_ptc/tcp_mock.hpp
new file mode 100644
index 00000000..3916c13f
--- /dev/null
+++ b/nebula_hw_interfaces/test/hesai/test_ptc/tcp_mock.hpp
@@ -0,0 +1,47 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "nebula_hw_interfaces/nebula_hw_interfaces_hesai/connections/tcp.hpp"
+
+#include <gmock/gmock.h>
+
+#include <string>
+#include <vector>
+
+namespace nebula::drivers::connections
+{
+
+class MockTcpSocket : public AbstractTcpSocket
+{
+public:
+  MOCK_METHOD(
+    void, init,
+    (const std::string & host_ip, uint16_t host_port, const std::string & remote_ip,
+     uint16_t remote_port),
+    (override));
+
+  MOCK_METHOD(void, bind, (), (override));
+
+  MOCK_METHOD(void, close, (), (override));
+
+  MOCK_METHOD(
+    void, async_ptc_request,
+    (std::vector<uint8_t> & ptc_packet, header_callback_t cb_header, payload_callback_t cb_payload,
+     completion_callback_t cb_completion),
+    (override));
+};
+
+}  // namespace nebula::drivers::connections
diff --git a/nebula_ros/include/nebula_ros/common/rclcpp_logger.hpp b/nebula_ros/include/nebula_ros/common/rclcpp_logger.hpp
new file mode 100644
index 00000000..33343528
--- /dev/null
+++ b/nebula_ros/include/nebula_ros/common/rclcpp_logger.hpp
@@ -0,0 +1,59 @@
+// Copyright 2024 TIER IV, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <nebula_common/loggers/logger.hpp>
+#include <rclcpp/logger.hpp>
+#include <rclcpp/logging.hpp>
+
+#include <memory>
+#include <string>
+
+namespace nebula::drivers::loggers
+{
+
+class RclcppLogger : public Logger
+{
+public:
+  explicit RclcppLogger(const std::string & name) : underlying_logger_(rclcpp::get_logger(name)) {}
+  explicit RclcppLogger(const rclcpp::Logger & underlying) : underlying_logger_(underlying) {}
+
+  void debug(const std::string & message) override
+  {
+    RCLCPP_DEBUG_STREAM(underlying_logger_, message);
+  }
+  void info(const std::string & message) override
+  {
+    RCLCPP_INFO_STREAM(underlying_logger_, message);
+  }
+  void warn(const std::string & message) override
+  {
+    RCLCPP_WARN_STREAM(underlying_logger_, message);
+  }
+  void error(const std::string & message) override
+  {
+    RCLCPP_ERROR_STREAM(underlying_logger_, message);
+  }
+
+  std::shared_ptr<Logger> child(const std::string & name) override
+  {
+    return std::make_shared<RclcppLogger>(underlying_logger_.get_child(name));
+  }
+
+private:
+  rclcpp::Logger underlying_logger_;
+};
+
+}  // namespace nebula::drivers::loggers
diff --git a/nebula_ros/include/nebula_ros/hesai/hesai_ros_wrapper.hpp b/nebula_ros/include/nebula_ros/hesai/hesai_ros_wrapper.hpp
index f8d221c3..14fba275 100644
--- a/nebula_ros/include/nebula_ros/hesai/hesai_ros_wrapper.hpp
+++ b/nebula_ros/include/nebula_ros/hesai/hesai_ros_wrapper.hpp
@@ -51,7 +51,11 @@ class HesaiRosWrapper final : public rclcpp::Node
 
 public:
   explicit HesaiRosWrapper(const rclcpp::NodeOptions & options);
-  ~HesaiRosWrapper() noexcept override = default;
+  ~HesaiRosWrapper() noexcept override
+  {
+    if (!hw_interface_wrapper_) return;
+    hw_interface_wrapper_->hw_interface()->SensorInterfaceStop();
+  };
 
   /// @brief Get current status of this driver
   /// @return Current status
@@ -62,7 +66,7 @@ class HesaiRosWrapper final : public rclcpp::Node
   Status stream_start();
 
 private:
-  void receive_cloud_packet_callback(std::vector<uint8_t> & packet);
+  void receive_cloud_packet_callback(const std::vector<uint8_t> & packet);
 
   void receive_scan_message_callback(std::unique_ptr<pandar_msgs::msg::PandarScan> scan_msg);
 
diff --git a/nebula_ros/include/nebula_ros/hesai/hw_interface_wrapper.hpp b/nebula_ros/include/nebula_ros/hesai/hw_interface_wrapper.hpp
index 4b2572b1..399d92f4 100644
--- a/nebula_ros/include/nebula_ros/hesai/hw_interface_wrapper.hpp
+++ b/nebula_ros/include/nebula_ros/hesai/hw_interface_wrapper.hpp
@@ -15,6 +15,7 @@
 #pragma once
 
 #include <nebula_common/hesai/hesai_common.hpp>
+#include <nebula_hw_interfaces/nebula_hw_interfaces_hesai/connections/tcp.hpp>
 #include <nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp>
 #include <rclcpp/rclcpp.hpp>
 
diff --git a/nebula_ros/src/hesai/hesai_ros_wrapper.cpp b/nebula_ros/src/hesai/hesai_ros_wrapper.cpp
index 73869d1f..da9e2477 100644
--- a/nebula_ros/src/hesai/hesai_ros_wrapper.cpp
+++ b/nebula_ros/src/hesai/hesai_ros_wrapper.cpp
@@ -8,6 +8,7 @@
 #include <nebula_common/nebula_common.hpp>
 #include <nebula_decoders/nebula_decoders_common/angles.hpp>
 
+#include <algorithm>
 #include <cstdint>
 #include <filesystem>
 #include <memory>
@@ -40,7 +41,7 @@ HesaiRosWrapper::HesaiRosWrapper(const rclcpp::NodeOptions & options)
   RCLCPP_INFO_STREAM(get_logger(), "Sensor Configuration: " << *sensor_cfg_ptr_);
 
   launch_hw_ = declare_parameter<bool>("launch_hw", param_read_only());
-  bool use_udp_only = declare_parameter<bool>("udp_only", param_read_only());
+  bool use_udp_only = declare_parameter<bool>("udp_only", false, param_read_only());
 
   if (use_udp_only) {
     RCLCPP_INFO_STREAM(
@@ -224,6 +225,13 @@ Status HesaiRosWrapper::validate_and_set_config(
   if (new_config->frame_id.empty()) {
     return Status::SENSOR_CONFIG_ERROR;
   }
+  if (!new_config->multicast_ip.empty() && new_config->host_ip == "255.255.255.255") {
+    RCLCPP_ERROR(
+      get_logger(),
+      "A concrete host IP must be given when multicast is enabled, otherwise the correct network "
+      "interface cannot be determined.");
+    return Status::SENSOR_CONFIG_ERROR;
+  }
   if (new_config->ptp_profile == nebula::drivers::PtpProfile::UNKNOWN_PROFILE) {
     RCLCPP_ERROR(
       get_logger(), "Invalid PTP Profile Provided. Please use '1588v2', '802.1as' or 'automotive'");
@@ -413,7 +421,7 @@ rcl_interfaces::msg::SetParametersResult HesaiRosWrapper::on_parameter_change(
   return rcl_interfaces::build<SetParametersResult>().successful(true).reason("");
 }
 
-void HesaiRosWrapper::receive_cloud_packet_callback(std::vector<uint8_t> & packet)
+void HesaiRosWrapper::receive_cloud_packet_callback(const std::vector<uint8_t> & packet)
 {
   if (!decoder_wrapper_ || decoder_wrapper_->status() != Status::OK) {
     return;
@@ -426,7 +434,7 @@ void HesaiRosWrapper::receive_cloud_packet_callback(std::vector<uint8_t> & packe
   auto msg_ptr = std::make_unique<nebula_msgs::msg::NebulaPacket>();
   msg_ptr->stamp.sec = static_cast<int>(timestamp_ns / 1'000'000'000);
   msg_ptr->stamp.nanosec = static_cast<int>(timestamp_ns % 1'000'000'000);
-  msg_ptr->data.swap(packet);
+  msg_ptr->data = packet;
 
   if (!packet_queue_.try_push(std::move(msg_ptr))) {
     RCLCPP_ERROR_THROTTLE(get_logger(), *get_clock(), 500, "Packet(s) dropped");
diff --git a/nebula_ros/src/hesai/hw_interface_wrapper.cpp b/nebula_ros/src/hesai/hw_interface_wrapper.cpp
index b5f99109..ef558f15 100644
--- a/nebula_ros/src/hesai/hw_interface_wrapper.cpp
+++ b/nebula_ros/src/hesai/hw_interface_wrapper.cpp
@@ -2,7 +2,11 @@
 
 #include "nebula_ros/hesai/hw_interface_wrapper.hpp"
 
+#include "nebula_hw_interfaces/nebula_hw_interfaces_hesai/hesai_hw_interface.hpp"
 #include "nebula_ros/common/parameter_descriptors.hpp"
+#include "nebula_ros/common/rclcpp_logger.hpp"
+
+#include <memory>
 
 namespace nebula::ros
 {
@@ -10,8 +14,10 @@ namespace nebula::ros
 HesaiHwInterfaceWrapper::HesaiHwInterfaceWrapper(
   rclcpp::Node * const parent_node,
   std::shared_ptr<const nebula::drivers::HesaiSensorConfiguration> & config, bool use_udp_only)
-: hw_interface_(new nebula::drivers::HesaiHwInterface()),
-  logger_(parent_node->get_logger().get_child("HwInterface")),
+: hw_interface_(std::make_shared<drivers::HesaiHwInterface>(
+    drivers::loggers::RclcppLogger(parent_node->get_logger()).child("HwInterface"),
+    std::move(std::make_unique<drivers::connections::TcpSocket>()))),
+  logger_(parent_node->get_logger().get_child("HwInterfaceWrapper")),
   status_(Status::NOT_INITIALIZED),
   use_udp_only_(use_udp_only)
 {
@@ -26,7 +32,6 @@ HesaiHwInterfaceWrapper::HesaiHwInterfaceWrapper(
       (std::stringstream{} << "Could not initialize HW interface: " << status_).str());
   }
 
-  hw_interface_->SetLogger(std::make_shared<rclcpp::Logger>(parent_node->get_logger()));
   hw_interface_->SetTargetModel(config->sensor_model);
 
   if (use_udp_only) {