// Filename: connection.cxx
// Created by:  jns (07Feb00)
//
////////////////////////////////////////////////////////////////////
//
// PANDA 3D SOFTWARE
// Copyright (c) Carnegie Mellon University.  All rights reserved.
//
// All use of this software is subject to the terms of the revised BSD
// license.  You should have received a copy of this license along
// with this source code in a file named "LICENSE."
//
////////////////////////////////////////////////////////////////////

#include "connection.h"
#include "connectionManager.h"
#include "netDatagram.h"
#include "datagramTCPHeader.h"
#include "datagramUDPHeader.h"
#include "config_net.h"
#include "config_express.h" // for collect_tcp
#include "trueClock.h"
#include "pnotify.h"
#include "lightReMutexHolder.h"
#include "socket_ip.h"
#include "socket_tcp.h"
#include "socket_udp.h"
#include "dcast.h"


////////////////////////////////////////////////////////////////////
//     Function: Connection::Constructor
//       Access: Published
//  Description: Creates a connection.  Normally this constructor
//               should not be used directly by user code; use one of
//               the methods in ConnectionManager to make a new
//               connection.
////////////////////////////////////////////////////////////////////
Connection::
Connection(ConnectionManager *manager, Socket_IP *socket) :
  _manager(manager),
  _socket(socket)
{
  _collect_tcp = collect_tcp;
  _collect_tcp_interval = collect_tcp_interval;
  _queued_data_start = 0.0;
  _queued_count = 0;

#if defined(HAVE_THREADS) && defined(SIMPLE_THREADS)
  // In the presence of SIMPLE_THREADS, we use non-blocking I/O.  We
  // simulate blocking by yielding the thread.
  if (_socket->SetNonBlocking() != ALL_OK) {
    net_cat.warning()
      << "Unable to set non-blocking status on socket\n";
  }
#endif
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::Destructor
//       Access: Published
//  Description: Closes a connection.
////////////////////////////////////////////////////////////////////
Connection::
~Connection() {
  net_cat.info()
    << "Deleting connection " << (void *)this << "\n";

  if (_socket != (Socket_IP *)NULL) {
    flush();

    _socket->Close();
    delete _socket;
  }
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::get_address
//       Access: Published
//  Description: Returns the address bound to this connection, if it
//               is a TCP connection.
////////////////////////////////////////////////////////////////////
NetAddress Connection::
get_address() const {
  Socket_Address addr = _socket->GetPeerName();
  return NetAddress(addr);
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::get_manager
//       Access: Published
//  Description: Returns a pointer to the ConnectionManager object
//               that serves this connection.
////////////////////////////////////////////////////////////////////
ConnectionManager *Connection::
get_manager() const {
  return _manager;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::get_socket
//       Access: Published
//  Description: Returns the internal Socket_IP that defines the
//               connection.
////////////////////////////////////////////////////////////////////
Socket_IP *Connection::
get_socket() const {
  return _socket;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_collect_tcp
//       Access: Published
//  Description: Enables or disables "collect-tcp" mode.  In this
//               mode, individual TCP packets are not sent
//               immediately, but rather they are collected together
//               and accumulated to be sent periodically as one larger
//               TCP packet.  This cuts down on overhead from the
//               TCP/IP protocol, especially if many small packets
//               need to be sent on the same connection, but it
//               introduces additional latency (since packets must be
//               held before they can be sent).
//
//               See set_collect_tcp_interval() to specify the
//               interval of time for which to hold packets before
//               sending them.
//
//               If you enable this mode, you may also need to
//               periodically call consider_flush() to flush the queue
//               if no packets have been sent recently.
////////////////////////////////////////////////////////////////////
void Connection::
set_collect_tcp(bool collect_tcp) {
  _collect_tcp = collect_tcp;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::get_collect_tcp
//       Access: Published
//  Description: Returns the current setting of "collect-tcp" mode.
//               See set_collect_tcp().
////////////////////////////////////////////////////////////////////
bool Connection::
get_collect_tcp() const {
  return _collect_tcp;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_collect_tcp_interval
//       Access: Published
//  Description: Specifies the interval in time, in seconds, for which
//               to hold TCP packets before sending all of the
//               recently received packets at once.  This only has
//               meaning if "collect-tcp" mode is enabled; see
//               set_collect_tcp().
////////////////////////////////////////////////////////////////////
void Connection::
set_collect_tcp_interval(double interval) {
  _collect_tcp_interval = interval;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::get_collect_tcp_interval
//       Access: Published
//  Description: Returns the interval in time, in seconds, for which
//               to hold TCP packets before sending all of the
//               recently received packets at once.  This only has
//               meaning if "collect-tcp" mode is enabled; see
//               set_collect_tcp().
////////////////////////////////////////////////////////////////////
double Connection::
get_collect_tcp_interval() const {
  return _collect_tcp_interval;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::consider_flush
//       Access: Published
//  Description: Sends the most recently queued TCP datagram(s) if
//               enough time has elapsed.  This only has meaning if
//               set_collect_tcp() has been set to true.
////////////////////////////////////////////////////////////////////
bool Connection::
consider_flush() {
  LightReMutexHolder holder(_write_mutex);

  if (!_collect_tcp) {
    return do_flush();

  } else {
    double elapsed = 
      TrueClock::get_global_ptr()->get_short_time() - _queued_data_start;
    // If the elapsed time is negative, someone must have reset the
    // clock back, so just go ahead and flush.
    if (elapsed < 0.0 || elapsed >= _collect_tcp_interval) {
      return do_flush();
    }
  }

  return true;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::flush
//       Access: Published
//  Description: Sends the most recently queued TCP datagram(s) now.
//               This only has meaning if set_collect_tcp() has been
//               set to true.
////////////////////////////////////////////////////////////////////
bool Connection::
flush() {
  LightReMutexHolder holder(_write_mutex);
  return do_flush();
}

/*
This method is disabled.  We don't provide enough interface to use
non-blocking I/O effectively at this level, so we shouldn't provide
this call.  Specifically, we don't provide a way to query whether an
operation failed because it would have blocked or not.

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_nonblock
//       Access: Published
//  Description: Sets whether nonblocking I/O should be in effect.
////////////////////////////////////////////////////////////////////
void Connection::
set_nonblock(bool flag) {
  if (flag) {
    _socket->SetNonBlocking();
  } else {
    _socket->SetBlocking();
  }
}
*/

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_linger
//       Access: Published
//  Description: Sets the time to linger on close if data is present.
//               If flag is false, when you close a socket with data
//               available the system attempts to deliver the data to
//               the peer (the default behavior).  If flag is false
//               but time is zero, the system discards any undelivered
//               data when you close the socket.  If flag is false but
//               time is nonzero, the system waits up to time seconds
//               to deliver the data.
////////////////////////////////////////////////////////////////////
void Connection::
set_linger(bool flag, double time) {
  Socket_TCP *tcp;
  DCAST_INTO_V(tcp, _socket);

  if (flag) {
    tcp->SetLinger((int)time);
  } else {
    tcp->DontLinger();
  }
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_reuse_addr
//       Access: Published
//  Description: Sets whether local address reuse is allowed.
////////////////////////////////////////////////////////////////////
void Connection::
set_reuse_addr(bool flag) {
  _socket->SetReuseAddress(flag);
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_keep_alive
//       Access: Published
//  Description: Sets whether the connection is periodically tested to
//               see if it is still alive.
////////////////////////////////////////////////////////////////////
void Connection::
set_keep_alive(bool flag) {
  // TODO.
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_recv_buffer_size
//       Access: Published
//  Description: Sets the size of the receive buffer, in bytes.
////////////////////////////////////////////////////////////////////
void Connection::
set_recv_buffer_size(int size) {
  _socket->SetRecvBufferSize(size);
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_send_buffer_size
//       Access: Published
//  Description: Sets the size of the send buffer, in bytes.
////////////////////////////////////////////////////////////////////
void Connection::
set_send_buffer_size(int size) {
  Socket_TCP *tcp;
  DCAST_INTO_V(tcp, _socket);

  tcp->SetSendBufferSize(size);
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_ip_time_to_live
//       Access: Published
//  Description: Sets IP time-to-live.
////////////////////////////////////////////////////////////////////
void Connection::
set_ip_time_to_live(int ttl) {
  // TODO.
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_ip_type_of_service
//       Access: Published
//  Description: Sets IP type-of-service and precedence.
////////////////////////////////////////////////////////////////////
void Connection::
set_ip_type_of_service(int tos) {
  // TODO.
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_no_delay
//       Access: Published
//  Description: If flag is true, this disables the Nagle algorithm,
//               and prevents delaying of send to coalesce packets.
////////////////////////////////////////////////////////////////////
void Connection::
set_no_delay(bool flag) {
  Socket_TCP *tcp;
  DCAST_INTO_V(tcp, _socket);

  tcp->SetNoDelay(flag);
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::set_max_segment
//       Access: Published
//  Description: Sets the maximum segment size.
////////////////////////////////////////////////////////////////////
void Connection::
set_max_segment(int size) {
  // TODO.
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::send_datagram
//       Access: Private
//  Description: This method is intended only to be called by
//               ConnectionWriter.  It atomically writes the given
//               datagram to the socket, returning true on success,
//               false on failure.  If the socket seems to be closed,
//               it notifies the ConnectionManager.
////////////////////////////////////////////////////////////////////
bool Connection::
send_datagram(const NetDatagram &datagram, int tcp_header_size) {
  nassertr(_socket != (Socket_IP *)NULL, false);

  if (_socket->is_exact_type(Socket_UDP::get_class_type())) {
    // We have to send UDP right away.
    Socket_UDP *udp;
    DCAST_INTO_R(udp, _socket, false);

    LightReMutexHolder holder(_write_mutex);
    DatagramUDPHeader header(datagram);
    string data;
    data += header.get_header();
    data += datagram.get_message();
    
    if (net_cat.is_debug()) {
      header.verify_datagram(datagram);
    }
    
    int bytes_to_send = data.length();
    Socket_Address addr = datagram.get_address().get_addr();

    bool okflag = udp->SendTo(data, addr);
#if defined(HAVE_THREADS) && defined(SIMPLE_THREADS)
    while (!okflag && udp->GetLastError() == LOCAL_BLOCKING_ERROR && udp->Active()) {
      Thread::force_yield();
      okflag = udp->SendTo(data, addr);
    }
#endif  // SIMPLE_THREADS
      
    if (net_cat.is_spam()) {
      net_cat.spam()
        << "Sent UDP datagram with " 
        << bytes_to_send << " bytes to " << (void *)this 
        << ", ok = " << okflag << "\n";
    }
      
    return check_send_error(okflag);
  }

  // We might queue up TCP packets for later sending.
  if (tcp_header_size == 2 && datagram.get_length() >= 0x10000) {
    net_cat.error()
      << "Attempt to send TCP datagram of " << datagram.get_length()
      << " bytes--too long!\n";
    nassert_raise("Datagram too long");
    return false;
  }

  DatagramTCPHeader header(datagram, tcp_header_size);

  LightReMutexHolder holder(_write_mutex);
  _queued_data += header.get_header();
  _queued_data += datagram.get_message();
  _queued_count++;
  
  if (net_cat.is_debug()) {
    header.verify_datagram(datagram, tcp_header_size);
  }

  if (!_collect_tcp || 
      TrueClock::get_global_ptr()->get_short_time() - _queued_data_start >= _collect_tcp_interval) {
    return do_flush();
  }

  return true;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::send_raw_datagram
//       Access: Private
//  Description: This method is intended only to be called by
//               ConnectionWriter.  It atomically writes the given
//               datagram to the socket, without the Datagram header.
////////////////////////////////////////////////////////////////////
bool Connection::
send_raw_datagram(const NetDatagram &datagram) {
  nassertr(_socket != (Socket_IP *)NULL, false);

  if (_socket->is_exact_type(Socket_UDP::get_class_type())) {
    // We have to send UDP right away.
    Socket_UDP *udp;
    DCAST_INTO_R(udp, _socket, false);

    string data = datagram.get_message();

    LightReMutexHolder holder(_write_mutex);
    Socket_Address addr = datagram.get_address().get_addr();
    bool okflag = udp->SendTo(data, addr);
#if defined(HAVE_THREADS) && defined(SIMPLE_THREADS)
    while (!okflag && udp->GetLastError() == LOCAL_BLOCKING_ERROR && udp->Active()) {
      Thread::force_yield();
      okflag = udp->SendTo(data, addr);
    }
#endif  // SIMPLE_THREADS
    
    if (net_cat.is_spam()) {
      net_cat.spam()
        << "Sent UDP datagram with " 
        << data.size() << " bytes to " << (void *)this 
        << ", ok = " << okflag << "\n";
    }

    return check_send_error(okflag);
  }

  // We might queue up TCP packets for later sending.
  LightReMutexHolder holder(_write_mutex);
  _queued_data += datagram.get_message();
  _queued_count++;

  if (!_collect_tcp || 
      TrueClock::get_global_ptr()->get_short_time() - _queued_data_start >= _collect_tcp_interval) {
    return do_flush();
  }

  return true;
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::do_flush
//       Access: Private
//  Description: The private implementation of flush(), this assumes
//               the _write_mutex is already held.
////////////////////////////////////////////////////////////////////
bool Connection::
do_flush() {
  if (_queued_data.empty()) {
    _queued_count = 0;
    _queued_data_start = TrueClock::get_global_ptr()->get_short_time();
    return true;
  }

  if (net_cat.is_spam()) {
    net_cat.spam()
      << "Sending " << _queued_count << " TCP datagram(s) with " 
      << _queued_data.length() << " total bytes to " << (void *)this << "\n";
  }

  Socket_TCP *tcp;
  DCAST_INTO_R(tcp, _socket, false);

  string sending_data;
  _queued_data.swap(sending_data);

  _queued_count = 0;
  _queued_data_start = TrueClock::get_global_ptr()->get_short_time();

#if defined(HAVE_THREADS) && defined(SIMPLE_THREADS)
  int max_send = net_max_write_per_epoch;
  int data_sent = tcp->SendData(sending_data.data(), min((size_t)max_send, sending_data.size()));
  bool okflag = (data_sent == (int)sending_data.size());
  if (!okflag) {
    int total_sent = 0;
    if (data_sent > 0) {
      total_sent += data_sent;
    }
    double last_report = 0;
    while (!okflag && tcp->Active() &&
           (data_sent > 0 || tcp->GetLastError() == LOCAL_BLOCKING_ERROR)) {
      if (data_sent == 0) {
        Thread::force_yield();
      } else {
        Thread::consider_yield();
      }
      data_sent = tcp->SendData(sending_data.data() + total_sent, min((size_t)max_send, sending_data.size() - total_sent));
      if (data_sent > 0) {
        total_sent += data_sent;
      }
      okflag = (total_sent == (int)sending_data.size());
    }
  }

#else  // SIMPLE_THREADS
  int data_sent = tcp->SendData(sending_data);
  bool okflag = (data_sent == (int)sending_data.size());

#endif  // SIMPLE_THREADS

  return check_send_error(okflag);
}

////////////////////////////////////////////////////////////////////
//     Function: Connection::check_send_error
//       Access: Private
//  Description: Checks the return value of a Send() or SendTo()
//               call.
////////////////////////////////////////////////////////////////////
bool Connection::
check_send_error(bool okflag) {
  if (!okflag) {
    static ConfigVariableBool abort_send_error("abort-send-error", false);
    if (abort_send_error) {
      nassertr(false, false);
    }

    // Assume any error means the connection has been reset; tell
    // our manager about it and ignore it.
    if (_manager != (ConnectionManager *)NULL) {
      _manager->flush_read_connection(this);
      _manager->connection_reset(this, okflag);
    }
    return false;
  }

  return true;
}
