ConnectionHub.cpp

#include "ConnectionHub.h"
 
#if !defined( EMSCRIPTEN )
 
#include "../Logging/Logger.h"
 
#include <chrono>
#include <thread>
 
using namespace std::chrono_literals;
 
namespace {
  Cogs::Logging::Log logger = Cogs::Logging::getLogger("ConnectionHub");
}
 
bool Cogs::Network::ConnectionHub::processOutgoing() {
  if (isConnected()) {
    for (;;) {
      if (!outgoingMessage) {
        outgoingMessage = getNextMessage();
 
        if (outgoingMessage) {
          Header    header = { outgoingMessage->getID(), outgoingMessage->size() };
          char*     ptr = reinterpret_cast<char*>(&header);
          int       retries = 1000;
 
          for (uint64_t remaining = sizeof(header); remaining; ) {
            uint64_t  bytessent;
 
            if (ConnectionTCP::send(ptr, remaining, bytessent)) {
              if (bytessent) {
                remaining -= bytessent;
                ptr += bytessent;
              }
              else {
                if (--retries <= 0) {
                  outgoingMessage = nullptr;
                  LOG_ERROR(logger, "Failed to send message header to remote host. Forcing disconnect.");
                  disconnect();
                  return false;
                }
                std::this_thread::sleep_for(1ms);
              }
            }
            else {
              outgoingMessage = nullptr;
              return false;
            }
          }
        }
        else {
          return true;
        }
      }
      if (bytesSent < outgoingMessage->size()) {
        uint64_t    justSent;
        const char* ptr = reinterpret_cast<const char*>(outgoingMessage->data()) + bytesSent;
 
        if (ConnectionTCP::send(ptr, outgoingMessage->size() - bytesSent, justSent)) {
          if (justSent) {
            bytesSent += justSent;
          }
          else {
            return true;
          }
        }
        else {
          outgoingMessage = nullptr;
          bytesSent = 0;
          return false;
        }
      }
      if (bytesSent >= outgoingMessage->size()) {
        outgoingMessage = nullptr;
        bytesSent = 0;
      }
    }
  }
  outgoingMessage = nullptr;
  bytesSent = 0;
 
  while (getNextMessage()) {
  }
  return false;
}
 
bool Cogs::Network::ConnectionHub::processIncoming() {
  if (!incomingMessage) {
    Header  header;
    char*   dest = reinterpret_cast<char*>(&header);
 
    for (uint64_t remaining = sizeof(header); remaining; ) {
      uint64_t  bytesReceived;
 
      if (ConnectionTCP::receive(dest, remaining, bytesReceived)) {
        remaining -= bytesReceived;
        dest += bytesReceived;
      }
      else {
        return false;
      }
    }
    incomingMessage = Message::allocate(header.id);
    incomingMessage->resize(header.size, false);
  }
 
  uint64_t  remaining = incomingMessage->unreadSize();
  char*     dest = reinterpret_cast<char*>(incomingMessage->data()) + incomingMessage->pos();
 
  while (remaining) {
    uint64_t  bytesReceived;
 
    if (ConnectionTCP::receive(dest, remaining, bytesReceived)) {
      if (bytesReceived) {
        incomingMessage->seek(bytesReceived, Cogs::Memory::MemoryBuffer::Anchor::Current);
        remaining -= bytesReceived;
        dest += bytesReceived;
      }
      else {
        break;
      }
    }
    else {
      return false;
    }
  }
  if (!remaining) {
    incomingMessage->seek(0, Cogs::Memory::MemoryBuffer::Anchor::Start);
    handleReceivedMessage(incomingMessage);
    incomingMessage = nullptr;
  }
  return true;
}
 
void Cogs::Network::ConnectionHub::handleReceivedMessage(const Message::Ptr& message) {
  sendMessage(message);
}
 
bool Cogs::Network::ConnectionHub::disconnect() {
  incomingMessage = nullptr;
  outgoingMessage = nullptr;
  bytesSent = 0;
 
  return ConnectionTCP::disconnect();
}
 
#endif