<commit_before>/*
 * File description: axon_client.cpp
 * Author information: <NAME> <EMAIL>
 * Copyright information: Copyright <NAME>
 */

#include "communication/messaging/axon_client.h"
#include "communication/timeout_exception.h"

#include <functional>

using namespace std;

namespace axon { namespace communication {

struct CMessageSocket
{
	CMessageSocket(const CMessage &a_outboundMessage)
		: OutboundMessage(a_outboundMessage)
	{
	}

	const CMessage &OutboundMessage;
	CMessage::Ptr IncomingMessage;
};

CAxonClient::CAxonClient()
{
	SetDefaultProtocol();
}

CAxonClient::CAxonClient(const std::string& a_connectionString)
{
	SetDefaultProtocol();

	Connect(a_connectionString);
}

CAxonClient::CAxonClient(IDataConnection::Ptr a_connection)
{
	SetDefaultProtocol();

	Connect(move(a_connection));
}

CAxonClient::CAxonClient(const std::string& a_connectionString, IProtocol::Ptr a_protocol)
{
	SetProtocol(move(a_protocol));

	Connect(a_connectionString);
}

CAxonClient::CAxonClient(IDataConnection::Ptr a_connection, IProtocol::Ptr a_protocol)
{
	SetProtocol(move(a_protocol));

	Connect(move(a_connection));
}





CAxonClient::Ptr CAxonClient::Create()
{
	return make_shared<CAxonClient>();
}

CAxonClient::Ptr CAxonClient::Create(const std::string& a_connectionString)
{
	return make_shared<CAxonClient>(a_connectionString);
}

CAxonClient::Ptr CAxonClient::Create(IDataConnection::Ptr a_connection)
{
	return make_shared<CAxonClient>(move(a_connection));
}

CAxonClient::Ptr CAxonClient::Create(const std::string& a_connectionString,
		IProtocol::Ptr a_protocol)
{
	return make_shared<CAxonClient>(a_connectionString, move(a_protocol));
}

CAxonClient::Ptr CAxonClient::Create(IDataConnection::Ptr a_connection,
		IProtocol::Ptr a_protocol)
{
	return make_shared<CAxonClient>(move(a_connection), move(a_protocol));
}

void CAxonClient::Connect(const std::string& a_connectionString)
{
	Connect(IDataConnection::Create(a_connectionString));
}

void CAxonClient::Connect(IDataConnection::Ptr a_connection)
{
	m_connection = move(a_connection);

	if (m_connection)
	{
#ifdef IS_WINDOWS
		int l_hack;
		m_connection->SetReceiveHandler(
			[this, l_hack](CDataBuffer a_buf)
			{
				p_OnDataReceived(move(a_buf));
			});
#else
		m_connection->SetReceiveHandler(bind(&CAxonClient::p_OnDataReceived, this, placeholders::_1));
#endif
	}
}

void CAxonClient::SetDefaultProtocol()
{
	SetProtocol(GetDefaultProtocol());
}

void CAxonClient::SetProtocol(IProtocol::Ptr a_protocol)
{
	if (!a_protocol)
		throw runtime_error("Invalid protocol. Cannot be null.");

	m_protocol = move(a_protocol);

	m_protocol->SetHandler(bind(&CAxonClient::p_OnMessageReceived, this, placeholders::_1));
}

CMessage::Ptr CAxonClient::Send(const CMessage& a_message)
{

	return Send(a_message, 0);
}

CMessage::Ptr CAxonClient::Send(const CMessage &a_message, uint32_t a_timeout)
{
	if (!m_connection || !m_connection->IsOpen())
		throw runtime_error("Cannot send data over a dead connection.");
	// Default timeout is 1 minute
	if (a_timeout == 0)
		a_timeout = 60000;

	p_Send(a_message);

	auto l_start = chrono::steady_clock::now();

	auto l_durLeft = chrono::milliseconds(a_timeout);

	CMessageSocket l_socket(a_message);

	unique_lock<mutex> l_waitLock(m_pendingLock);

	// Add the socket to the map of messages waiting to be handled
	m_pendingList.push_back(&l_socket);

	size_t l_sockIdx = m_pendingList.size() - 1;

	while (!l_socket.IncomingMessage)
	{
		cv_status l_status = cv_status::no_timeout;

		if (a_timeout)
			l_status = m_newMessageEvent.wait_for(l_waitLock, l_durLeft);
		else
			m_newMessageEvent.wait(l_waitLock);

		l_durLeft -= chrono::duration_cast<chrono::milliseconds>(chrono::steady_clock::now() - l_start);

		if (l_socket.IncomingMessage
				|| l_status == cv_status::timeout
				|| l_durLeft < chrono::milliseconds(0))
		{
			m_pendingList.erase(m_pendingList.begin() + l_sockIdx);

			if (!l_socket.IncomingMessage)<commit_msg>Fixed a concurrency bug in AxonClient<commit_after>/*
 * File description: axon_client.cpp
 * Author information: <NAME> <EMAIL>
 * Copyright information: Copyright <NAME>
 */

#include "communication/messaging/axon_client.h"
#include "communication/timeout_exception.h"

#include <functional>

using namespace std;

namespace axon { namespace communication {

struct CMessageSocket
{
	CMessageSocket(const CMessage &a_outboundMessage)
		: OutboundMessage(a_outboundMessage)
	{
	}

	const CMessage &OutboundMessage;
	CMessage::Ptr IncomingMessage;
};

CAxonClient::CAxonClient()
{
	SetDefaultProtocol();
}

CAxonClient::CAxonClient(const std::string& a_connectionString)
{
	SetDefaultProtocol();

	Connect(a_connectionString);
}

CAxonClient::CAxonClient(IDataConnection::Ptr a_connection)
{
	SetDefaultProtocol();

	Connect(move(a_connection));
}

CAxonClient::CAxonClient(const std::string& a_connectionString, IProtocol::Ptr a_protocol)
{
	SetProtocol(move(a_protocol));

	Connect(a_connectionString);
}

CAxonClient::CAxonClient(IDataConnection::Ptr a_connection, IProtocol::Ptr a_protocol)
{
	SetProtocol(move(a_protocol));

	Connect(move(a_connection));
}





CAxonClient::Ptr CAxonClient::Create()
{
	return make_shared<CAxonClient>();
}

CAxonClient::Ptr CAxonClient::Create(const std::string& a_connectionString)
{
	return make_shared<CAxonClient>(a_connectionString);
}

CAxonClient::Ptr CAxonClient::Create(IDataConnection::Ptr a_connection)
{
	return make_shared<CAxonClient>(move(a_connection));
}

CAxonClient::Ptr CAxonClient::Create(const std::string& a_connectionString,
		IProtocol::Ptr a_protocol)
{
	return make_shared<CAxonClient>(a_connectionString, move(a_protocol));
}

CAxonClient::Ptr CAxonClient::Create(IDataConnection::Ptr a_connection,
		IProtocol::Ptr a_protocol)
{
	return make_shared<CAxonClient>(move(a_connection), move(a_protocol));
}

void CAxonClient::Connect(const std::string& a_connectionString)
{
	Connect(IDataConnection::Create(a_connectionString));
}

void CAxonClient::Connect(IDataConnection::Ptr a_connection)
{
	m_connection = move(a_connection);

	if (m_connection)
	{
#ifdef IS_WINDOWS
		int l_hack;
		m_connection->SetReceiveHandler(
			[this, l_hack](CDataBuffer a_buf)
			{
				p_OnDataReceived(move(a_buf));
			});
#else
		m_connection->SetReceiveHandler(bind(&CAxonClient::p_OnDataReceived, this, placeholders::_1));
#endif
	}
}

void CAxonClient::SetDefaultProtocol()
{
	SetProtocol(GetDefaultProtocol());
}

void CAxonClient::SetProtocol(IProtocol::Ptr a_protocol)
{
	if (!a_protocol)
		throw runtime_error("Invalid protocol. Cannot be null.");

	m_protocol = move(a_protocol);

	m_protocol->SetHandler(bind(&CAxonClient::p_OnMessageReceived, this, placeholders::_1));
}

CMessage::Ptr CAxonClient::Send(const CMessage& a_message)
{

	return Send(a_message, 0);
}

CMessage::Ptr CAxonClient::Send(const CMessage &a_message, uint32_t a_timeout)
{
	if (!m_connection || !m_connection->IsOpen())
		throw runtime_error("Cannot send data over a dead connection.");
	// Default timeout is 1 minute
	if (a_timeout == 0)
		a_timeout = 60000;

	p_Send(a_message);

	auto l_start = chrono::steady_clock::now();

	auto l_durLeft = chrono::milliseconds(a_timeout);

	CMessageSocket l_socket(a_message);

	unique_lock<mutex> l_waitLock(m_pendingLock);

	// Add the socket to the map of messages waiting to be handled
	m_pendingList.push_back(&l_socket);

	while (!l_socket.IncomingMessage)
	{
		cv_status l_status = cv_status::no_timeout;

		if (a_timeout)
			l_status = m_newMessageEvent.wait_for(l_waitLock, l_durLeft);
		else
			m_newMessageEvent.wait(l_waitLock);

		l_durLeft -= chrono::duration_cast<chrono::milliseconds>(chrono::steady_clock::now() - l_start);

		if (l_socket.IncomingMessage
				|| l_status == cv_status::timeout
				|| l_durLeft < chrono::milliseconds(0))
		{
			auto iter = find(m_pendingList.begin(), m_pendingList.end(), &l_socket);
            
            if (iter != m_pendingList.end())
                m_pendingList.erase(iter);
            else
                throw runtime_error("The completed socket wasn't in the list of pending communications."
                                    " Not even sure this can happen... but if it did... bummer.");

			if (!l_socket.IncomingMessage)
{'fraction_non_alphanumeric': 0.1227471992206527, 'fraction_numerical': 0.0032878714076960544, 'mean_word_length': 6.486782133090246, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7639139', 'n_tokens_mistral': 2951, 'n_tokens_neox': 2722, 'n_words': 480}