Teaspeak-Server/server/src/client/voice/VoiceClient.cpp

#include <algorithm>
#include <memory>
#include <tomcrypt.h>
#include <arpa/inet.h>
#include <ThreadPool/Timer.h>
#include <misc/endianness.h>
#include <misc/memtracker.h>
#include <log/LogUtils.h>

#include "VoiceClient.h"
#include "src/VirtualServer.h"
#include "../../server/VoiceServer.h"

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

VoiceClient::VoiceClient(const std::shared_ptr<VoiceServer>& server, const sockaddr_storage* address) : SpeakingClient(server->server->sql, server->server), voice_server(server) {
    assert(address);
    memtrack::allocated<VoiceClient>(this);
    memcpy(&this->remote_address, address, sizeof(sockaddr_storage));

    debugMessage(this->server->getServerId(), " Creating VoiceClient instance at {}", (void*) this);
}

void VoiceClient::initialize() {
    this->event_handle_packet = make_shared<event::ProxiedEventEntry<VoiceClient>>(dynamic_pointer_cast<VoiceClient>(this->ref()), &VoiceClient::execute_handle_packet);

    this->properties()[property::CLIENT_TYPE] = ClientType::CLIENT_TEAMSPEAK;
    this->properties()[property::CLIENT_TYPE_EXACT] = ClientType::CLIENT_TEAMSPEAK;

    this->state = ConnectionState::INIT_HIGH;
    this->connection = new connection::VoiceClientConnection(this);
}

VoiceClient::~VoiceClient() {
    debugMessage(this->getServerId(), " Deleting VoiceClient instance at {}", (void*) this);

    this->state = ConnectionState::DISCONNECTED;
    delete this->connection;
    this->connection = nullptr;

    if(this->flushing_thread)
        logCritical(this->getServerId(), "Deleting a VoiceClient which should still be hold within the flush thread!");

    memtrack::freed<VoiceClient>(this);
}

void VoiceClient::sendCommand0(const std::string_view& cmd, bool low, bool direct, std::unique_ptr<threads::Future<bool>> listener) {
    if(cmd.empty()) {
        logCritical(this->getServerId(), "{} Attempted to send an empty command!", CLIENT_STR_LOG_PREFIX);
        return;
    }

    auto packet = make_shared<protocol::ServerPacket>(
            low ? protocol::PacketTypeInfo::CommandLow : protocol::PacketTypeInfo::Command,
            pipes::buffer_view{(void*) cmd.data(), cmd.length()}
    );
    if(low) {
        packet->enable_flag(protocol::PacketFlag::NewProtocol);
    }
    packet->setListener(std::move(listener));
    this->connection->sendPacket(packet, false, direct);

#ifdef PKT_LOG_CMD
    logTrace(this->getServerId(), "{}[Command][Server -> Client] Sending command {}. Command low: {}. Full command: {}", CLIENT_STR_LOG_PREFIX, cmd.substr(0, cmd.find(' ')), low, cmd);
#endif
}
void VoiceClient::sendAcknowledge(uint16_t packetId, bool low) {
    char buffer[2];
    le2be16(packetId, buffer);

    auto packet = make_shared<protocol::ServerPacket>(low ? protocol::PacketTypeInfo::AckLow : protocol::PacketTypeInfo::Ack, pipes::buffer_view{buffer, 2});
    packet->enable_flag(PacketFlag::Unencrypted);
    if(!low) packet->enable_flag(protocol::PacketFlag::NewProtocol);
    this->connection->sendPacket(packet);
#ifdef PKT_LOG_ACK
    logTrace(this->getServerId(), "{}[Acknowledge][Server -> Client] Sending acknowledge for {}", CLIENT_STR_LOG_PREFIX, packetId);
#endif
}

void VoiceClient::tick(const std::chrono::system_clock::time_point &time) {
    SpeakingClient::tick(time);
    {
        ALARM_TIMER(A1, "VoiceClient::tick", milliseconds(3));
        if(this->state == ConnectionState::CONNECTED) {
            if(this->lastPingRequest > this->lastPingResponse) { //Client is behind :)
                if(this->lastPingRequest - this->lastPingResponse > chrono::seconds(20)) {
                    debugMessage(this->getServerId(), "{} Got a ping timeout. (Last successful ping: {}ms ago. Last request {}ms. Last response {}ms). Trying to recover via command acknowledge.",
                                 CLIENT_STR_LOG_PREFIX,
                                 duration_cast<milliseconds>(this->lastPingRequest - this->lastPingResponse).count(),
                                 duration_cast<milliseconds>(time - this->lastPingRequest).count(),
                                 duration_cast<milliseconds>(time - this->lastPingResponse).count());

                    bool force;
                    this->request_connection_info(nullptr, force);
                    this->lastPingResponse = system_clock::now();
                    return;
                }
            }
            if(time - this->lastPingRequest >= chrono::milliseconds(1000)) {
                //TODO calculate the ping smooth
                if(this->lastPingResponse < this->lastPingRequest){
                    if(time - this->lastPingRequest >= chrono::milliseconds(1500)) { //Max
                        this->sendPingRequest();
                    }
                } else
                    this->sendPingRequest();
            }
        } else if(this->state == ConnectionState::INIT_LOW || this->state == ConnectionState::INIT_HIGH) {
            if(this->last_packet_handshake.time_since_epoch().count() != 0) {
                if(time - this->last_packet_handshake > seconds(5)) {
                    debugMessage(this->getServerId(), "{} Got handshake timeout. {}. State: {} Time: {}", CLIENT_STR_LOG_PREFIX,
                                 this->getLoggingPeerIp() + ":" + to_string(this->getPeerPort()),
                                 this->state == ConnectionState::INIT_HIGH ? "INIT_HIGH" : "INIT_LOW",
                                 duration_cast<seconds>(time - this->last_packet_handshake).count()
                    );
                    this->closeConnection(system_clock::now() + seconds(1));
                }
            }
        }
    }
}

