#include <misc/endianness.h>
#include <algorithm>
#include <log/LogUtils.h>
#include "../../server/VoiceServer.h"
#include <misc/memtracker.h>
#include <protocol/Packet.h>
#include <ThreadPool/Timer.h>
#include "VoiceClientConnection.h"
#include "src/client/ConnectedClient.h"
#include "VoiceClient.h"


//#define LOG_AUTO_ACK_AUTORESPONSE
//#define FUZZING_TESTING_INCOMMING
//#define FUZZING_TESTING_OUTGOING
#define FIZZING_TESTING_DISABLE_HANDSHAKE
#define FUZZING_TESTING_DROP 5
#define FUZZING_TESTING_DROP_MAX 10

//#define CONNECTION_NO_STATISTICS

#define QLZ_COMPRESSION_LEVEL 1
#include "qlz/QuickLZ.h"

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

VoiceClientConnection::VoiceClientConnection(VoiceClient* client) : client(client) {
	memtrack::allocated<VoiceClientConnection>(this);

	this->crypt_handler.reset();
	debugMessage(client->getServer()->getServerId(), "Allocated new voice client connection at {}", (void*) this);
}

VoiceClientConnection::~VoiceClientConnection() {
	/* locking here should be useless, but just to ensure! */
	{
		lock_guard write_queue_lock(this->write_queue_lock);
		this->write_queue.clear();
	}
	{
		lock_guard write_queue_lock(this->write_prepare_queue_lock);
		this->write_prepare_queue.clear();
	}
	this->client = nullptr;
	memtrack::freed<VoiceClientConnection>(this);
}

void VoiceClientConnection::triggerWrite() {
	if(this->client->voice_server)
		this->client->voice_server->triggerWrite(dynamic_pointer_cast<VoiceClient>(this->client->_this.lock()));
}

#ifdef CLIENT_LOG_PREFIX
#undef CLIENT_LOG_PREFIX
#endif
#define CLIENT_LOG_PREFIX "[" << this->client->getPeerIp() << ":" << this->client->getPeerPort() << " | " << this->client->getDisplayName() << "]"

//Message handle methods

