/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.
 */

#include "UDPIpc.hpp"

#include <sstream>
#include <string>
#include <vector>

#include "FwkStrCvt.hpp"
#include "GsRandom.hpp"
#include "config.h"

namespace apache {
namespace geode {
namespace client {
namespace testframework {

bool UDPMessage::ping(ACE_SOCK_Dgram& io, ACE_INET_Addr& who) {
  clear();
  setCmd(ACK_REQUEST);
  setSender(who);
  return send(io);
}

std::string UDPMessage::dump(int32_t max) {
  char buf[1024];
  std::string dmp("Dump of message parts: ");
  sprintf(buf, "\nTag: %u\nCmd: %u  %s\nId: %u\nLength: %u\nMessage:\n",
          m_hdr.tag, m_hdr.cmd, cmdString(m_hdr.cmd), m_hdr.id,
          ntohl(m_hdr.length));
  dmp += buf;
  if (!m_msg.empty()) {
    if (max > 0) {
      dmp += m_msg.substr(0, max);
    } else {
      dmp += m_msg;
    }
  } else {
    dmp += "No Message.";
  }
  dmp += "\nEnd of Dump";
  return dmp;
}

bool UDPMessage::receiveFrom(ACE_SOCK_Dgram& io,
                             const ACE_Time_Value* timeout) {
  bool ok = true;
  ACE_Time_Value wait(2);
  if (!timeout) {
    timeout = &wait;
  }
  iovec buffs;
  int32_t red = static_cast<int32_t>(io.recv(&buffs, m_sender, 0, timeout));
  if (red < 0) {
    if (errno != ETIME) {
      FWKEXCEPTION("UDPMessage::receiveFrom: Failed, errno: " << errno);
    }
    return false;
  }
  uint32_t len = buffs.iov_len;
  if (len < UDP_HEADER_SIZE) {  // We must at least have a header
    FWKEXCEPTION("UDPMessage::receiveFrom: Failed, header length: " << len);
  }
  clear();
  memcpy(&m_hdr, buffs.iov_base, UDP_HEADER_SIZE);
  if (m_hdr.tag != UDP_MSG_TAG) {
    FWKEXCEPTION("UDPMessage::receiveFrom: Failed, invalid tag: " << m_hdr.tag);
  }
  char* ptr = reinterpret_cast<char*>(buffs.iov_base);
  ptr += UDP_HEADER_SIZE;
  len -= UDP_HEADER_SIZE;
  uint32_t sent = ntohl(m_hdr.length);
  if (sent != len) {
    FWKEXCEPTION("UDPMessage::receiveFrom: Failed, expected "
                 << sent << " bytes, received " << len);
  }
  if (len > 0) {
    m_msg = std::string(ptr, len);
  }
  delete[] reinterpret_cast<char*>(buffs.iov_base);
  // FWKINFO( "UDPMessage::receiveFrom: " << dump( 50 ) );;
  if (needToAck()) {
    UDPMessage ack(ACK);
    ok = (ok && ack.sendTo(io, m_sender));
  }
  return ok;
}

bool UDPMessage::sendTo(ACE_SOCK_Dgram& io, ACE_INET_Addr& who) {
  setSender(who);
  return send(io);
}

bool UDPMessage::send(ACE_SOCK_Dgram& io) {
  bool ok = true;
  int32_t tot = 0;
  int32_t vcnt = 1;
  iovec buffs[2];
  m_hdr.length = 0;
  buffs[0].iov_base = reinterpret_cast<char*>(&m_hdr);
  buffs[0].iov_len = UDP_HEADER_SIZE;
  tot += UDP_HEADER_SIZE;
  if (!m_msg.empty()) {
    auto len = static_cast<uint32_t>(m_msg.size());
    m_hdr.length = htonl(len);
    buffs[1].iov_base = const_cast<char*>(m_msg.c_str());
    buffs[1].iov_len = len;
    vcnt = 2;
    tot += len;
  }
  // FWKINFO( "UDPMessage::send: " << dump( 50 ) );;
  int32_t sent = static_cast<int32_t>(io.send(buffs, vcnt, m_sender));
  if (sent < 0) {
    FWKEXCEPTION("UDPMessage::send: Failed, errno: " << errno);
  }
  if (sent != tot) {
    ok = false;
    FWKSEVERE("UDPMessage::send: Failed to completely send, " << sent << ", "
                                                              << tot);
  }
  if (needToAck()) {
    UDPMessage ack;
    ok = (ok && ack.receiveFrom(io));
  }
  return ok;
}

UDPMessageClient::UDPMessageClient(std::string server)
    : m_server(server.c_str()) {
  int32_t result = -1;
  int32_t tries = 100;
  ACE_INET_Addr* client = new ACE_INET_Addr();
  while ((result < 0) && (tries > 0)) {
    uint32_t port = GsRandom::random(1111u, 31111u) + tries;
    client->set(port, "localhost");
    result = m_io.open(*client);
  }
  delete client;
  if (result < 0) {
    FWKEXCEPTION("Client failed to open io, " << errno);
  }
  const ACE_Time_Value timeout(20);
  UDPMessage msg;
  tries = 3;
  bool pingMsg = false;
  while (!pingMsg && tries-- > 0) {
    try {
      pingMsg = msg.ping(m_io, m_server);
    } catch (...) {
      continue;
    }
  }
  if (pingMsg) {
    // if ( msg.ping( m_io, m_server ) ) {
    msg.setSender(m_server);
    bool connectionOK = false;
    tries = 10;
    while (!connectionOK && (--tries > 0)) {
      try {
        msg.clear();
        msg.setCmd(ADDR_REQUEST);
        if (msg.sendTo(m_io, m_server)) {
          if (msg.receiveFrom(m_io, &timeout)) {
            std::string newConn = msg.what();
            ACE_INET_Addr conn(newConn.c_str());
            if (msg.ping(m_io, conn)) {  // We have a working addr
              m_server = conn;
              connectionOK = true;
              tries = 0;
            }
          }
        }
      } catch (...) {
        continue;
      }
    }
    if (!connectionOK) {
      FWKEXCEPTION(
          "UDPMessageClient failed to establish connection to server.");
    }
  } else {
    FWKEXCEPTION("Failed to contact " << server);
  }
}

int32_t Receiver::doTask() {
  auto msg = std::unique_ptr<UDPMessage>(new UDPMessage());
  UDPMessage cmsg;
  try {
    while (*m_run) {
      if (isListener()) {
        cmsg.clear();
        ACE_Time_Value wait(30);  // Timeout is relative time.
        if (cmsg.receiveFrom(*m_io, &wait)) {
          if (cmsg.getCmd() == ADDR_REQUEST) {
            auto&& addr = m_addrs.front();
            m_addrs.pop_front();
            m_addrs.push_back(addr);
            cmsg.clear();
            cmsg.setMessage(addr);
            cmsg.setCmd(ADDR_RESPONSE);
            cmsg.send(*m_io);
          }
        }
      } else {
        msg->clear();
        ACE_Time_Value timeout(2);
        if (msg->receiveFrom(
                *m_io, &timeout)) {  // Timeout is relative time, send ack.
          if (msg->getCmd() == ADDR_REQUEST) {
            auto&& addr = m_addrs.front();
            m_addrs.pop_front();
            m_addrs.push_back(addr);
            cmsg.clear();
            cmsg.setMessage(addr);
            cmsg.setCmd(ADDR_RESPONSE);
            cmsg.send(*m_io);
          }
          if (msg->length() > 0) {
            m_queues->putInbound(msg.release());
            msg.reset(new UDPMessage());
          }
        }
      }
    }
  } catch (FwkException& ex) {
    FWKSEVERE("Receiver::doTask() caught exception: " << ex.what());
  } catch (...) {
    FWKSEVERE("Receiver::doTask() caught unknown exception");
  }
  return 0;
}

void Receiver::initialize() {
  int32_t tries = 100;
  uint16_t port = m_basePort;
  int32_t lockResult = m_mutex.tryacquire();
  int32_t result = -1;
  if (lockResult != -1) {  // The listener thread
    ACE_INET_Addr addr(port, "localhost");
    m_listener = ACE_Thread::self();
    result = m_io->open(addr);
  } else {
    while ((result < 0) && (--tries > 0)) {
      port += ++m_offset;
      ACE_INET_Addr addr(port, "localhost");
      result = m_io->open(addr);
      if (result == 0) {
        char hbuff[256];
        char* hst = &hbuff[0];
        char* fqdn = ACE_OS::getenv("GF_FQDN");
        if (fqdn) {
          hst = fqdn;
        } else {
          addr.get_host_name(hbuff, 255);
        }
        char buff[1024];
        sprintf(buff, "%s:%u", hst, port);
        m_addrs.push_back(buff);
      }
    }
  }
  if (result < 0) {
    FWKEXCEPTION("Server failed to open io, " << errno << ", on port " << port);
  }
}

int32_t STReceiver::doTask() {
  auto msg = std::unique_ptr<UDPMessage>(new UDPMessage());
  try {
    while (*m_run) {
      msg->clear();
      ACE_Time_Value timeout(2);  // Timeout is relative time
      if (msg->receiveFrom(m_io, &timeout)) {
        if (msg->getCmd() == ADDR_REQUEST) {
          msg->clear();
          msg->setMessage(m_addr);
          msg->setCmd(ADDR_RESPONSE);
          msg->send(m_io);
        } else {
          if (msg->length() > 0) {
            m_queues->putInbound(msg.release());
            msg.reset(new UDPMessage());
          }
        }
      }
    }
  } catch (FwkException& ex) {
    FWKSEVERE("STReceiver::doTask() caught exception: " << ex.what());
  } catch (...) {
    FWKSEVERE("STReceiver::doTask() caught unknown exception");
  }
  return 0;
}

void STReceiver::initialize() {
  int32_t result = -1;
  ACE_INET_Addr addr(m_basePort, "localhost");
  result = m_io.open(addr);
  if (result == 0) {
    char hbuff[256];
    char* hst = &hbuff[0];
    char* fqdn = ACE_OS::getenv("GF_FQDN");
    if (fqdn) {
      hst = fqdn;
    } else {
      addr.get_host_name(hbuff, 255);
    }
    char buff[1024];
    sprintf(buff, "%s:%u", hst, m_basePort);
    m_addr = buff;
  }
  if (result < 0) {
    FWKEXCEPTION("STReceiver::initialize failed to open io, "
                 << errno << ", on port " << m_basePort);
  }
}

int32_t Responder::doTask() {
  try {
    while (*m_run) {
      UDPMessage* msg = m_queues->getOutbound();
      if (msg) {
        msg->send(*m_io);
        delete msg;
      }
    }
  } catch (FwkException& ex) {
    FWKSEVERE("Responder::doTask() caught exception: " << ex.what());
  } catch (...) {
    FWKSEVERE("Responder::doTask() caught unknown exception");
  }
  return 0;
}

void Responder::initialize() {
  int32_t result = -1;
  int32_t tries = 100;
  while ((result < 0) && (--tries > 0)) {
    uint16_t port = ++m_offset + 111 + m_basePort;
    result = m_io->open(ACE_INET_Addr(port, "localhost"));
    if (result < 0) {
      FWKWARN("Server failed to open io, " << errno << ", on port " << port);
    }
  }
  if (result < 0) {
    FWKEXCEPTION("Server failed to open io, " << errno);
  }
}

}  // namespace testframework
}  // namespace client
}  // namespace geode
}  // namespace apache