// DWBS ------------------------------------------------------------------ DWBS
//
// PLATFORM		: ALL
// PRODUCT		: BIT_DEMON
// VISIBILITY	: PUBLIC
//
// DWBS ------------------------------------------------------------------ DWBS

#include <bdNet/bdNet.h>
#include <bdNet/bdBandwidthMonitor/bdBandwidthMonitor.h>

#include <bdConnection/bdMessage/bdReceivedMessage.h>
#include <bdConnection/bdMessage/bdMessage.h>
#include <bdConnection/bdConnection.h>

#define BD_LOG_LEVEL	"bdNet/bandwidth"

BD_REGISTER_SINGLETON(bdBandwidthMonitorImpl)

void bdBandwidthMonitorImpl::setAvailableUpstreamBandwidth(const bdUInt bytesPerSecond)
{
	m_upstreamBwInBytesPerSecond = bytesPerSecond;
}

void bdBandwidthMonitorImpl::setAvailableDownstreamBandwidth(const bdUInt bytesPerSecond)
{
	m_downstreamBwInBytesPerSecond = bytesPerSecond;
}

void bdBandwidthMonitorImpl::update()
{
	if(m_sendHWMark > 0 && m_sendLWMark > 0)
	{
		const bdFloat32 elapsedTime = m_timer.getElapsedTimeInSeconds();
		if (elapsedTime > 0.0f)
		{	
			m_timer.start();
			const bdUInt64 totalBytesSent = bdPlatformSocket::getBytesSent();
			const bdUInt64 totalBytesReceived = bdPlatformSocket::getBytesReceived();

			const bdUInt64 bytesSent = (totalBytesSent - m_prevBytesSent);
			const bdUInt32 bytesReceived = static_cast<bdUInt32>(totalBytesReceived - m_prevBytesReceived);
			m_prevBytesSent = totalBytesSent;
			m_prevBytesReceived = totalBytesReceived;

			m_avgBytesSent = static_cast<bdUInt32>((m_avgBytesSent + bytesSent) / 2);
			m_bytesSentLastSecond = static_cast<bdUInt32>(m_avgBytesSent / elapsedTime);
			m_avgBytesReceived = (m_avgBytesReceived + bytesReceived) / 2;
			m_bytesReceivedLastSecond = static_cast<bdUInt32>(m_avgBytesReceived / elapsedTime);

			if (!m_listeners.isEmpty())
			{
				// check if we went past the water marks and call the registered
				// listener functions
				if (m_bytesSentLastSecond > m_sendHWMark && !m_toggleSendWM)
				{
					m_toggleSendWM = true;
					for(bdUInt i = 0; i < m_listeners.getSize(); i++)
					{
						m_listeners[i]->onSendHighWaterMark(m_bytesSentLastSecond);
					}
				}
				else if (m_bytesSentLastSecond < m_sendLWMark && m_toggleSendWM)
				{
					m_toggleSendWM = false;
					for(bdUInt i = 0; i < m_listeners.getSize(); i++)
					{
						m_listeners[i]->onSendLowWaterMark(m_bytesSentLastSecond);
					}
				}
			}
		}
	}
}

bdUInt32 bdBandwidthMonitorImpl::getBytesPerSecondSent() const
{
	return m_bytesSentLastSecond;
}

bdUInt32 bdBandwidthMonitorImpl::getBytesPerSecondReceived() const
{
	return m_bytesReceivedLastSecond;
}

void bdBandwidthMonitorImpl::setSendHighWaterMark(bdUInt32 bytesPerSecond)
{
	m_sendHWMark = bytesPerSecond;
}

void bdBandwidthMonitorImpl::setSendLowWaterMark(bdUInt32 bytesPerSecond)
{
	m_sendLWMark = bytesPerSecond;
}

bdUInt32 bdBandwidthMonitorImpl::getSendHighWaterMark() const
{
	return m_sendHWMark;
}

bdUInt32 bdBandwidthMonitorImpl::getSendLowWaterMark() const
{
	return m_sendLWMark;
}

void bdBandwidthMonitorImpl::registerListener(bdBandwidthListener *const listener)
{
#ifdef BD_DEBUG
	bdUInt32 index;
	if(m_listeners.findFirst(listener, index))
	{
		BD_WARN(BD_LOG_LEVEL,"Same listener registered multiple times.");
	}
#endif // BD_DEBUG

	m_listeners.pushBack(listener);
}

void bdBandwidthMonitorImpl::unregisterListener(bdBandwidthListener *const listener)
{
	m_listeners.removeAllKeepOrder(listener);
}

bdBool bdBandwidthMonitorImpl::accept(const bdReceivedMessage& receivedMsg)
{
	bdBool accepted = false;

	if(receivedMsg.getMessage()->getType() == BD_CONNECTION_SET_PARAM)
	{
		accepted = true;

		bdMessageRef msg = receivedMsg.getMessage();

		bdUInt32 transmissionRate = 0;
		if(msg->getPayload()->readUInt32(transmissionRate))
		{
			const bdUInt maxTransmissionRate = m_upstreamBwInBytesPerSecond / m_connections.getSize();

			receivedMsg.getConnection()->setTransmissionRate(BD_MIN(transmissionRate, maxTransmissionRate));
		}
		else
		{
			BD_WARN(BD_LOG_LEVEL, "Invalid message.");
		}
	}

	return accepted;
}

void bdBandwidthMonitorImpl::onConnect(bdConnectionRef connection)
{
	m_connections.addTail(connection);

	const bdUInt bw = (m_downstreamBwInBytesPerSecond / m_connections.getSize());

	bdMessageRef msg = new bdMessage(BD_CONNECTION_SET_PARAM);
	msg->getPayload()->writeUInt32(bw);

	bdLinkedList<bdConnectionRef>::Position pos = m_connections.getHeadPosition();
	while(pos)
	{
		bdConnectionRef conn = m_connections.forward(pos);
		conn->send(msg, true);
	}
}

void bdBandwidthMonitorImpl::onDisconnect(bdConnectionRef connection)
{

	bdLinkedList<bdConnectionRef>::Position pos = m_connections.find(connection);

	if(pos)
	{
		m_connections.removeAt(pos);
	}
	else
	{
		BD_WARN(BD_LOG_LEVEL, "onDisconnect for connection not in list!");
	}

	const bdUInt numConns = m_connections.getSize() > 0 ? m_connections.getSize() : 1;
	const bdUInt bw = (m_downstreamBwInBytesPerSecond / numConns);

	bdMessageRef msg = new bdMessage(BD_CONNECTION_SET_PARAM);
	msg->getPayload()->writeUInt32(bw);

	pos = m_connections.getHeadPosition();
	while(pos)
	{
		bdConnectionRef conn = m_connections.forward(pos);
		conn->send(msg, true);
	}
}

// protected:

bdBandwidthMonitorImpl::bdBandwidthMonitorImpl()
: m_upstreamBwInBytesPerSecond(BD_DEFAULT_UPSTREAM_BW_BPS),
  m_downstreamBwInBytesPerSecond(BD_DEFAULT_DOWNSTREAM_BW_BPS),
  m_prevBytesSent(0),
  m_avgBytesSent(0),
  m_bytesSentLastSecond(0),
  m_prevBytesReceived(0),
  m_avgBytesReceived(0),
  m_bytesReceivedLastSecond(0),
  m_sendHWMark(0),
  m_sendLWMark(0),
  m_toggleSendWM(0)
{
	m_timer.reset();
	m_prevBytesReceived = bdPlatformSocket::getBytesReceived();
	m_prevBytesSent = bdPlatformSocket::getBytesSent();
}
