Network/Network/UdpClientImpl.cpp

#include "UdpClientImpl.h"
#include <Logging.h>


namespace Network {

UdpClientImpl::UdpClientImpl(const std::string& listenAddress, const std::string& targetAddress, int targetPort, std::function<void(const std::string&)> callback) :
	m_bufferLength(1024),
	m_deadline(m_ioServiceAnswer),
	m_listenAddress(asio::ip::address::from_string(listenAddress)),
	m_targetAddress(asio::ip::address::from_string(targetAddress)),
	m_targetPort(targetPort),
	m_receiveSocket(m_ioService),
	m_answerSocket(m_ioServiceAnswer),
	m_targetEndpoint(m_targetAddress, m_targetPort),
	m_callback(callback)
{
	m_thread = std::thread([&] { m_ioService.run(); });

#ifdef __linux__
	pthread_setname_np(m_thread.native_handle(), "UdpClientImpl");
#endif

	m_socketReceiveData.reserve(m_bufferLength);
	m_socketAnswerData.reserve(m_bufferLength);

	asio::ip::udp::endpoint listenEndpoint(m_targetAddress, m_targetPort);

	m_receiveSocket.open(listenEndpoint.protocol());
	m_receiveSocket.set_option(asio::ip::udp::socket::reuse_address(true));
	m_receiveSocket.bind(listenEndpoint);
}

UdpClientImpl::~UdpClientImpl()
{
	m_ioService.stop();
	m_thread.join();
}

void UdpClientImpl::Receive()
{
	ReceiveSocket();

	m_deadline.expires_at(asio::system_timer::time_point(std::chrono::system_clock::duration::max()));
	CheckDeadline();
}

std::string UdpClientImpl::Send(const std::string& data, int sourcePort)
{
	asio::ip::udp::socket sendSocket(m_ioService);
	sendSocket.open(asio::ip::udp::v4());

	sendSocket.set_option(asio::ip::udp::socket::reuse_address(true));

	if (sourcePort != 0)
	{
		asio::ip::udp::endpoint sendEndpoint(m_listenAddress, sourcePort);
		sendSocket.bind(sendEndpoint);
	}

	try
	{
		sendSocket.connect(m_targetEndpoint);

		asio::ip::udp::endpoint listenEndpoint(m_listenAddress, sendSocket.local_endpoint().port());
		m_answerSocket.open(listenEndpoint.protocol());
		m_answerSocket.set_option(asio::ip::udp::socket::reuse_address(true));
		m_answerSocket.bind(listenEndpoint);

		sendSocket.send(asio::buffer(data));

		asio::error_code error = asio::error::would_block;
		std::size_t length = 0;

		m_deadline.expires_from_now(std::chrono::seconds(1));
		m_answerSocket.async_receive(asio::buffer(m_socketAnswerData.data(), m_bufferLength), std::bind(&UdpClientImpl::HandleAnswer, this, std::placeholders::_1, std::placeholders::_2, &error, &length));

		do m_ioServiceAnswer.run_one();
		while (error == asio::error::would_block);

		std::string answer;
		if (error != asio::error::operation_aborted)
			answer = std::string(m_socketAnswerData.begin(), m_socketAnswerData.begin() + length);
		else
			answer = "Error";

		m_answerSocket.close();
		return answer;
	}
	catch (const asio::system_error& ex)
	{
		std::stringstream ss;
		ss << "UdpClientImpl::Send() - Error: " << ex.what() << std::endl;
		Logging::Log(Logging::Severity::Error, ss.str());
		return "Error";
	}
}

void UdpClientImpl::Broadcast(const std::string& data, int sourcePort)
{
	asio::ip::udp::socket sendSocket(m_ioService);
	sendSocket.open(asio::ip::udp::v4());

	sendSocket.set_option(asio::ip::udp::socket::reuse_address(true));
	sendSocket.set_option(asio::socket_base::broadcast(true));

	if (sourcePort != 0)
	{
		asio::ip::udp::endpoint sendEndpoint(m_listenAddress, sourcePort);
		sendSocket.bind(sendEndpoint);
	}

	asio::ip::udp::endpoint targetEndpoint(asio::ip::address_v4::broadcast(), m_targetPort);

	try
	{
		sendSocket.connect(targetEndpoint);

		sendSocket.send(asio::buffer(data));
	}
	catch (const asio::system_error& ex)
	{
		std::stringstream ss;
		ss << "UdpClientImpl::Broadcast() - Error: " << ex.what() << std::endl;
		Logging::Log(Logging::Severity::Error, ss.str());
	}
	sendSocket.close();
}

void UdpClientImpl::EnableMulticast()
{
	m_receiveSocket.set_option(asio::ip::multicast::join_group(m_targetAddress.to_v4(), m_listenAddress.to_v4()));
}

void UdpClientImpl::CheckDeadline()
{
	if (m_deadline.expires_at() <= std::chrono::system_clock::now())
	{
		m_answerSocket.close();
		m_deadline.expires_at(asio::system_timer::time_point(std::chrono::system_clock::duration::max()));
	}

	m_deadline.async_wait(std::bind(&UdpClientImpl::CheckDeadline, this));
}

void UdpClientImpl::HandleAnswer(const asio::error_code& error, std::size_t length, asio::error_code* out_error, std::size_t* out_length) const
{
	*out_error = error;
	*out_length = length;
}

void UdpClientImpl::ReceiveSocket()
{
	m_receiveSocket.async_receive_from(asio::buffer(m_socketReceiveData.data(), m_bufferLength), m_sourceEndpoint, std::bind(&UdpClientImpl::HandleReceiveSocket, this, std::placeholders::_1, std::placeholders::_2));
}

void UdpClientImpl::HandleReceiveSocket(const asio::error_code& error, size_t bytes)
{
	if (error)
		return;

	if (m_callback)
	{
		std::vector<char>::iterator begin = m_socketReceiveData.begin();
		m_callback(std::string(begin, begin + bytes));
	}

	ReceiveSocket();
}

} // namespace Network