tcp_connection.h

Go to the documentation of this file.

/* Diagnostic Client library
 * Copyright (C) 2024  Avijit Dey
 * 
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */
#ifndef DIAG_CLIENT_LIB_LIB_BOOST_SUPPORT_SRC_BOOST_SUPPORT_CONNECTION_TCP_TCP_CONNECTION_H_
#define DIAG_CLIENT_LIB_LIB_BOOST_SUPPORT_SRC_BOOST_SUPPORT_CONNECTION_TCP_TCP_CONNECTION_H_
 
#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <string_view>
#include <utility>
 
#include "boost-support/error_domain/boost_support_error_domain.h"
#include "core/include/result.h"
#include "utility/thread.h"
 
namespace boost_support {
namespace connection {
namespace tcp {
 
enum class ConnectionType : std::uint8_t { kClient, kServer };
 
template<ConnectionType Connection, typename Socket>
class TcpConnection;
 
template<typename Socket>
class TcpConnection<ConnectionType::kClient, Socket> final {
 public:
  using TcpMessage = typename Socket::TcpMessage;
 
  using TcpMessagePtr = typename Socket::TcpMessagePtr;
 
  using TcpMessageConstPtr = typename Socket::TcpMessageConstPtr;
 
  using HandlerRead = std::function<void(TcpMessagePtr)>;
 
 public:
  explicit TcpConnection(std::string_view connection_name, Socket socket) noexcept
      : socket_{std::move(socket)},
        handler_read_{},
        exit_request_{false},
        running_{false},
        cond_var_{},
        connection_name_{connection_name},
        thread_{},
        mutex_{} {}
 
  TcpConnection(const TcpConnection &other) noexcept = delete;
  TcpConnection &operator=(const TcpConnection &other) &noexcept = delete;
 
  TcpConnection(TcpConnection &&other) noexcept = default;
  TcpConnection &operator=(TcpConnection &&other) &noexcept = default;
 
  ~TcpConnection() noexcept = default;
 
  void SetReadHandler(HandlerRead read_handler) { handler_read_ = std::move(read_handler); }
 
  void Initialize() noexcept {
    // Open socket
    socket_.Open();
    // Start thread to receive messages
    thread_ = utility::thread::Thread{
        connection_name_, [this]() {
          std::unique_lock<std::mutex> lck(mutex_);
          while (!exit_request_) {
            if (!running_) {
              cond_var_.wait(lck, [this]() { return exit_request_ || running_; });
            }
            if (!exit_request_.load() && running_) {
              lck.unlock();
              running_ = ReadMessage();
              lck.lock();
            }
          }
        }};
  }
 
  void DeInitialize() noexcept {
    socket_.Close();
    {
      std::unique_lock<std::mutex> lck(mutex_);
      exit_request_ = true;
      running_ = false;
    }
    cond_var_.notify_all();
    thread_.Join();
  }
 
  auto ConnectToHost(std::string_view host_ip_address, std::uint16_t host_port_num) noexcept
      -> bool {
    return socket_.Connect(host_ip_address, host_port_num)
        .AndThen([this]() noexcept {
          {  // start reading
            std::lock_guard<std::mutex> lock{mutex_};
            running_ = true;
            cond_var_.notify_all();
          }
        })
        .HasValue();
  }
 
  void DisconnectFromHost() noexcept {
    {  // stop reading
      std::lock_guard<std::mutex> lock{mutex_};
      running_ = false;
    }
    socket_.Disconnect();
  }
 
  core_type::Result<void> Transmit(TcpMessageConstPtr message) noexcept {
    return socket_.Transmit(std::move(message))
        .AndThen([]() { return core_type::Result<void>::FromValue(); })
        .MapError([](typename Socket::SocketError const &) {
          return error_domain::MakeErrorCode(error_domain::BoostSupportErrorErrc::kSocketError);
        });
  }
 
 private:
  Socket socket_;
 
  HandlerRead handler_read_;
 
  std::atomic_bool exit_request_;
 
  std::atomic_bool running_;
 
  std::condition_variable cond_var_;
 
  std::string connection_name_;
 
  utility::thread::Thread thread_;
 
  std::mutex mutex_;
 