void VoiceClientConnection::handleDatagramReceived(const pipes::buffer_view& buffer) {
#ifdef FUZZING_TESTING_INCOMMING
	#ifdef FIZZING_TESTING_DISABLE_HANDSHAKE
	if (this->client->state == ConnectionState::CONNECTED) {
	#endif
		if ((rand() % FUZZING_TESTING_DROP_MAX) < FUZZING_TESTING_DROP) {
			debugMessage(this->client->getServerId(), "{}[FUZZING] Dropping incoming packet of length {}", CLIENT_STR_LOG_PREFIX_(this->client), buffer.length());
			return;
		}
	#ifdef FIZZING_TESTING_DISABLE_HANDSHAKE
	}
	#endif
#endif

	auto packet = std::make_unique<ClientPacket>(buffer);
	//packet

	//packet->type().type()
	auto packet_type = packet->type();
	auto packet_id = packet->packetId();
	auto ordered = packet_type.type() == protocol::COMMAND || packet_type.type() == protocol::COMMAND_LOW;

	if(packet_type.type() < 0 || packet_type.type() >= this->_packet_buffers.size()) {
		logError(this->client->getServerId(), "{} Received invalid packet. Invalid packet type {}. Dropping packet.", CLIENT_STR_LOG_PREFIX_(this->client), packet_type.type());
		return;
	}

	auto& read_queue = this->_packet_buffers[packet_type.type()];
	packet->generationId(read_queue.generation(packet_id));

	if(ordered) {
		unique_lock queue_lock(read_queue.buffer_lock);
		auto result = read_queue.accept_index(packet_id);
		if(result != 0) { /* packet index is ahead buffer index */
			debugMessage(this->client->getServerId(), "{} Got packet of type {} which is out of the buffer range of {} ({}). Packet ID: {}, Current index: {}. Dropping packet",
					CLIENT_STR_LOG_PREFIX_(this->client),
				    packet_type.name(),
					read_queue.capacity(),
					result == -1 ? "underflow" : "overflow",
					packet_id,
					read_queue.current_index()
			);

			if(result == -1) { /* underflow */
				/* we've already got the packet, but the client dosnt know that so we've to send the acknowledge again */
				if(this->client->crypto.protocol_encrypted && (packet->type() == PacketTypeInfo::Command || packet->type() == PacketTypeInfo::CommandLow)){ //needs an acknowledge
					this->client->sendAcknowledge(packet->packetId(), packet->type() == PacketTypeInfo::CommandLow);
				}
			}
			return;
		}
	}

	packet->setEncrypted(!packet->has_flag(PacketFlag::Unencrypted)); // && packet->type() != PacketType::Init1
	if(packet->type() == PacketTypeInfo::Command || packet->type() == PacketTypeInfo::CommandLow){
		packet->setCompressed(packet->has_flag(PacketFlag::Compressed));
	}
	//NOTICE I found out that the Compressed flag is set if the packet contains an audio header

	string error = "success";
	if(this->client->state == ConnectionState::INIT_LOW && packet->type() != PacketTypeInfo::Init1){
		//Sends command packet as legacy support (skip step 1-3 | Direct clientinit with default key)
		return;
	}

	if(!this->crypt_handler.progressPacketIn(packet.get(), error, false)){
		//FIXME Only try to decrypt by default when its no flood attack!
		if(!this->client->crypto.client_init && !this->crypt_handler.use_default()) {
			if(!this->crypt_handler.progressPacketIn(packet.get(), error, true)){
				debugMessage(
						this->client->getServerId(),
						"{} Cant decrypt packet even with default key! Type: {}, Error: {}, Packet ID: {}, Generation: {}",
						CLIENT_STR_LOG_PREFIX_(this->client),
						packet->type().name(),
						error,
						packet_id,
						packet->generationId()
				);
				return;
			} else {
				debugMessage(
						this->client->getServerId(),
						"{} Cant decrypt packet with configured key {}. Error: {}. Succeeded with default key!",
						CLIENT_STR_LOG_PREFIX_(this->client),
						packet->type().name(),
						error
				);
			}
		} else {
			bool succeeded = false;
			if(packet_type == PacketTypeInfo::Voice) {
				/* FIXME: This try and error should not happen! */
				packet->generationId(packet->generationId() + 1);
				succeeded = this->crypt_handler.progressPacketIn(packet.get(), error, false);
			}
			if(succeeded) {
				auto old_packet_id = read_queue.current_index();
				read_queue.set_generation_packet(packet->generationId(), packet->packetId());

				logWarning(this->client->getServerId(), "{} Voice packet generation counter missed generation increasement. From {} to {} from packet id {} to {}",
				           CLIENT_STR_LOG_PREFIX_(this->client),
				           packet->generationId() - 1,
				           packet->generationId(),
				           old_packet_id,
				           packet->packetId()
				);
			} else {
				debugMessage(
						this->client->getServerId(),
						"{} Cant decrypt packet of type {}. Packet ID: {}, Estimated generation: {}, Full counter: {}. Dropping packet. Error: {}",
						CLIENT_STR_LOG_PREFIX_(this->client),
						packet->type().name(),
						packet->packetId(),
						packet->generationId(),
						read_queue.full_index(),
						error
				);
				return;
			}
		}
	}

	if(packet->type() == PacketTypeInfo::Command || packet->type() == PacketTypeInfo::CommandLow){
		if(packet->has_flag(PacketFlag::Unencrypted) && this->client->state != ConnectionState::INIT_HIGH){
			this->client->disconnect("Invalid packet. Command should not be unencrypted!");
			logger::logger(this->client->getServer()->getServerId())->error("{} Voice manager {}/{} tried to send a unencrypted command packet.", CLIENT_STR_LOG_PREFIX_(this->client), client->getDisplayName(), this->client->getLoggingPeerIp());
			return;
		}
	}

#ifndef CONNECTION_NO_STATISTICS
	if(this->client && this->client->getServer())
		this->client->connectionStatistics->logIncomingPacket(*packet);
#endif

#ifdef LOG_INCOMPING_PACKET_FRAGMENTS
	debugMessage(lstream << CLIENT_LOG_PREFIX << "Recived packet. PacketId: " << packet->packetId() << " PacketType: " << packet->type().name() << " Flags: " << packet->flags() << " - " << packet->data() << endl);
#endif
	if(packet->type() == PacketTypeInfo::Command || packet->type() == PacketTypeInfo::CommandLow){ //needs an acknowledge
		if(this->client->crypto.protocol_encrypted) {
			#ifdef LOG_AUTO_ACK_AUTORESPONSE
			logMessage(this->client->getServerId(), "{}[Ack] Sending ack for incoming command packet {}", CLIENT_STR_LOG_PREFIX_(this->client), packet->packetId());
			#endif
			this->client->sendAcknowledge(packet->packetId(), packet->type() == PacketTypeInfo::CommandLow);
		} else {
			debugMessage(this->client->getServerId(), "{}[Ack] Ignoring ack for {}", CLIENT_STR_LOG_PREFIX_(this->client), packet->packetId());
		}
	}

	{
		unique_lock queue_lock(read_queue.buffer_lock);

		if(ordered) { /* ordered */
			if(!read_queue.insert_index(packet_id, move(packet))) {
				debugMessage(this->client->getServerId(), "{} Got ordered packet of type {} which is out of the buffer range of {}. Packet ID: {}, Full index: {}. Dropping packet",
				             CLIENT_STR_LOG_PREFIX_(this->client),
				             packet_type.name(),
				             read_queue.capacity(),
				             packet_id,
				             read_queue.full_index()
				);
				/* return; dont stop here because we've to progress the packets */
			}
		} else {
			//TODO: Needs rethinking because read_queue.push_back increases the index, but this has not to be the packet id
			if(!read_queue.push_back(move(packet))) {
				debugMessage(this->client->getServerId(), "{} Got unordered packet of type {} which is out of the buffer capacity of {}. Packet ID: {}. Dropping packet.",
				             CLIENT_STR_LOG_PREFIX_(this->client),
				             packet_type.name(),
				             read_queue.capacity(),
				             packet_id,
				             read_queue.full_index()
				);
			}
			{
				//A max entropie of 16 packets should not happen. This indicates more that 16 or more packets got lost
				auto current_index = read_queue.current_index();
				if(current_index + 16 < packet_id)
					read_queue.set_full_index_to(packet_id);
			}
		}
	}

	auto voice_server = this->client->voice_server;
	if(voice_server)
		voice_server->schedule_execute(this->client);
}

