tcp_connection.h

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/ AERA
//_/_/ Autocatalytic Endogenous Reflective Architecture
//_/_/ 
//_/_/ Copyright (c) 2018-2025 Jeff Thompson
//_/_/ Copyright (c) 2018-2025 Kristinn R. Thorisson
//_/_/ Copyright (c) 2018-2025 Icelandic Institute for Intelligent Machines
//_/_/ Copyright (c) 2021-2025 Leonard Eberding
//_/_/ http://www.iiim.is
//_/_/
//_/_/ --- Open-Source BSD License, with CADIA Clause v 1.0 ---
//_/_/
//_/_/ Redistribution and use in source and binary forms, with or without
//_/_/ modification, is permitted provided that the following conditions
//_/_/ are met:
//_/_/ - Redistributions of source code must retain the above copyright
//_/_/   and collaboration notice, this list of conditions and the
//_/_/   following disclaimer.
//_/_/ - Redistributions in binary form must reproduce the above copyright
//_/_/   notice, this list of conditions and the following disclaimer 
//_/_/   in the documentation and/or other materials provided with 
//_/_/   the distribution.
//_/_/
//_/_/ - Neither the name of its copyright holders nor the names of its
//_/_/   contributors may be used to endorse or promote products
//_/_/   derived from this software without specific prior 
//_/_/   written permission.
//_/_/   
//_/_/ - CADIA Clause: The license granted in and to the software 
//_/_/   under this agreement is a limited-use license. 
//_/_/   The software may not be used in furtherance of:
//_/_/    (i)   intentionally causing bodily injury or severe emotional 
//_/_/          distress to any person;
//_/_/    (ii)  invading the personal privacy or violating the human 
//_/_/          rights of any person; or
//_/_/    (iii) committing or preparing for any act of war.
//_/_/
//_/_/ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND 
//_/_/ CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, 
//_/_/ INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF 
//_/_/ MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 
//_/_/ DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR 
//_/_/ CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 
//_/_/ SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 
//_/_/ BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
//_/_/ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
//_/_/ INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, 
//_/_/ WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/ NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 
//_/_/ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY 
//_/_/ OF SUCH DAMAGE.
//_/_/ 
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
 
#pragma once
 
#pragma comment (lib, "Ws2_32.lib")
 
#include <queue>
#include <mutex>
#include <condition_variable>
#if defined(_WIN32)
#include <winsock2.h>
#include <windows.h>
#include <ws2tcpip.h>
#endif
#include <string>
#include <stdlib.h>
#include <stdio.h>
#include <thread>
#include <bitset>
 
#include "tcp_data_message.pb.h"
 
namespace tcp_io_device {
 
  class SafeQueue
  {
  public:
    SafeQueue()
      : queue_()
      , mutex_()
    {
      max_elements_ = 1;
    }
 
    SafeQueue(int max_elements)
      : queue_()
      , mutex_()
    {
      max_elements_ = max_elements;
    }
 
    ~SafeQueue()
    {}
 
    void enqueue(std::unique_ptr<TCPMessage> t)
    {
      std::lock_guard<std::recursive_mutex> lock(mutex_);
      while (queue_.size() >= max_elements_) {
        if (queue_.front()->messagetype() == TCPMessage_Type_DATA) {
          queue_.front()->release_datamessage();
        }
        queue_.pop();
      }
      queue_.push(std::move(t));
    }
 
 
    std::unique_ptr<TCPMessage> dequeue()
    {
      std::lock_guard<std::recursive_mutex> lock(mutex_);
      if (queue_.empty()) {
        return NULL;
      }
      std::unique_ptr<TCPMessage> val = std::move(queue_.front());
      queue_.pop();
      return val;
    }
 
    void clear() {
      std::lock_guard<std::recursive_mutex> lock(mutex_);
      while (queue_.size() > 0) {
        if (queue_.front()->messagetype() == TCPMessage_Type_DATA) {
          queue_.front()->release_datamessage();
        }
        queue_.pop();
      }
    }
 
 
 
  private:
    std::queue<std::unique_ptr<TCPMessage>> queue_;
    mutable std::recursive_mutex mutex_;
    int max_elements_;
  };
 
 
  class TCPConnection {
 
  public:
 
    static std::map<int, std::string> type_to_name_map_;
 
    TCPConnection(std::shared_ptr<SafeQueue> receive_queue, std::shared_ptr<SafeQueue> send_queue, uint64_t msg_length_buf_size);
    ~TCPConnection();
 
    int listenAndAwaitConnection(std::string port);
 
    int establishConnection(std::string host, std::string port);
 
    void start();
 
    void stop();
 
    bool isRunning() { return state_ == RUNNING; }
 
#if defined(_WIN32)
    static int receiveIsReady(SOCKET fd);
#else
    static int receiveIsReady(int fd);
#endif
 
  protected:
 
    typedef enum {
      NOT_STARTED = 0,
      RUNNING = 1,
      STOPPED = 2,
    }State;
 
    typedef enum {
      CLIENT = 0,
      SERVER = 1,
    }SocketType;
 
    State state_;
    SocketType socket_type_;
 
    std::shared_ptr<std::thread> tcp_background_thread_;
 
#if defined(_WIN32)
    SOCKET tcp_socket_;
    SOCKET server_listen_socket_;
#else
    int tcp_socket_;
    int server_listen_socket_;
#endif
 
    uint64_t msg_buf_size_;
    uint64_t msg_length_buf_size_;
 
    std::shared_ptr<SafeQueue> incoming_queue_;
    std::shared_ptr<SafeQueue> outgoing_queue_;
 
    std::string host_;
    std::string port_;
 
    void tcpBackgroundHandler();
 
    std::unique_ptr<TCPMessage> receiveMessage();
 
    int sendMessage(std::unique_ptr<TCPMessage> msg);
  };
 
} // namespace tcp_io_device