shard.hpp

//
// shard.hpp
// *********
//
// Copyright (c) 2019 Sharon W (sharon at aegis dot gg)
//
// Distributed under the MIT License. (See accompanying file LICENSE)
//

#pragma once

#include "aegis/config.hpp"
#include "aegis/utility.hpp"
#include "aegis/fwd.hpp"
#include <memory>
#include <map>
#include <string>
#include <chrono>
#include <stdint.h>
#include <asio/io_context.hpp>
#ifdef WIN32
# include "aegis/push.hpp"
#endif
#include <websocketpp/config/asio_client.hpp>
#include <websocketpp/common/connection_hdl.hpp>
#include <websocketpp/client.hpp>
#ifdef WIN32
# include "aegis/pop.hpp"
#endif
#include <spdlog/fmt/fmt.h>
#include "aegis/zstr/zstr.hpp"
#include "aegis/gateway/objects/presence.hpp"
#include "aegis/gateway/objects/activity.hpp"

namespace aegis
{

namespace shards
{

class shard
{
public:
    using connection_ptr = websocketpp::client<websocketpp::config::asio_tls_client>::connection_type::ptr;

    AEGIS_DECL shard(asio::io_context & _io, websocketpp::client<websocketpp::config::asio_tls_client> & _ws, int32_t id);

    AEGIS_DECL void do_reset(shard_status _status = shard_status::closed) noexcept;


    int64_t get_sequence() const noexcept
    {
        return _sequence;
    }


    int32_t get_id() const noexcept
    {
        return _id;
    }


    AEGIS_DECL bool is_connected() const noexcept;


    AEGIS_DECL bool is_online() const noexcept;


    AEGIS_DECL void send(const std::string & payload, websocketpp::frame::opcode::value op = websocketpp::frame::opcode::text);

    AEGIS_DECL void send_now(const std::string & payload, websocketpp::frame::opcode::value op = websocketpp::frame::opcode::text);


    std::string get_transfer_str() const noexcept
    {
        return utility::format_bytes(transfer_bytes);
    }


    std::string get_transfer_u_str() const noexcept
    {
        return utility::format_bytes(transfer_bytes_u);
    }


    uint64_t get_transfer() const noexcept
    {
        return transfer_bytes;
    }


    uint64_t get_transfer_u() const noexcept
    {
        return transfer_bytes_u;
    }

    struct
    {
        int64_t messages;
        int64_t presence_changes;
        int64_t reconnects;
    } counters = { 0,0,0 };

    AEGIS_DECL void connect();


    AEGIS_DECL std::string uptime_str() const noexcept
    {
        return utility::uptime_str(_ready_time);
    }


    int64_t uptime() const noexcept
    {
        return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - _ready_time).count();
    }

    void set_heartbeat(std::function<void(const asio::error_code &, const std::chrono::milliseconds, shard *)> fnkeepalive)
    {
        keepalivefunc = fnkeepalive;
    }

    AEGIS_DECL void start_heartbeat(int32_t heartbeat) noexcept;

#if defined(AEGIS_DEBUG_HISTORY)
    std::deque<std::tuple<std::chrono::steady_clock::time_point, std::string>> debug_messages;
#endif
    asio::steady_timer keepalivetimer;
    asio::steady_timer delayedauth;
    asio::steady_timer write_timer;

    std::queue<std::tuple<std::string, websocketpp::frame::opcode::value>> write_queue;

    int32_t heartbeattime;

    std::chrono::steady_clock::time_point heartbeat_ack;
    std::chrono::steady_clock::time_point lastheartbeat;
    std::chrono::steady_clock::time_point lastwsevent;
    std::chrono::steady_clock::time_point last_status_time;
    std::chrono::steady_clock::time_point _ready_time;
    std::chrono::steady_clock::time_point last_ws_write;
    std::chrono::steady_clock::time_point connect_time;
    std::chrono::steady_clock::time_point closing_time;

    shard_status connection_state;
    std::string session_id;
    std::function<void(const asio::error_code &, const std::chrono::milliseconds, shard *)> keepalivefunc;

    void set_sequence(int64_t seq) noexcept
    {
        _sequence = seq;
    }

    void set_id(int32_t shard_id) noexcept
    {
        _id = shard_id;
    }

    connection_ptr get_connection() noexcept
    {
        return _connection;
    }

    const std::vector<std::string> & get_trace() const noexcept
    {
        return _trace;
    }


    AEGIS_DECL void update_presence(const std::string& text, gateway::objects::activity::activity_type type = gateway::objects::activity::Game, gateway::objects::presence::user_status status = gateway::objects::presence::Online);


private:
    friend class shard_mgr;
    friend class aegis::core;

    bool state_valid()
    {
        if (_connection == nullptr || _strand == nullptr)
            return false;
        return true;
    }

    AEGIS_DECL void process_writes(const asio::error_code & ec);
    AEGIS_DECL void _reset();
    AEGIS_DECL void set_connected();

    connection_ptr _connection;

    int64_t _sequence;
    int32_t _id;

    asio::io_context & _io_context;

    uint64_t transfer_bytes;

    uint64_t transfer_bytes_u;

    websocketpp::client<websocketpp::config::asio_tls_client> & _websocket;

    std::stringstream ws_buffer;
    std::unique_ptr<zstr::istream> zlib_ctx;

    // Websocket++ socket connection
    websocketpp::connection_hdl hdl;
    std::vector<std::string> _trace;
    std::shared_ptr<asio::io_context::strand> _strand;

    heartbeat_status _heartbeat_status = heartbeat_status::normal;
};

}

}

#if defined(AEGIS_HEADER_ONLY)
#include "aegis/shards/impl/shard.cpp"
#endif