bool VoiceClientConnection::verify_encryption(const pipes::buffer_view &packet /* incl. mac etc */) {
	if((packet[12] & 0x80) != 0) /* we want an encrypted packet to verify the encryption */
		return false;

	auto packet_type = (protocol::PacketType) (packet[12] & 0xF);
	if(packet_type == protocol::PING || packet_type == protocol::PONG) return false; /* these packets could never be encrypted */

	auto packet_id = (uint16_t) be2le16(&packet[8]);
	auto generation = this->_packet_buffers[packet_type].generation(packet_id);
	return this->crypt_handler.verify_encryption(packet, packet_id, generation);
}

void VoiceClientConnection::execute_handle_packet(const std::chrono::system_clock::time_point& /* scheduled */) {
	if(this->client->state == ConnectionState::DISCONNECTED || !this->client->getServer())
		return;

	bool reexecute_handle = false;
	unique_lock<std::recursive_timed_mutex> buffer_execute_lock;
	auto packet = this->next_reassembled_packet(buffer_execute_lock, reexecute_handle);

	if(packet){
		auto startTime = system_clock::now();
		try {
			const auto packet_type = packet->type();
			if(packet_type == PacketTypeInfo::Command || packet_type == PacketTypeInfo::CommandLow)
				this->client->handlePacketCommand(packet);
			else if(packet_type == PacketTypeInfo::Ack || packet_type == PacketTypeInfo::AckLow)
				this->client->handlePacketAck(packet);
			else if(packet_type == PacketTypeInfo::Voice || packet_type == PacketTypeInfo::VoiceWhisper)
				this->client->handlePacketVoice(packet);
			else if(packet_type == PacketTypeInfo::Ping || packet_type == PacketTypeInfo::Pong)
				this->client->handlePacketPing(packet);
			else if(packet_type == PacketTypeInfo::Init1)
				this->client->handlePacketInit(packet);
		} catch (std::exception& ex) {
			logCritical(this->client->getServerId(), "{} Exception reached root tree! {}", CLIENT_STR_LOG_PREFIX_(this->client), ex.what());
		}

		auto end = system_clock::now();
		if(end - startTime > milliseconds(10)) {
			if(packet->type() != PacketTypeInfo::Command && packet->type() != PacketTypeInfo::CommandLow) {
				logError(this->client->getServerId(),
				         "{} Handling of packet {} needs more than 10ms ({}ms)",
				         CLIENT_STR_LOG_PREFIX_(this->client),
				         packet->type().name(),
				         duration_cast<milliseconds>(end - startTime).count()
				);
			}
		}
	}
	if(buffer_execute_lock.owns_lock())
		buffer_execute_lock.unlock();

	auto voice_server = this->client->voice_server;
	if(voice_server && reexecute_handle)
		this->client->voice_server->schedule_execute(this->client);
}

