mess with some websocket stuff to try and fix things

to be honest, im not sure what ive done here. whatever memory i have of the issue i was trying to fix has long disappeared by the time im committing this. theres still some issues with being force disconnected and i really dont understand it
ill figure it out eventually... maybe :/
This commit is contained in:
ouwou
2022-11-15 02:14:49 -05:00
parent 38c5230a1d
commit 929ebf1a60
6 changed files with 137 additions and 34 deletions

View File

@@ -9,7 +9,8 @@ using namespace std::string_literals;
DiscordClient::DiscordClient(bool mem_store) DiscordClient::DiscordClient(bool mem_store)
: m_decompress_buf(InflateChunkSize) : m_decompress_buf(InflateChunkSize)
, m_store(mem_store) { , m_store(mem_store)
, m_websocket("gateway-ws") {
m_msg_dispatch.connect(sigc::mem_fun(*this, &DiscordClient::MessageDispatch)); m_msg_dispatch.connect(sigc::mem_fun(*this, &DiscordClient::MessageDispatch));
auto dispatch_cb = [this]() { auto dispatch_cb = [this]() {
m_generic_mutex.lock(); m_generic_mutex.lock();
@@ -2171,6 +2172,11 @@ void DiscordClient::HandleGatewayVoiceStateUpdate(const GatewayMessage &msg) {
if (data.UserID == m_user_data.ID) { if (data.UserID == m_user_data.ID) {
spdlog::get("discord")->debug("Voice session ID: {}", data.SessionID); spdlog::get("discord")->debug("Voice session ID: {}", data.SessionID);
m_voice.SetSessionID(data.SessionID); m_voice.SetSessionID(data.SessionID);
// channel_id = null means disconnect. stop cuz out of order maybe
if (!data.ChannelID.has_value()) {
m_voice.Stop();
}
} else { } else {
if (data.GuildID.has_value() && data.Member.has_value()) { if (data.GuildID.has_value() && data.Member.has_value()) {
if (data.Member->User.has_value()) { if (data.Member->User.has_value()) {
@@ -2504,9 +2510,8 @@ void DiscordClient::SetSuperPropertiesFromIdentity(const IdentifyMessage &identi
void DiscordClient::HandleSocketOpen() { void DiscordClient::HandleSocketOpen() {
} }
void DiscordClient::HandleSocketClose(uint16_t code) { void DiscordClient::HandleSocketClose(const ix::WebSocketCloseInfo &info) {
printf("got socket close code: %d\n", code); auto close_code = static_cast<GatewayCloseCode>(info.code);
auto close_code = static_cast<GatewayCloseCode>(code);
auto cb = [this, close_code]() { auto cb = [this, close_code]() {
m_heartbeat_waiter.kill(); m_heartbeat_waiter.kill();
if (m_heartbeat_thread.joinable()) m_heartbeat_thread.join(); if (m_heartbeat_thread.joinable()) m_heartbeat_thread.join();

View File

@@ -289,7 +289,7 @@ private:
void SetSuperPropertiesFromIdentity(const IdentifyMessage &identity); void SetSuperPropertiesFromIdentity(const IdentifyMessage &identity);
void HandleSocketOpen(); void HandleSocketOpen();
void HandleSocketClose(uint16_t code); void HandleSocketClose(const ix::WebSocketCloseInfo &info);
static bool CheckCode(const http::response_type &r); static bool CheckCode(const http::response_type &r);
static bool CheckCode(const http::response_type &r, int expected); static bool CheckCode(const http::response_type &r, int expected);

View File

@@ -126,18 +126,26 @@ UDPSocket::type_signal_data UDPSocket::signal_data() {
} }
DiscordVoiceClient::DiscordVoiceClient() { DiscordVoiceClient::DiscordVoiceClient() {
sodium_init(); if (sodium_init() == -1) {
spdlog::get("voice")->critical("sodium_init() failed");
}
m_ws.signal_open().connect([this]() { m_ws = std::make_unique<Websocket>("voice-ws");
m_ws->signal_open().connect([this]() {
spdlog::get("voice")->info("Websocket open"); spdlog::get("voice")->info("Websocket open");
SetState(State::Opened);
}); });
m_ws.signal_close().connect([this](uint16_t code) { m_ws->signal_close().connect([this](const ix::WebSocketCloseInfo &info) {
spdlog::get("voice")->info("Websocket closed with code {}", code); if (info.remote) {
SetState(State::ClosingByServer);
}
Stop(); Stop();
}); });
m_ws.signal_message().connect([this](const std::string &str) { m_ws->signal_message().connect([this](const std::string &str) {
spdlog::get("voice-ws")->trace("Recv: {}", str);
std::lock_guard<std::mutex> _(m_dispatch_mutex); std::lock_guard<std::mutex> _(m_dispatch_mutex);
m_message_queue.push(str); m_message_queue.push(str);
m_dispatcher.emit(); m_dispatcher.emit();
@@ -175,7 +183,8 @@ DiscordVoiceClient::~DiscordVoiceClient() {
} }
void DiscordVoiceClient::Start() { void DiscordVoiceClient::Start() {
m_ws.StartConnection("wss://" + m_endpoint + "/?v=7"); SetState(State::Opening);
m_ws->StartConnection("wss://" + m_endpoint + "/?v=7");
m_heartbeat_waiter.revive(); m_heartbeat_waiter.revive();
m_keepalive_waiter.revive(); m_keepalive_waiter.revive();
m_connected = true; m_connected = true;
@@ -183,15 +192,29 @@ void DiscordVoiceClient::Start() {
} }
void DiscordVoiceClient::Stop() { void DiscordVoiceClient::Stop() {
m_ws.Stop(); if (IsOpening() || IsOpened()) {
m_udp.Stop(); spdlog::get("voice")->debug("Requested voice client stop");
m_heartbeat_waiter.kill(); SetState(State::ClosingByClient);
if (m_heartbeat_thread.joinable()) m_heartbeat_thread.join(); m_ws->Stop();
m_keepalive_waiter.kill(); } else if (IsClosing()) {
if (m_keepalive_thread.joinable()) m_keepalive_thread.join(); spdlog::get("voice")->debug("Completing stop in closing");
if (m_connected) { if (m_state == State::ClosingByClient) {
m_connected = false; SetState(State::ClosedByClient);
m_signal_disconnected.emit(); } else if (m_state == State::ClosingByServer) {
SetState(State::ClosedByServer);
}
m_ws->Stop();
m_udp.Stop();
m_heartbeat_waiter.kill();
if (m_heartbeat_thread.joinable()) m_heartbeat_thread.join();
m_keepalive_waiter.kill();
if (m_keepalive_thread.joinable()) m_keepalive_thread.join();
if (m_connected) {
m_connected = false;
m_signal_disconnected.emit();
}
} else {
spdlog::get("voice")->debug("Stop called, but already stopped");
} }
} }
@@ -228,7 +251,6 @@ bool DiscordVoiceClient::IsConnected() const noexcept {
void DiscordVoiceClient::OnGatewayMessage(const std::string &str) { void DiscordVoiceClient::OnGatewayMessage(const std::string &str) {
VoiceGatewayMessage msg = nlohmann::json::parse(str); VoiceGatewayMessage msg = nlohmann::json::parse(str);
puts(msg.Data.dump(4).c_str());
switch (msg.Opcode) { switch (msg.Opcode) {
case VoiceGatewayOp::Hello: { case VoiceGatewayOp::Hello: {
HandleGatewayHello(msg); HandleGatewayHello(msg);
@@ -278,7 +300,7 @@ void DiscordVoiceClient::HandleGatewaySessionDescription(const VoiceGatewayMessa
msg.Delay = 0; msg.Delay = 0;
msg.SSRC = m_ssrc; msg.SSRC = m_ssrc;
msg.Speaking = VoiceSpeakingType::Microphone; msg.Speaking = VoiceSpeakingType::Microphone;
m_ws.Send(msg); m_ws->Send(msg);
m_secret_key = d.SecretKey; m_secret_key = d.SecretKey;
m_udp.SetSSRC(m_ssrc); m_udp.SetSSRC(m_ssrc);
@@ -304,7 +326,7 @@ void DiscordVoiceClient::Identify() {
msg.SessionID = m_session_id; msg.SessionID = m_session_id;
msg.Token = m_token; msg.Token = m_token;
msg.Video = true; msg.Video = true;
m_ws.Send(msg); m_ws->Send(msg);
} }
void DiscordVoiceClient::Discovery() { void DiscordVoiceClient::Discovery() {
@@ -341,7 +363,7 @@ void DiscordVoiceClient::SelectProtocol(std::string_view ip, uint16_t port) {
msg.Address = ip; msg.Address = ip;
msg.Port = port; msg.Port = port;
msg.Protocol = "udp"; msg.Protocol = "udp";
m_ws.Send(msg); m_ws->Send(msg);
} }
void DiscordVoiceClient::OnUDPData(std::vector<uint8_t> data) { void DiscordVoiceClient::OnUDPData(std::vector<uint8_t> data) {
@@ -370,7 +392,7 @@ void DiscordVoiceClient::HeartbeatThread() {
VoiceHeartbeatMessage msg; VoiceHeartbeatMessage msg;
msg.Nonce = static_cast<uint64_t>(ms); msg.Nonce = static_cast<uint64_t>(ms);
m_ws.Send(msg); m_ws->Send(msg);
} }
} }
@@ -386,6 +408,47 @@ void DiscordVoiceClient::KeepaliveThread() {
} }
} }
void DiscordVoiceClient::SetState(State state) {
m_state = state;
switch (state) {
case State::Opening:
spdlog::get("voice")->debug("WS state: Opening");
break;
case State::Opened:
spdlog::get("voice")->debug("WS state: Opened");
break;
case State::ClosingByClient:
spdlog::get("voice")->debug("WS state: Closing (Client)");
break;
case State::ClosingByServer:
spdlog::get("voice")->debug("WS state: Closing (Server)");
break;
case State::ClosedByClient:
spdlog::get("voice")->debug("WS state: Closed (Client)");
break;
case State::ClosedByServer:
spdlog::get("voice")->debug("WS state: Closed (Server)");
break;
}
}
bool DiscordVoiceClient::IsOpening() const noexcept {
return m_state == State::Opening;
}
bool DiscordVoiceClient::IsOpened() const noexcept {
return m_state == State::Opened;
}
bool DiscordVoiceClient::IsClosing() const noexcept {
return m_state == State::ClosingByClient || m_state == State::ClosingByServer;
}
bool DiscordVoiceClient::IsClosed() const noexcept {
return m_state == State::ClosedByClient || m_state == State::ClosedByServer;
}
DiscordVoiceClient::type_signal_disconnected DiscordVoiceClient::signal_connected() { DiscordVoiceClient::type_signal_disconnected DiscordVoiceClient::signal_connected() {
return m_signal_connected; return m_signal_connected;
} }

View File

@@ -15,6 +15,7 @@
// clang-format on // clang-format on
enum class VoiceGatewayCloseCode : uint16_t { enum class VoiceGatewayCloseCode : uint16_t {
Normal = 4000,
UnknownOpcode = 4001, UnknownOpcode = 4001,
InvalidPayload = 4002, InvalidPayload = 4002,
NotAuthenticated = 4003, NotAuthenticated = 4003,
@@ -228,7 +229,10 @@ private:
std::array<uint8_t, 32> m_secret_key; std::array<uint8_t, 32> m_secret_key;
Websocket m_ws; // this is a unique_ptr because Websocket/ixwebsocket seems to have some strange behavior
// and quite frankly i do not feel like figuring out what is wrong
// so using a unique_ptr will just let me nuke the whole thing and make a new one
std::unique_ptr<Websocket> m_ws;
UDPSocket m_udp; UDPSocket m_udp;
Glib::Dispatcher m_dispatcher; Glib::Dispatcher m_dispatcher;
@@ -246,6 +250,24 @@ private:
std::atomic<bool> m_connected = false; std::atomic<bool> m_connected = false;
enum class State {
Opening,
Opened,
ClosingByClient,
ClosedByClient,
ClosingByServer,
ClosedByServer,
};
void SetState(State state);
[[nodiscard]] bool IsOpening() const noexcept;
[[nodiscard]] bool IsOpened() const noexcept;
[[nodiscard]] bool IsClosing() const noexcept;
[[nodiscard]] bool IsClosed() const noexcept;
std::atomic<State> m_state;
using type_signal_connected = sigc::signal<void()>; using type_signal_connected = sigc::signal<void()>;
using type_signal_disconnected = sigc::signal<void()>; using type_signal_disconnected = sigc::signal<void()>;
using type_signal_speaking = sigc::signal<void(VoiceSpeakingData)>; using type_signal_speaking = sigc::signal<void(VoiceSpeakingData)>;

View File

@@ -1,19 +1,25 @@
#include "websocket.hpp" #include "websocket.hpp"
#include <spdlog/sinks/stdout_color_sinks.h>
#include <utility> #include <utility>
Websocket::Websocket() Websocket::Websocket(const std::string &id)
: m_close_code(ix::WebSocketCloseConstants::kNormalClosureCode) { : m_close_info { 1000, "Normal", false } {
if (m_log = spdlog::get(id); !m_log) {
m_log = spdlog::stdout_color_mt(id);
}
m_open_dispatcher.connect([this]() { m_open_dispatcher.connect([this]() {
m_signal_open.emit(); m_signal_open.emit();
}); });
m_close_dispatcher.connect([this]() { m_close_dispatcher.connect([this]() {
Stop(); Stop();
m_signal_close.emit(m_close_code); m_signal_close.emit(m_close_info);
}); });
} }
void Websocket::StartConnection(const std::string &url) { void Websocket::StartConnection(const std::string &url) {
m_log->debug("Starting connection to {}", url);
m_websocket.disableAutomaticReconnection(); m_websocket.disableAutomaticReconnection();
m_websocket.setUrl(url); m_websocket.setUrl(url);
m_websocket.setOnMessageCallback([this](auto &&msg) { OnMessage(std::forward<decltype(msg)>(msg)); }); m_websocket.setOnMessageCallback([this](auto &&msg) { OnMessage(std::forward<decltype(msg)>(msg)); });
@@ -34,16 +40,18 @@ void Websocket::SetPrintMessages(bool show) noexcept {
} }
void Websocket::Stop() { void Websocket::Stop() {
m_log->debug("Stopping with default close code");
Stop(ix::WebSocketCloseConstants::kNormalClosureCode); Stop(ix::WebSocketCloseConstants::kNormalClosureCode);
} }
void Websocket::Stop(uint16_t code) { void Websocket::Stop(uint16_t code) {
m_log->debug("Stopping with close code {}", code);
m_websocket.stop(code); m_websocket.stop(code);
} }
void Websocket::Send(const std::string &str) { void Websocket::Send(const std::string &str) {
if (m_print_messages) if (m_print_messages)
printf("sending %s\n", str.c_str()); m_log->trace("Send: {}", str);
m_websocket.sendText(str); m_websocket.sendText(str);
} }
@@ -54,10 +62,12 @@ void Websocket::Send(const nlohmann::json &j) {
void Websocket::OnMessage(const ix::WebSocketMessagePtr &msg) { void Websocket::OnMessage(const ix::WebSocketMessagePtr &msg) {
switch (msg->type) { switch (msg->type) {
case ix::WebSocketMessageType::Open: { case ix::WebSocketMessageType::Open: {
m_log->debug("Received open frame, dispatching");
m_open_dispatcher.emit(); m_open_dispatcher.emit();
} break; } break;
case ix::WebSocketMessageType::Close: { case ix::WebSocketMessageType::Close: {
m_close_code = msg->closeInfo.code; m_log->debug("Received close frame, dispatching. {} ({}){}", msg->closeInfo.code, msg->closeInfo.reason, msg->closeInfo.remote ? " Remote" : "");
m_close_info = msg->closeInfo;
m_close_dispatcher.emit(); m_close_dispatcher.emit();
} break; } break;
case ix::WebSocketMessageType::Message: { case ix::WebSocketMessageType::Message: {

View File

@@ -6,10 +6,11 @@
#include <glibmm.h> #include <glibmm.h>
#include <nlohmann/json.hpp> #include <nlohmann/json.hpp>
#include <sigc++/sigc++.h> #include <sigc++/sigc++.h>
#include <spdlog/spdlog.h>
class Websocket { class Websocket {
public: public:
Websocket(); Websocket(const std::string &id);
void StartConnection(const std::string &url); void StartConnection(const std::string &url);
void SetUserAgent(std::string agent); void SetUserAgent(std::string agent);
@@ -30,7 +31,7 @@ private:
public: public:
using type_signal_open = sigc::signal<void>; using type_signal_open = sigc::signal<void>;
using type_signal_close = sigc::signal<void, uint16_t>; using type_signal_close = sigc::signal<void, ix::WebSocketCloseInfo>;
using type_signal_message = sigc::signal<void, std::string>; using type_signal_message = sigc::signal<void, std::string>;
type_signal_open signal_open(); type_signal_open signal_open();
@@ -46,5 +47,7 @@ private:
Glib::Dispatcher m_open_dispatcher; Glib::Dispatcher m_open_dispatcher;
Glib::Dispatcher m_close_dispatcher; Glib::Dispatcher m_close_dispatcher;
std::atomic<uint16_t> m_close_code; ix::WebSocketCloseInfo m_close_info;
std::shared_ptr<spdlog::logger> m_log;
}; };