bool VoiceClient::disconnect(const std::string &reason) {
    return this->disconnect(VREASON_SERVER_KICK, reason, this->server->serverRoot, true);
}

bool VoiceClient::disconnect(ts::ViewReasonId reason_id, const std::string &reason, const std::shared_ptr<ts::server::ConnectedClient>& invoker, bool notify_viewer) {
    {
        threads::MutexLock disconnect_lock(this->disconnectLock);
        if(this->state == ConnectionState::DISCONNECTING || this->state == ConnectionState::DISCONNECTED) return false; //Already disconnecting/disconnected
        this->state = ConnectionState::DISCONNECTING;
    }

    Command cmd("notifyclientleftview");
    cmd["reasonmsg"] = reason;
    cmd["reasonid"] = reason_id;
    cmd["clid"] = this->getClientId();
    cmd["cfid"] = this->currentChannel ? this->currentChannel->channelId() : 0; //Failed when cid = 0????
    cmd["ctid"] = 0;

    if (invoker) {
        cmd["invokerid"] = invoker->getClientId();
        cmd["invokername"] = invoker->getDisplayName();
        cmd["invokeruid"] = invoker->getUid();
    }

    if(notify_viewer && this->server) {
        unique_lock channel_lock(this->server->channel_tree_lock);
        this->server->client_move(this->ref(), nullptr, invoker, reason, reason_id, false, channel_lock);
    } else {
        threads::MutexLock lock(this->command_lock);
        auto server_channel = dynamic_pointer_cast<ServerChannel>(this->currentChannel);
        if(server_channel)
            server_channel->unregister_client(_this.lock());
        this->currentChannel = nullptr;
    }

    auto listener = make_unique<threads::Future<bool>>();
    auto weak_self = this->_this;
    listener->waitAndGetLater([weak_self](bool* success) {
        if(weak_self.expired()) return;
        auto self = weak_self.lock();
        if(!self || self->state != DISCONNECTING) return;

        if(!success || !*success) {
            debugMessage(self->getServerId(), "{} Failed to receive disconnect acknowledge!", CLIENT_STR_LOG_PREFIX_(self));
        } else
            debugMessage(self->getServerId(), "{} Received disconnect acknowledge!", CLIENT_STR_LOG_PREFIX_(self));

        self->closeConnection();
    }, system_clock::now() + seconds(5));
    this->sendCommand0(cmd.build(), false, false, std::move(listener));
    return true;
}