/* buffer_execute_lock: lock for in order execution */
unique_ptr<protocol::ClientPacket> VoiceClientConnection::next_reassembled_packet(unique_lock<std::recursive_timed_mutex>& buffer_execute_lock, bool& more) {
	packet_buffer_t* buffer = nullptr;
	unique_lock<std::recursive_timed_mutex> buffer_lock; /* general buffer lock */

	{
		auto base_index = this->_packet_buffers_index;
		auto select_index = base_index;
		auto max_index = this->_packet_buffers.size();

		for(uint8_t index = 0; index < max_index; index++) {
			if(!buffer)
				select_index++;

			auto& buf = this->_packet_buffers[base_index++ % max_index];
			unique_lock ring_lock(buf.buffer_lock, try_to_lock);
			if(!ring_lock.owns_lock()) continue;

			if(buf.front_set()) {
				if(!buffer) { /* lets still test for reexecute */
					buffer_execute_lock = unique_lock(buf.execute_lock, try_to_lock);
					if(!buffer_execute_lock.owns_lock()) continue;

					buffer_lock = move(ring_lock);
					buffer = &buf;
				} else {
					more = true;
					break;
				}
			}
		}
		this->_packet_buffers_index = static_cast<uint8_t>(select_index % max_index); /* guarantee that we will not hangup with commands! */
	}

	if(!buffer)
		return nullptr; /* we've no packets */

	auto current_packet = &*buffer->slot_value(0);
	if(!current_packet) {
		logCritical(this->client->getServer()->getServerId(), "buffer->slot_value(0) returned nullptr, but set flag has been set!");
		return buffer->pop_front(); /* should be null! */
	}

	if(current_packet->type() != PacketTypeInfo::Command && current_packet->type() != PacketTypeInfo::CommandLow) {
		return buffer->pop_front(); /* we don't have to reassemble anything */
	}

	unique_ptr<ClientPacket> final_packet;
	uint16_t sequence_length = 1;

	if(current_packet->has_flag(PacketFlag::Fragmented)) {
		size_t buffer_length = ClientPacket::META_SIZE;
		do {
			if(sequence_length >= buffer->capacity()) {
				logError(this->client->getServerId(), "{} Received fragmented packets which have a too long order. Dropping queue, which will cause a client drop.", CLIENT_STR_LOG_PREFIX_(this->client));
				buffer->clear();
				return nullptr; /* we've nothing to handle */
			}

			buffer_length += current_packet->data_length();
			current_packet = &*buffer->slot_value(sequence_length++);
		} while(current_packet && !current_packet->has_flag(PacketFlag::Fragmented));
		if(!current_packet)
			return nullptr; /* we haven't found the end yet! */
		buffer_length += current_packet->data_length();

		/* okey we have all fragments lets reassemble */

		/*
		 * Packet sequence could never be so long. If it is so then the data_length() returned an invalid value.
		 * We're checking it here because we dont want to make a huge allocation
		 */
		assert(buffer_length < 512 * 1024 * 1024);

		pipes::buffer packet_buffer{buffer_length};
		char* packet_buffer_ptr = &packet_buffer[0];
		size_t packet_count = 0;

		unique_ptr<ClientPacket> packet;

		/* initialize packet flags etc */
		{
			packet = buffer->pop_front();
			packet_count++;

			if(!packet) {
				logCritical(this->client->getServer()->getServerId(), "buffer->pop_front() returned nullptr, but set flag has been set (0)!");
				return nullptr;
			}

			const auto length = packet->buffer().length();
			memcpy(packet_buffer_ptr, &packet->buffer()[0], length);
			packet_buffer_ptr += length;
		}

		while(packet_count < sequence_length) {
			packet = buffer->pop_front();
			packet_count++;

			if(!packet) {
				logCritical(this->client->getServer()->getServerId(), "buffer->pop_front() returned nullptr, but set flag has been set (1)!");
				return nullptr;
			}

			const auto length = packet->data_length();
			memcpy(packet_buffer_ptr, &packet->data()[0], length);
			packet_buffer_ptr += length;
		}

		if((packet_buffer_ptr - 1) != &packet_buffer[packet_buffer.length() - 1]) {
			logCritical(this->client->getServer()->getServerId(),
					"Buffer over/underflow: packet_buffer_ptr != &packet_buffer[packet_buffer.length() - 1]; packet_buffer_ptr := {}; packet_buffer.end() := {}",
					(void*) packet_buffer_ptr,
					(void*) &packet_buffer[packet_buffer.length() - 1]
			);
		}

		final_packet = make_unique<ClientPacket>(packet_buffer);
		final_packet->setCompressed(final_packet->has_flag(PacketFlag::Compressed));
	} else {
		final_packet = buffer->pop_front();
		if(!final_packet) {
			logCritical(this->client->getServer()->getServerId(), "buffer->pop_front() returned nullptr, but set flag has been set (3)!");
			return nullptr;
		}
	}

	buffer_lock.unlock();

	std::string error = "success";
	if(!this->compress_handler.progressPacketIn(&*final_packet, error)) {
		logError(this->client->getServerId(), "{} Failed to decompress received packet. Error: {}", CLIENT_STR_LOG_PREFIX_(this->client), error);
		final_packet = nullptr;
	}

	return final_packet;
}


