TeaSpeakLibrary/src/protocol/AcknowledgeManager.cpp

#include "AcknowledgeManager.h"
#include <math.h>
#include <misc/endianness.h>

using namespace ts;
using namespace ts::connection;
using namespace ts::protocol;
using namespace std;
using namespace std::chrono;

AcknowledgeManager::AcknowledgeManager() {}

AcknowledgeManager::~AcknowledgeManager() {
	{
		lock_guard<recursive_mutex> lock(this->entry_lock);
		for(const auto& entry : this->entries)
			if(entry->acknowledge_listener)
				entry->acknowledge_listener->executionFailed("deleted");
		this->entries.clear();
	}
}

void AcknowledgeManager::reset() {
	{
		lock_guard<recursive_mutex> lock(this->entry_lock);
		for(const auto& entry : this->entries)
			if(entry->acknowledge_listener)
				entry->acknowledge_listener->executionFailed("reset");
		this->entries.clear();
	}
}

size_t AcknowledgeManager::awaiting_acknowledge() {
	lock_guard<recursive_mutex> lock(this->entry_lock);
	return this->entries.size();
}

void AcknowledgeManager::process_packet(ts::protocol::BasicPacket &packet) {
	if(!packet.type().requireAcknowledge()) return;

	auto entry = make_shared<Entry>();
	entry->acknowledge_listener = std::move(packet.getListener());

	entry->buffer = packet.buffer();

	entry->resend_count = 0;
	entry->resend_period = milliseconds((int) ceil(this->average_response * 3/2));
	entry->first_send = system_clock::now();
	entry->next_resend = entry->first_send + entry->resend_period;

	entry->packet_type = packet.type().type();
	entry->packet_id = packet.packetId();

	entry->acknowledged = false;
	entry->send_count = 1;
	{
		lock_guard<recursive_mutex> lock(this->entry_lock);
		this->entries.push_front(std::move(entry));
	}
}

bool AcknowledgeManager::process_acknowledge(const ts::protocol::BasicPacket &packet, std::string& error) {
	PacketType target_type = PacketType::UNDEFINED;
	uint16_t target_id = 0;

	if(packet.type().type() == PacketType::ACK_LOW) target_type = PacketType::COMMAND_LOW;
	else if(packet.type().type() == PacketType::ACK) target_type = PacketType::COMMAND;
	target_id = be2le16((char*) packet.data().data_ptr());
	//debugMessage(0, "Got ack for {} {}", target_type, target_id);

	if(target_type == PacketType::UNDEFINED) {
		error = "Invalid packet type (" + to_string(target_type) + ")";
		return false;
	}

	std::shared_ptr<Entry> entry;
	{
		lock_guard<recursive_mutex> lock(this->entry_lock);
		for(auto it = this->entries.begin(); it != this->entries.end(); it++) {
			if((*it)->packet_type == target_type && (*it)->packet_id == target_id) {
				entry = *it;
				entry->send_count--;
				if(entry->send_count == 0)
					this->entries.erase(it);
				break;
			}
		}
	}
	if(!entry) {
		error = "Missing packet id (" + to_string(target_id) + ")";
		return false;
	}

	auto time = system_clock::now() - entry->next_resend + entry->resend_period;
	auto ms_time = duration_cast<milliseconds>(time).count();
	if(ms_time > 5) {
		this->average_response = this->average_response / 2 + ms_time / 2;
	}

	entry->acknowledged = true;
	if(entry->acknowledge_listener)
		entry->acknowledge_listener->executionSucceed(true);
	entry->acknowledge_listener.reset();

	return true;
}

ssize_t AcknowledgeManager::execute_resend(const system_clock::time_point& now , std::chrono::system_clock::time_point &next_resend,std::deque<pipes::buffer>& buffers, string& error) {
	ssize_t resend_count = 0;

	deque<shared_ptr<Entry>> need_resend;
	{
		deque<shared_ptr<Entry>> erase;

		lock_guard<recursive_mutex> lock(this->entry_lock);
		for (auto &entry : this->entries) {
			if(!entry->acknowledged && entry->next_resend <= now) {
				entry->resend_period = entry->resend_period + milliseconds((int) ceil(this->average_response * 2));
				if(entry->resend_period.count() > 1000)
					entry->resend_period = milliseconds(1000);
				else if(entry->resend_period.count() < 25)
					entry->resend_period = milliseconds(25);

				entry->next_resend = now + entry->resend_period;
				need_resend.push_front(entry);
			}
			if(entry->acknowledged) {
				if(entry->next_resend + entry->resend_period <= now) { //Timeout for may (more acknowledges)
					erase.push_back(entry);
				}
			} else {
				if(next_resend > entry->next_resend)
					next_resend = entry->next_resend;
			}
		}

		for(const auto& e : erase) {
			auto it = find(this->entries.begin(), this->entries.end(), e);
			if(it != this->entries.end())
				this->entries.erase(it);
		}
	}

	for(const auto& packet : need_resend) {
		if(packet->resend_count > 15 && packet->first_send + seconds(15) < now) { //FIXME configurable
			error = "Failed to receive acknowledge for packet " + to_string(packet->packet_id) + " of type " + PacketTypeInfo::fromid(packet->packet_type).name();
			return -1;
		}

		resend_count++;
		packet->resend_count++;
		packet->send_count++;
		buffers.push_back(packet->buffer);
	}

	return resend_count;
}