Attempt a full reconnection when a resume attempt failed. (#1465)

lgtm ship it!

Author: kar8uncle

include/dpp/discordclient.h

@@ -139,17 +139,33 @@ enum shard_frame_type : int {
 	ft_request_soundboard_sounds = 31,
 };
 
+/**
+ * @brief Callback to send a Gateway request to start a new voice connection.
+ */
+using voice_connection_gateway_request_callback_t = std::function<void(discord_client*)>;
+
 /**
  * @brief Represents a connection to a voice channel.
  * A client can only connect to one voice channel per guild at a time, so these are stored in a map
  * in the dpp::discord_client keyed by guild_id.
  */
-class DPP_EXPORT voiceconn {
+class DPP_EXPORT voiceconn : public std::enable_shared_from_this<voiceconn> {
 	/**
 	 * @brief Owning dpp::discord_client instance
 	 */
 	class discord_client* creator;
+
+	/**
+	 * @brief Function to ask a discord_client instance to make a request to Gateway to connect to a voice channel
+	 */
+	voice_connection_gateway_request_callback_t request_callback;
+
 public:
+	/**
+	 * @brief Guild to connect to the voice channel on
+	 */
+	snowflake guild_id;
+
 	/**
 	 * @brief Voice Channel ID
 	 */
@@ -173,28 +189,25 @@ class DPP_EXPORT voiceconn {
 	/**
 	 * @brief voice websocket client
 	 */
-	class discord_voice_client* voiceclient;
+	std::unique_ptr<class discord_voice_client> voiceclient;
 
 	/**
 	 * @brief True to enable DAVE E2EE
 	 * @warning This is an EXPERIMENTAL feature!
 	 */
 	bool dave;
 
-	/**
-	 * @brief Construct a new voiceconn object
-	 */
-	voiceconn() = default;
-
 	/**
 	 * @brief Construct a new voiceconn object
 	 * 
 	 * @param o owner
-	 * @param _channel_id voice channel id
+	 * @param request_callback Function to ask a discord_client instance to make a request to Gateway to connect to a voice channel
+	 * @param guild_id Guild to connect to the voice channel on
+	 * @param channel_id voice channel id
 	 * @param enable_dave True to enable DAVE E2EE
 	 * @warn DAVE is an EXPERIMENTAL feature!
 	 */
-	voiceconn(class discord_client* o, snowflake _channel_id, bool enable_dave);
+	voiceconn(class discord_client* o, voice_connection_gateway_request_callback_t request_callback, snowflake guild_id, snowflake channel_id, bool enable_dave);
 
 	/**
 	 * @brief Destroy the voiceconn object
@@ -216,16 +229,17 @@ class DPP_EXPORT voiceconn {
 	 */
 	bool is_active() const;
 
+	voiceconn& request();
+
 	/**
 	 * @brief Create websocket object and connect it.
 	 * Needs hostname, token and session_id to be set or does nothing.
-	 * 
-	 * @param guild_id Guild to connect to the voice channel on
+	 *
 	 * @return reference to self
 	 * @note It can spawn a thread to establish the connection, so this is NOT a synchronous blocking call!
 	 * You shouldn't call this directly. Use a wrapper function instead. e.g. dpp::guild::connect_member_voice
 	 */
-	voiceconn& connect(snowflake guild_id);
+	voiceconn& connect();
 
 	/**
 	 * @brief Disconnect from the currently connected voice channel
@@ -436,7 +450,7 @@ class DPP_EXPORT discord_client : public websocket_client
 	/**
 	 * @brief List of voice channels we are connecting to keyed by guild id
 	 */
-	std::unordered_map<snowflake, std::unique_ptr<voiceconn>> connecting_voice_channels;
+	std::unordered_map<snowflake, std::shared_ptr<voiceconn>> connecting_voice_channels;
 
 	/**
 	 * @brief The gateway address we reconnect to when we resume a session

include/dpp/discordvoiceclient.h

@@ -230,6 +230,11 @@ struct dave_binary_header_t {
  */
 using privacy_code_callback_t = std::function<void(const std::string&)>;
 
+/**
+ * @brief A callback for a full reconnection after the voice client disconnects due to error.
+ */
+using full_reconnection_callback_t = std::function<void()>;
+
 /** @brief Implements a discord voice connection.
  * Each discord_voice_client connects to one voice channel and derives from a websocket client.
  */
@@ -709,6 +714,8 @@ class DPP_EXPORT discord_voice_client : public websocket_client
 	 */
 	class dpp::cluster* creator{};
 
+	full_reconnection_callback_t reconnection_callback;
+
 	/**
 	 * @brief True when the thread is shutting down
 	 */
@@ -872,6 +879,7 @@ class DPP_EXPORT discord_voice_client : public websocket_client
 
 	/** Constructor takes shard id, max shards and token.
 	 * @param _cluster The cluster which owns this voice connection, for related logging, REST requests etc
+	 * @param _reconnection_callback The callback this voice client calls if the voice connection was closed due to error
 	 * @param _channel_id The channel id to identify the voice connection as
 	 * @param _server_id The server id (guild id) to identify the voice connection as
 	 * @param _token The voice session token to use for identifying to the websocket
@@ -881,7 +889,7 @@ class DPP_EXPORT discord_voice_client : public websocket_client
 	 * @throw dpp::voice_exception Opus failed to initialise, or D++ is not compiled with voice support
 	 * @warning DAVE E2EE is an EXPERIMENTAL feature!
 	 */
-	discord_voice_client(dpp::cluster* _cluster, snowflake _channel_id, snowflake _server_id, const std::string &_token, const std::string &_session_id, const std::string &_host, bool enable_dave = false);
+	discord_voice_client(dpp::cluster* _cluster, full_reconnection_callback_t _reconnection_callback, snowflake _channel_id, snowflake _server_id, const std::string &_token, const std::string &_session_id, const std::string &_host, bool enable_dave = false);
 
 	/**
 	 * @brief Destroy the discord voice client object

src/dpp/discordclient.cpp

@@ -514,21 +514,24 @@ discord_client& discord_client::connect_voice(snowflake guild_id, snowflake chan
 			return *this;
 		}
 	}
-	connecting_voice_channels[guild_id] = std::make_unique<voiceconn>(this, channel_id, enable_dave);
-	/* Once sent, this expects two events (in any order) on the websocket:
-	* VOICE_SERVER_UPDATE and VOICE_STATUS_UPDATE
-	*/
-	log(ll_debug, "Sending op 4 to join VC, guild " + std::to_string(guild_id) + " channel " + std::to_string(channel_id) + (enable_dave ? " WITH DAVE" : ""));
-	queue_message(jsonobj_to_string(json({
-		{ "op", ft_voice_state_update },
-		{ "d", {
-				{ "guild_id", std::to_string(guild_id) },
-				{ "channel_id", std::to_string(channel_id) },
-				{ "self_mute", self_mute },
-				{ "self_deaf", self_deaf },
+	connecting_voice_channels[guild_id] = std::make_shared<voiceconn>(this, [guild_id, channel_id, self_mute, self_deaf, enable_dave](discord_client* client) {
+		/* Once sent, this expects two events (in any order) on the websocket:
+		 * VOICE_SERVER_UPDATE and VOICE_STATUS_UPDATE
+		 */
+		// Always honor the given client rather than this because
+		// the owner can change after shard reconnection.
+		client->log(ll_debug, "Sending op 4 to join VC, guild " + std::to_string(guild_id) + " channel " + std::to_string(channel_id) + (enable_dave ? " WITH DAVE" : ""));
+		client->queue_message(client->jsonobj_to_string(json({
+			{ "op", ft_voice_state_update },
+			{ "d", {
+					{ "guild_id", std::to_string(guild_id) },
+					{ "channel_id", std::to_string(channel_id) },
+					{ "self_mute", self_mute },
+					{ "self_deaf", self_deaf },
+				}
 			}
-		}
-	})), false);
+	})), false);}, guild_id, channel_id, enable_dave);
+
 #endif
 	return *this;
 }
@@ -581,7 +584,8 @@ voiceconn* discord_client::get_voice(snowflake guild_id) {
 }
 
 
-voiceconn::voiceconn(discord_client* o, snowflake _channel_id, bool enable_dave) : creator(o), channel_id(_channel_id), voiceclient(nullptr), dave(enable_dave) {
+voiceconn::voiceconn(discord_client* o, voice_connection_gateway_request_callback_t request_callback, snowflake guild_id, snowflake channel_id, bool enable_dave) : creator(o), request_callback(std::move(request_callback)), guild_id(guild_id), channel_id(channel_id), voiceclient(nullptr), dave(enable_dave) {
+	request();
 }
 
 bool voiceconn::is_ready() const {
@@ -594,8 +598,7 @@ bool voiceconn::is_active() const {
 
 voiceconn& voiceconn::disconnect() {
 	if (this->is_active()) {
-		delete voiceclient;
-		voiceclient = nullptr;
+		voiceclient.reset();
 	}
 	return *this;
 }
@@ -604,16 +607,31 @@ voiceconn::~voiceconn() {
 	this->disconnect();
 }
 
-voiceconn& voiceconn::connect(snowflake guild_id) {
+voiceconn& voiceconn::request() {
+	this->token.clear();
+	this->session_id.clear();
+	this->websocket_hostname.clear();
+	this->voiceclient.reset();
+
+	this->request_callback(this->creator);
+	return *this;
+}
+
+voiceconn& voiceconn::connect() {
 	if (this->is_ready() && !this->is_active()) {
 		try {
-			this->creator->log(ll_debug, "Connecting voice for guild " + std::to_string(guild_id) + " channel " + std::to_string(this->channel_id));
-			this->voiceclient = new discord_voice_client(creator->creator, this->channel_id, guild_id, this->token, this->session_id, this->websocket_hostname, this->dave);
+			this->creator->log(ll_debug, "Connecting voice for guild " + std::to_string(this->guild_id) + " channel " + std::to_string(this->channel_id));
+			full_reconnection_callback_t reconnection_callback = [weak_this=weak_from_this()] {
+			if (std::shared_ptr<voiceconn> strong_this = weak_this.lock()) {
+				strong_this->request();
+			}
+		};
+		this->voiceclient = std::make_unique<discord_voice_client>(creator->creator, std::move(reconnection_callback), this->channel_id, this->guild_id, this->token, this->session_id, this->websocket_hostname, this->dave);
 			/* Note: Spawns thread! */
 			this->voiceclient->run();
 		}
 		catch (std::exception &e) {
-			this->creator->log(ll_debug, "Can't connect to voice websocket (guild_id: " + std::to_string(guild_id) + ", channel_id: " + std::to_string(this->channel_id) + "): " + std::string(e.what()));
+			this->creator->log(ll_debug, "Can't connect to voice websocket (guild_id: " + std::to_string(this->guild_id) + ", channel_id: " + std::to_string(this->channel_id) + "): " + std::string(e.what()));
 		}
 	}
 	return *this;

src/dpp/events/voice_server_update.cpp

@@ -49,11 +49,12 @@ void voice_server_update::handle(discord_client* client, json &j, const std::str
 		auto v = client->connecting_voice_channels.find(vsu.guild_id);
 		/* Check to see if there is a connection in progress for a voice channel on this guild */
 		if (v != client->connecting_voice_channels.end()) {
-			if (!v->second->is_ready()) {
-				v->second->token = vsu.token;
-				v->second->websocket_hostname = vsu.endpoint;
-				if (!v->second->is_active()) {
-					v->second->connect(vsu.guild_id);
+			voiceconn& connection = *v->second;
+			if (!connection.is_ready()) {
+				connection.token = vsu.token;
+				connection.websocket_hostname = vsu.endpoint;
+				if (!connection.is_active()) {
+					connection.connect();
 				}
 			}
 		}
@@ -66,4 +67,4 @@ void voice_server_update::handle(discord_client* client, json &j, const std::str
 	}
 }
 
-};
+};

src/dpp/events/voice_state_update.cpp

@@ -75,9 +75,10 @@ void voice_state_update::handle(discord_client* client, json &j, const std::stri
 			auto v = client->connecting_voice_channels.find(vsu.state.guild_id);
 			/* Check to see if we have a connection to a voice channel in progress on this guild */
 			if (v != client->connecting_voice_channels.end()) {
-				v->second->session_id = vsu.state.session_id;
-				if (v->second->is_ready() && !v->second->is_active()) {
-					v->second->connect(vsu.state.guild_id);
+				voiceconn& connection = *v->second;
+				connection.session_id = vsu.state.session_id;
+				if (connection.is_ready() && !connection.is_active()) {
+					connection.connect();
 				}
 			}
 		}

src/dpp/voice/enabled/constructor.cpp

@@ -32,7 +32,7 @@
 
 namespace dpp {
 
-discord_voice_client::discord_voice_client(dpp::cluster* _cluster, snowflake _channel_id, snowflake _server_id, const std::string &_token, const std::string &_session_id, const std::string &_host, bool enable_dave)
+discord_voice_client::discord_voice_client(dpp::cluster* _cluster, full_reconnection_callback_t _reconnection_callback, snowflake _channel_id, snowflake _server_id, const std::string &_token, const std::string &_session_id, const std::string &_host, bool enable_dave)
 	: websocket_client(_cluster, _host.substr(0, _host.find(':')), _host.substr(_host.find(':') + 1, _host.length()), "/?v=" + std::to_string(voice_protocol_version), OP_TEXT),
 	connect_time(0),
 	mixer(std::make_unique<audio_mixer>()),
@@ -54,6 +54,7 @@ discord_voice_client::discord_voice_client(dpp::cluster* _cluster, snowflake _ch
 	tracks(0),
 	dave_version(enable_dave ? dave_version_1 : dave_version_none),
 	creator(_cluster),
+	reconnection_callback(std::move(_reconnection_callback)),
 	terminating(false),
 	heartbeat_interval(0),
 	last_heartbeat(time(nullptr)),

src/dpp/voice/enabled/handle_frame.cpp

@@ -343,9 +343,6 @@ bool discord_voice_client::handle_frame(const std::string &data, ws_opcode opcod
 					this->heartbeat_interval = j["d"]["heartbeat_interval"].get<uint32_t>();
 				}
 
-				/* Reset receive_sequence on HELLO */
-				receive_sequence = -1;
-
 				if (!modes.empty()) {
 					log(dpp::ll_debug, "Resuming voice session " + this->sessionid + "...");
 					json obj = {
@@ -362,6 +359,8 @@ bool discord_voice_client::handle_frame(const std::string &data, ws_opcode opcod
 					};
 					this->write(obj.dump(-1, ' ', false, json::error_handler_t::replace), OP_TEXT);
 				} else {
+					/* Reset receive_sequence on HELLO */
+					receive_sequence = -1;
 					log(dpp::ll_debug, "Connecting new voice session (DAVE: " + std::string(dave_version == dave_version_1 ? "Enabled" : "Disabled") + ")...");
 					json obj = {
 						{ "op", voice_opcode_connection_identify },

src/dpp/voice/enabled/thread.cpp

@@ -45,7 +45,8 @@ void discord_voice_client::on_disconnect() {
 
 	/* If we've looped 5 or more times, abort the loop. */
 	if (terminating || times_looped >= 5) {
-		log(dpp::ll_warning, "Reached max loops whilst attempting to read from the websocket. Aborting websocket.");
+		log(dpp::ll_warning, "Reached max loops whilst attempting to read from the websocket. Starting a full reconnection.");
+		owner->queue_work(1, reconnection_callback);
 		return;
 	}
 	last_loop_time = current_time;
@@ -72,4 +73,4 @@ void discord_voice_client::run() {
 	ssl_connection::read_loop();
 }
 
-}
+}

src/dpp/voice/stub/stubs.cpp

@@ -27,7 +27,7 @@
 
 namespace dpp {
 
-	discord_voice_client::discord_voice_client(dpp::cluster* _cluster, snowflake _channel_id, snowflake _server_id, const std::string &_token, const std::string &_session_id, const std::string &_host, bool enable_dave)
+	discord_voice_client::discord_voice_client(dpp::cluster* _cluster, full_reconnection_callback_t _reconnect_callback, snowflake _channel_id, snowflake _server_id, const std::string &_token, const std::string &_session_id, const std::string &_host, bool enable_dave)
 		: websocket_client(_cluster, _host.substr(0, _host.find(':')), _host.substr(_host.find(':') + 1, _host.length()), "/?v=" + std::to_string(voice_protocol_version), OP_TEXT)
 	{
 		throw dpp::voice_exception(err_no_voice_support, "Voice support not enabled in this build of D++");