void VoiceClientConnection::sendPacket(const shared_ptr<protocol::ServerPacket>& original_packet, bool copy, bool prepare_directly) {
	if(this->client->state == ConnectionState::DISCONNECTED) return;

	shared_ptr<protocol::ServerPacket> packet;
	if(copy) {
		packet.reset(new protocol::ServerPacket(original_packet->buffer().dup(buffer::allocate_buffer(original_packet->buffer().length()))));
		if(original_packet->getListener())
			packet->setListener(std::move(original_packet->getListener()));
		packet->memory_state.flags = original_packet->memory_state.flags;
	} else
		packet = original_packet;

	if(prepare_directly) {
		vector<pipes::buffer> buffers;
		this->prepare_process_count++;
		if(!this->prepare_packet_for_write(buffers, packet)) {
			logError(this->client->getServerId(), "{} Dropping packet!", CLIENT_STR_LOG_PREFIX_(this->client));
			this->prepare_process_count--;
			return;
		}

		/* enqueue buffers for write */
		{
			lock_guard write_queue_lock(this->write_queue_lock);
			this->write_queue.insert(this->write_queue.end(), buffers.begin(), buffers.end());
		}
		this->prepare_process_count--; /* we're now done preparing */
	} else {
		lock_guard prepare_queue_lock(this->write_prepare_queue_lock);
		this->write_prepare_queue.push_back(packet);
	}
	this->triggerWrite();
}