 private:
  auto ReadMessage() -> bool {
    // Try reading from socket
    return socket_.Read()
        .AndThen([this](TcpMessagePtr tcp_message) {
          if (handler_read_) { handler_read_(std::move(tcp_message)); }
          return core_type::Result<void, typename Socket::SocketError>::FromValue();
        })
        .HasValue();
  }
};
 
template<typename Socket>
class TcpConnection<ConnectionType::kServer, Socket> final {
 public:
  using TcpMessage = typename Socket::TcpMessage;
 
  using TcpMessagePtr = typename Socket::TcpMessagePtr;
 
  using TcpMessageConstPtr = typename Socket::TcpMessageConstPtr;
 
  using HandlerRead = std::function<void(TcpMessagePtr)>;
 
 public:
  explicit TcpConnection(std::string_view connection_name, Socket socket) noexcept
      : socket_{std::move(socket)},
        handler_read_{},
        exit_request_{false},
        running_{false},
        cond_var_{},
        connection_name_{connection_name},
        thread_{},
        mutex_{} {}
 
  TcpConnection(const TcpConnection &other) noexcept = delete;
  TcpConnection &operator=(const TcpConnection &other) &noexcept = delete;
 
  TcpConnection(TcpConnection &&other) noexcept
      : socket_{std::move(other.socket_)},
        handler_read_{std::move(other.handler_read_)},
        exit_request_{other.exit_request_.load()},
        running_{other.running_.load()},
        thread_{std::move(other.thread_)} {}
 
  TcpConnection &operator=(TcpConnection &&other) &noexcept {
    socket_ = std::move(other.socket_);
    handler_read_ = std::move(other.handler_read_);
    exit_request_.store(other.exit_request_.load());
    running_.store(other.running_.load());
    thread_ = std::move(other.thread_);
    return *this;
  }
 
  ~TcpConnection() noexcept = default;
 
  void SetReadHandler(HandlerRead read_handler) { handler_read_ = std::move(read_handler); }
 
  void Initialize() noexcept {
    // Start thread to receive messages
    thread_ = utility::thread::Thread(connection_name_, [this]() {
      std::unique_lock<std::mutex> lck(mutex_);
      while (!exit_request_) {
        if (!running_) {
          cond_var_.wait(lck, [this]() { return exit_request_ || running_; });
        }
        if (!exit_request_ && running_) {
          lck.unlock();
          running_ = ReadMessage();
          lck.lock();
        }
      }
    });
    {  // start reading
      std::lock_guard<std::mutex> lock{mutex_};
      running_ = true;
      cond_var_.notify_all();
    }
  }
 
  void DeInitialize() noexcept {
    socket_.Close();
    {
      std::unique_lock<std::mutex> lck(mutex_);
      exit_request_ = true;
      running_ = false;
    }
    cond_var_.notify_all();
    thread_.Join();
  }
 
  core_type::Result<void> Transmit(TcpMessageConstPtr message) noexcept {
    return socket_.Transmit(std::move(message))
        .AndThen([]() { return core_type::Result<void>::FromValue(); })
        .MapError([](typename Socket::SocketError const &) {
          return error_domain::MakeErrorCode(error_domain::BoostSupportErrorErrc::kSocketError);
        });
  }
 
 private:
  Socket socket_;
 
  HandlerRead handler_read_;
 
  std::atomic_bool exit_request_;
 
  std::atomic_bool running_;
 
  std::condition_variable cond_var_;
 
  std::string connection_name_;
 
  utility::thread::Thread thread_;
 
  std::mutex mutex_;
 
 private:
  auto ReadMessage() noexcept -> bool {
    // Try reading from socket
    return socket_.Read()
        .AndThen([this](TcpMessagePtr tcp_message) {
          if (handler_read_) { handler_read_(std::move(tcp_message)); }
          return core_type::Result<void, typename Socket::SocketError>::FromValue();
        })
        .HasValue();
  }
};
 
}  // namespace tcp
}  // namespace connection
}  // namespace boost_support
#endif  // DIAG_CLIENT_LIB_LIB_BOOST_SUPPORT_SRC_BOOST_SUPPORT_CONNECTION_TCP_TCP_CONNECTION_H_