bool VoiceClient::closeConnection(const system_clock::time_point &timeout) {
    auto self_lock  = dynamic_pointer_cast<VoiceClient>(_this.lock());
    assert(self_lock); //Should never happen!

    threads::MutexLock disconnect_lock(this->disconnectLock);
    bool flush = timeout.time_since_epoch().count() > 0;
    if((this->state == ConnectionState::DISCONNECTING && flush && this->flushing_thread) || this->state == ConnectionState::DISCONNECTED){
        debugMessage(this->getServerId(), "{} Tried to disconnect, but isn't connected anymore! State: {}", CLIENT_STR_LOG_PREFIX, this->state);
        return false;
    }
    this->state = flush ? ConnectionState::DISCONNECTING : ConnectionState::DISCONNECTED;

    debugMessage(this->getServerId(), "{} Closing voice client connection. (Flush: {})", CLIENT_STR_LOG_PREFIX, flush);
    if(flush) {
        this->flushing_thread = std::make_shared<threads::Thread>(THREAD_SAVE_OPERATIONS | THREAD_EXECUTE_LATER, [this, self_lock, timeout](){
            //We could use this here cause its locked by self_locked
            debugMessage(this->getServerId(), "{} Awaiting write prepare, write and acknowledge queue flushed", CLIENT_STR_LOG_PREFIX);
            auto connection = this->getConnection();

            this->getConnection()->wait_empty_write_and_prepare_queue(timeout);
            debugMessage(this->getServerId(), "{} Write prepare queue progressed", CLIENT_STR_LOG_PREFIX);
            while(this->state == DISCONNECTING){
                if(system_clock::now() > timeout){
                    debugMessage(this->getServerId(), "{} Cant flush io!", CLIENT_STR_LOG_PREFIX);

                    if(!this->connection->wait_empty_write_and_prepare_queue(timeout)) {
                        debugMessage(this->getServerId(), "{}   Write queue not empty!", CLIENT_STR_LOG_PREFIX);
                    }
                    {
                        auto reminding = connection->acknowledge_handler.awaiting_acknowledge();
                        if(reminding > 0)
                            debugMessage(this->getServerId(), "{} Could not get acknowledge for all commands before disconnecting. Acknowledges left: {}", CLIENT_STR_LOG_PREFIX, reminding);
                    }
                    break;
                }
                if(!this->connection->wait_empty_write_and_prepare_queue(timeout))
                    continue;

                {
                    if(connection->acknowledge_handler.awaiting_acknowledge() > 0) {
                        usleep(5000);
                        continue;
                    }
                }
                debugMessage(this->getServerId(), "{} Write and acknowledge queue are flushed", CLIENT_STR_LOG_PREFIX);
                break;
            }
            if(this->state != DISCONNECTING) return;
            this->finalDisconnect();
        });
        flushing_thread->name("Flush thread VC").execute();
        return true;
    } else {
        this->state = DISCONNECTED;
        auto f_thread = this->flushing_thread;
        if(f_thread) {
            threads::NegatedMutexLock l(this->disconnectLock); //Unlock the close lock again until the flush queue has finished (may with close may by interupt)
            f_thread->join();
        }
        this->finalDisconnect();
    }
    return true;
}

void VoiceClient::finalDisconnect() {
    auto ownLock = dynamic_pointer_cast<VoiceClient>(_this.lock());
    assert(ownLock);

    threads::MutexLock disconnect_lock(this->disconnectLock);
    lock_guard disconnect_lock_final(this->finalDisconnectLock);
    if(this->final_disconnected) {
        logError(this->getServerId(), "Tried to final disconnect {}/{} twice", this->getLoggingPeerIp() + ":" + to_string(this->getPeerPort()), this->getDisplayName());
        return;
    }
    this->final_disconnected = true;
    this->state = ConnectionState::DISCONNECTED;

    threads::MutexLock command_lock(this->command_lock); //We should not progress any commands while disconnecting
    //Unload manager cache
    this->processLeave();
    {
        if(this->flushing_thread) this->flushing_thread->detach(); //The thread itself should be already done or executing this method
        this->flushing_thread.reset();
    }
    if(this->voice_server) this->voice_server->unregisterConnection(ownLock);
}

void VoiceClient::execute_handle_packet(const std::chrono::system_clock::time_point &time) {
    this->connection->execute_handle_packet(time);
}

void VoiceClient::send_voice_packet(const pipes::buffer_view &voice_buffer, const SpeakingClient::VoicePacketFlags &flags) {
    auto packet = make_shared<ServerPacket>(PacketTypeInfo::Voice, voice_buffer.length());
    {
        PacketFlag::PacketFlags packet_flags = PacketFlag::None;
        packet_flags |= flags.encrypted ? 0 : PacketFlag::Unencrypted;
        packet_flags |= flags.head ? PacketFlag::Compressed : 0;
        packet_flags |= flags.fragmented ? PacketFlag::Fragmented : 0;
        packet_flags |= flags.new_protocol ? PacketFlag::NewProtocol : 0;
        packet->set_flags(packet_flags);
    }

    memcpy(packet->data().data_ptr<void>(), voice_buffer.data_ptr<void>(), voice_buffer.length());
    this->connection->sendPacket(packet, false, false);
}

void VoiceClient::send_voice_whisper_packet(const pipes::buffer_view &voice_buffer, const SpeakingClient::VoicePacketFlags &flags) {
    auto packet = make_shared<ServerPacket>(PacketTypeInfo::VoiceWhisper, voice_buffer.length());
    {
        PacketFlag::PacketFlags packet_flags = PacketFlag::None;
        packet_flags |= flags.encrypted ? 0 : PacketFlag::Unencrypted;
        packet_flags |= flags.head ? PacketFlag::Compressed : 0;
        packet_flags |= flags.fragmented ? PacketFlag::Fragmented : 0;
        packet_flags |= flags.new_protocol ? PacketFlag::NewProtocol : 0;
        packet->set_flags(packet_flags);
    }

    memcpy(packet->data().data_ptr<void>(), voice_buffer.data_ptr<void>(), voice_buffer.length());
    this->connection->sendPacket(packet, false, false);
}