bool VoiceClientConnection::prepare_packet_for_write(vector<pipes::buffer> &result, const shared_ptr<ServerPacket> &packet) {
	string error = "success";

	if(packet->type().compressable() && !packet->memory_state.fragment_entry) {
		packet->enable_flag(PacketFlag::Compressed);
		if(!this->compress_handler.progressPacketOut(packet.get(), error)) {
			logError(this->getClient()->getServerId(), "{} Could not compress outgoing packet.\nThis could cause fatal failed for the client.\nError: {}", error);
			return false;
		}
	}

	std::vector<shared_ptr<ServerPacket>> fragments;
	fragments.reserve((size_t) (packet->data().length() / packet->type().max_length()) + 1);

	if(packet->data().length() > packet->type().max_length()){
		if(!packet->type().fragmentable()) {
			logError(this->client->getServerId(), "{} We've tried to send a too long, not fragmentable, packet. Dropping packet of type {} with length {}", CLIENT_STR_LOG_PREFIX_(this->client), packet->type().name(), packet->data().length());
			return false;
		}

		{ //Split packets
			auto buffer = packet->data();

			const auto max_length = packet->type().max_length();
			while(buffer.length() > max_length * 2) {
				fragments.push_back(make_shared<ServerPacket>(packet->type(), buffer.view(0, max_length).dup(buffer::allocate_buffer(max_length))));
				buffer = buffer.range((size_t) max_length);
			}

			if(buffer.length() > max_length) { //Divide rest by 2
				fragments.push_back(make_shared<ServerPacket>(packet->type(), buffer.view(0, buffer.length() / 2).dup(buffer::allocate_buffer(buffer.length() / 2))));
				buffer = buffer.range(buffer.length() / 2);
			}
			fragments.push_back(make_shared<ServerPacket>(packet->type(), buffer));

			for(const auto& frag : fragments) {
				frag->setFragmentedEntry(true);
				frag->enable_flag(PacketFlag::NewProtocol);
			}
		}

		assert(fragments.size() >= 2);
		fragments.front()->enable_flag(PacketFlag::Fragmented);
		if(packet->has_flag(PacketFlag::Compressed))
			fragments.front()->enable_flag(PacketFlag::Compressed);

		fragments.back()->enable_flag(PacketFlag::Fragmented);

		if(packet->getListener())
			fragments.back()->setListener(std::move(packet->getListener())); //Move the listener to the last :)
	} else
		fragments.push_back(packet);
	result.reserve(fragments.size());

	/* apply packet ids */
	{
		lock_guard id_lock(this->packet_id_manager_lock);
		for(const auto& fragment : fragments) {
			if(!fragment->memory_state.id_branded)
				fragment->applyPacketId(this->packet_id_manager);
		}
	}


	auto statistics = this->client && this->client->getServer() ? this->client->connectionStatistics : nullptr;
	for(const auto& fragment : fragments) {
		if(!this->crypt_handler.progressPacketOut(fragment.get(), error, false)){
			logError(this->client->getServerId(), "{} Failed to encrypt packet. Error: {}", CLIENT_STR_LOG_PREFIX_(this->client), error);
			return false;
		}
#ifndef CONNECTION_NO_STATISTICS
		if(statistics)
			statistics->logOutgoingPacket(*fragment);
#endif
		this->acknowledge_handler.process_packet(*fragment);
		result.push_back(fragment->buffer());
	}

	return true;
}

#ifdef VC_USE_READ_QUEUE
bool VoiceClientConnection::handleNextDatagram() {

	bool flag_empty;
	pipes::buffer buffer;
	{
		lock_guard<recursive_mutex> queue_lock(this->queueLock);
		if(this->readQueue.empty()) return false;
		buffer = std::move(this->readQueue.front());
		this->readQueue.pop_front();
		flag_empty = this->readQueue.empty();
	};

	try {
		this->handleDatagramReceived(buffer);
	} catch (std::exception& e) {
		logCritical(this->client->getServerId(), "Handling of raw packet thrown an uncaught exception! (Message: " + string(e.what()) + ")");
	}
	return flag_empty;
}
#endif

bool VoiceClientConnection::prepare_write_packets() {
	/* get the next packet to prepare */
	unique_lock prepare_queue_lock(this->write_prepare_queue_lock);
	if(this->write_prepare_queue.empty())
		return false;

	prepare_process_count++; /* we're not preparing a packet */
	auto packet = move(this->write_prepare_queue.front());
	this->write_prepare_queue.pop_front();
	auto flag_more = !this->write_prepare_queue.empty();
	prepare_queue_lock.unlock();

	/* prepare the next packet */
	vector<pipes::buffer> buffers;
	if(!this->prepare_packet_for_write(buffers, packet)) {
		logError(this->client->getServerId(), "{} Dropping packet!", CLIENT_STR_LOG_PREFIX_(this->client));
		this->prepare_process_count--;
		return flag_more;
	}

	/* enqueue buffers for write */
	{
		lock_guard write_queue_lock(this->write_queue_lock);
		this->write_queue.insert(this->write_queue.end(), buffers.begin(), buffers.end());
	}
	this->prepare_process_count--; /* we're now done preparing */

	return flag_more;
}

int VoiceClientConnection::pop_write_buffer(pipes::buffer& target) {
	if(this->client->state == DISCONNECTED)
		return 2;

	lock_guard write_queue_lock(this->write_queue_lock);
	size_t size = this->write_queue.size();
	if(size == 0)
		return 2;

	target = std::move(this->write_queue.front());
	this->write_queue.pop_front();

#ifdef FUZZING_TESTING_OUTGOING
	#ifdef FIZZING_TESTING_DISABLE_HANDSHAKE
	if (this->client->state == ConnectionState::CONNECTED) {
		#endif
		if ((rand() % FUZZING_TESTING_DROP_MAX) < FUZZING_TESTING_DROP) {
			debugMessage(this->client->getServerId(), "{}[FUZZING] Dropping outgoing packet", CLIENT_STR_LOG_PREFIX_(this->client));
			return 0;
		}
		#ifdef FIZZING_TESTING_DISABLE_HANDSHAKE
	}
	#endif
#endif
	return size > 1;
}

bool VoiceClientConnection::wait_empty_write_and_prepare_queue(chrono::time_point<chrono::system_clock> until) {
	while(true) {
		{
			lock_guard prepare_lock{this->write_prepare_queue_lock};
			if(!this->write_prepare_queue.empty())
				goto _wait;
			if(this->prepare_process_count != 0)
				goto _wait;
		}
		{
			lock_guard buffer_lock{this->write_queue_lock};
			if(!this->write_queue.empty())
				goto _wait;
			if(this->prepare_process_count != 0)
				goto _wait;
		}
		break;

		_wait:
		if(until.time_since_epoch().count() != 0 && system_clock::now() > until)
			return false;

	};
	threads::self::sleep_for(milliseconds(5));
	return true;
}

void VoiceClientConnection::reset() {
	this->acknowledge_handler.reset();
	this->crypt_handler.reset();
	{
		lock_guard manager_lock(this->packet_id_manager_lock);
		this->packet_id_manager.reset();
	}

	{
		lock_guard buffer_lock(this->packet_buffer_lock);
		for(auto& buffer : this->_packet_buffers)
			buffer.reset();
	}
}