Gateway: Twilight 0.15 support (#171)

This patch changes around quite a few things.
The main entrypoint for twilight besides process will now be the
TwilightMap which concists of command senders for each shard.

This simplifies parts of the code as there is not any difference
between shards and clusters anymore.

Cargo.toml

@@ -44,8 +44,8 @@ tokio-tungstenite = { optional = true, version = "0.18" }
 tokio-util = { features = ["io"], optional = true, version = "0.7" }
 tracing = { version = "0.1", features = ["log"] }
 tracing-futures = "0.2"
-twilight-gateway = { default-features = false, optional = true, version = "0.14.0" }
-twilight-model = { default-features = false, optional = true, version = "0.14.0" }
+twilight-gateway = { default-features = false, optional = true, version = "0.15.0" }
+twilight-model = { default-features = false, optional = true, version = "0.15.0" }
 typemap_rev = { optional = true, version = "0.3" }
 url = { optional = true, version = "2" }
 uuid = { features = ["v4"], optional = true, version = "1" }

examples/twilight/Cargo.toml

@@ -11,10 +11,10 @@ symphonia = { features = ["aac", "mp3", "isomp4", "alac"], version = "0.5.2" }
 tracing = "0.1"
 tracing-subscriber = "0.2"
 tokio = { features = ["macros", "rt-multi-thread", "sync"], version = "1" }
-twilight-gateway = "0.14"
-twilight-http = "0.14"
-twilight-model = "0.14"
-twilight-standby = "0.14"
+twilight-gateway = "0.15"
+twilight-http = "0.15"
+twilight-model = "0.15"
+twilight-standby = "0.15"
 
 [dependencies.songbird]
 default-features = false

examples/twilight/src/main.rs

@@ -23,12 +23,18 @@
 use futures::StreamExt;
 use songbird::{
     input::{Compose, YoutubeDl},
+    shards::TwilightMap,
     tracks::{PlayMode, TrackHandle},
     Songbird,
 };
 use std::{collections::HashMap, env, error::Error, future::Future, num::NonZeroU64, sync::Arc};
 use tokio::sync::RwLock;
-use twilight_gateway::{Cluster, Event, Intents};
+use twilight_gateway::{
+    stream::{self, ShardEventStream},
+    Event,
+    Intents,
+    Shard,
+};
 use twilight_http::Client as HttpClient;
 use twilight_model::{
     channel::Message,
@@ -62,21 +68,32 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Initialize the tracing subscriber.
     tracing_subscriber::fmt::init();
 
-    let (mut events, state) = {
+    let (mut shards, state) = {
         let token = env::var("DISCORD_TOKEN")?;
 
         let http = HttpClient::new(token.clone());
         let user_id = http.current_user().await?.model().await?.id;
 
         let intents =
-            Intents::GUILD_MESSAGES | Intents::MESSAGE_CONTENT | Intents::GUILD_VOICE_STATES;
-        let (cluster, events) = Cluster::new(token, intents).await?;
-        cluster.up().await;
+            Intents::GUILD_MESSAGES | Intents::GUILD_VOICE_STATES | Intents::MESSAGE_CONTENT;
+        let config = twilight_gateway::Config::new(token.clone(), intents);
+
+        let shards: Vec<Shard> =
+            stream::create_recommended(&http, config, |_, builder| builder.build())
+                .await?
+                .collect();
+
+        let senders = TwilightMap::new(
+            shards
+                .iter()
+                .map(|s| (s.id().number(), s.sender()))
+                .collect(),
+        );
 
-        let songbird = Songbird::twilight(Arc::new(cluster), user_id);
+        let songbird = Songbird::twilight(Arc::new(senders), user_id);
 
         (
-            events,
+            shards,
             Arc::new(StateRef {
                 http,
                 trackdata: Default::default(),
@@ -86,7 +103,22 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         )
     };
 
-    while let Some((_, event)) = events.next().await {
+    let mut stream = ShardEventStream::new(shards.iter_mut());
+    loop {
+        let event = match stream.next().await {
+            Some((_, Ok(event))) => event,
+            Some((_, Err(source))) => {
+                tracing::warn!(?source, "error receiving event");
+
+                if source.is_fatal() {
+                    break;
+                }
+
+                continue;
+            },
+            None => break,
+        };
+
         state.standby.process(&event);
         state.songbird.process(&event).await;
 

src/driver/mod.rs

@@ -30,6 +30,8 @@ pub use mix_mode::MixMode;
 #[cfg(test)]
 pub use test_config::*;
 
+#[cfg(feature = "builtin-queue")]
+use crate::tracks;
 #[cfg(feature = "builtin-queue")]
 use crate::tracks::TrackQueue;
 use crate::{

src/error.rs

@@ -11,7 +11,7 @@ pub use simd_json::Error as JsonError;
 #[cfg(feature = "gateway")]
 use std::{error::Error, fmt};
 #[cfg(feature = "twilight")]
-use twilight_gateway::{cluster::ClusterCommandError, shard::CommandError};
+use twilight_gateway::error::SendError;
 
 #[cfg(feature = "gateway")]
 #[derive(Debug)]
@@ -50,11 +50,8 @@ pub enum JoinError {
     /// Serenity-specific WebSocket send error.
     Serenity(TrySendError<InterMessage>),
     #[cfg(feature = "twilight")]
-    /// Twilight-specific WebSocket send error returned when using a shard cluster.
-    TwilightCluster(ClusterCommandError),
-    #[cfg(feature = "twilight")]
-    /// Twilight-specific WebSocket send error when explicitly using a single shard.
-    TwilightShard(CommandError),
+    /// Twilight-specific WebSocket send error when a message fails to send over websocket.
+    Twilight(SendError),
 }
 
 #[cfg(feature = "gateway")]
@@ -96,9 +93,7 @@ impl fmt::Display for JoinError {
             #[cfg(feature = "serenity")]
             JoinError::Serenity(e) => e.fmt(f),
             #[cfg(feature = "twilight")]
-            JoinError::TwilightCluster(e) => e.fmt(f),
-            #[cfg(feature = "twilight")]
-            JoinError::TwilightShard(e) => e.fmt(f),
+            JoinError::Twilight(e) => e.fmt(f),
         }
     }
 }
@@ -116,9 +111,7 @@ impl Error for JoinError {
             #[cfg(feature = "serenity")]
             JoinError::Serenity(e) => e.source(),
             #[cfg(feature = "twilight")]
-            JoinError::TwilightCluster(e) => e.source(),
-            #[cfg(feature = "twilight")]
-            JoinError::TwilightShard(e) => e.source(),
+            JoinError::Twilight(e) => e.source(),
         }
     }
 }
@@ -131,16 +124,9 @@ impl From<TrySendError<InterMessage>> for JoinError {
 }
 
 #[cfg(all(feature = "twilight", feature = "gateway"))]
-impl From<CommandError> for JoinError {
-    fn from(e: CommandError) -> Self {
-        JoinError::TwilightShard(e)
-    }
-}
-
-#[cfg(all(feature = "twilight", feature = "gateway"))]
-impl From<ClusterCommandError> for JoinError {
-    fn from(e: ClusterCommandError) -> Self {
-        JoinError::TwilightCluster(e)
+impl From<SendError> for JoinError {
+    fn from(e: SendError) -> Self {
+        JoinError::Twilight(e)
     }
 }
 

src/manager.rs

@@ -26,10 +26,9 @@ use serenity::{
 };
 use std::sync::Arc;
 use tokio::sync::Mutex;
+#[cfg(feature = "serenity")]
 use tracing::debug;
 #[cfg(feature = "twilight")]
-use twilight_gateway::Cluster;
-#[cfg(feature = "twilight")]
 use twilight_model::gateway::event::Event as TwilightEvent;
 
 #[derive(Clone, Copy, Debug)]
@@ -88,7 +87,7 @@ impl Songbird {
     /// [`process`].
     ///
     /// [`process`]: Songbird::process
-    pub fn twilight<U>(cluster: Arc<Cluster>, user_id: U) -> Self
+    pub fn twilight<U>(cluster: Arc<crate::shards::TwilightMap>, user_id: U) -> Self
     where
         U: Into<UserId>,
     {
@@ -103,17 +102,21 @@ impl Songbird {
     /// [`process`].
     ///
     /// [`process`]: Songbird::process
-    pub fn twilight_from_config<U>(cluster: Arc<Cluster>, user_id: U, config: Config) -> Self
+    pub fn twilight_from_config<U>(
+        sender_map: Arc<crate::shards::TwilightMap>,
+        user_id: U,
+        config: Config,
+    ) -> Self
     where
         U: Into<UserId>,
     {
         Self {
             client_data: OnceCell::with_value(ClientData {
-                shard_count: cluster.config().shard_scheme().total(),
+                shard_count: sender_map.shard_count(),
                 user_id: user_id.into(),
             }),
             calls: DashMap::new(),
-            sharder: Sharder::TwilightCluster(cluster),
+            sharder: Sharder::Twilight(sender_map),
             config: config.initialise_disposer().into(),
         }
     }
@@ -370,8 +373,8 @@ impl Songbird {
     pub async fn process(&self, event: &TwilightEvent) {
         match event {
             TwilightEvent::VoiceServerUpdate(v) => {
-                let id = GuildId::from(v.guild_id);
-                let call = self.get(id);
+                let guild_id = GuildId::from(v.guild_id);
+                let call = self.get(guild_id);
 
                 if let Some(call) = call {
                     let mut handler = call.lock().await;

src/shards.rs

@@ -9,18 +9,49 @@ use derivative::Derivative;
 use futures::channel::mpsc::{TrySendError, UnboundedSender as Sender};
 #[cfg(feature = "serenity")]
 use parking_lot::{lock_api::RwLockWriteGuard, Mutex as PMutex, RwLock as PRwLock};
+#[cfg(feature = "serenity")]
 use serde_json::json;
 #[cfg(feature = "serenity")]
 use serenity::gateway::InterMessage;
 #[cfg(feature = "serenity")]
 use std::result::Result as StdResult;
 use std::sync::Arc;
+#[cfg(feature = "serenity")]
 use tracing::{debug, error};
 #[cfg(feature = "twilight")]
-use twilight_gateway::{Cluster, Shard as TwilightShard};
+use twilight_gateway::MessageSender;
 #[cfg(feature = "twilight")]
 use twilight_model::gateway::payload::outgoing::update_voice_state::UpdateVoiceState as TwilightVoiceState;
 
+/// Map containing [`MessageSender`]s for Twilight.
+///
+/// [`MessageSender`]: twilight_gateway::MessageSender
+#[cfg(feature = "twilight")]
+#[derive(Debug)]
+pub struct TwilightMap {
+    map: std::collections::HashMap<u64, MessageSender>,
+}
+
+#[cfg(feature = "twilight")]
+impl TwilightMap {
+    /// Construct a map of shards and command senders to those shards.
+    ///
+    /// For correctness all shards should be in the map.
+    pub fn new(map: std::collections::HashMap<u64, MessageSender>) -> Self {
+        TwilightMap { map }
+    }
+
+    /// Get the message sender for `shard_id`.
+    pub fn get(&self, shard_id: u64) -> Option<&MessageSender> {
+        self.map.get(&shard_id)
+    }
+
+    /// Get the total number of shards in the map.
+    pub fn shard_count(&self) -> u64 {
+        self.map.len() as u64
+    }
+}
+
 #[derive(Derivative)]
 #[derivative(Debug)]
 #[non_exhaustive]
@@ -30,11 +61,8 @@ pub enum Sharder {
     /// Serenity-specific wrapper for sharder state initialised by the library.
     Serenity(SerenitySharder),
     #[cfg(feature = "twilight")]
-    /// Twilight-specific wrapper for sharder state initialised by the user.
-    TwilightCluster(Arc<Cluster>),
-    #[cfg(feature = "twilight")]
-    /// Twilight-specific wrapper for a single shard initialised by the user.
-    TwilightShard(Arc<TwilightShard>),
+    /// Twilight-specific wrapper for a map of command senders.
+    Twilight(Arc<TwilightMap>),
     /// A generic shard handle source.
     Generic(#[derivative(Debug = "ignore")] Arc<dyn GenericSharder + Send + Sync>),
 }
@@ -59,9 +87,7 @@ impl Sharder {
                 s.get_or_insert_shard_handle(shard_id as u32),
             )),
             #[cfg(feature = "twilight")]
-            Sharder::TwilightCluster(t) => Some(Shard::TwilightCluster(t.clone(), shard_id)),
-            #[cfg(feature = "twilight")]
-            Sharder::TwilightShard(t) => Some(Shard::TwilightShard(t.clone())),
+            Sharder::Twilight(t) => Some(Shard::Twilight(t.clone(), shard_id)),
             Sharder::Generic(src) => src.get_shard(shard_id).map(Shard::Generic),
         }
     }
@@ -126,11 +152,8 @@ pub enum Shard {
     /// Handle to one of serenity's shard runners.
     Serenity(Arc<SerenityShardHandle>),
     #[cfg(feature = "twilight")]
-    /// Handle to a twilight shard spawned from a cluster.
-    TwilightCluster(Arc<Cluster>, u64),
-    #[cfg(feature = "twilight")]
-    /// Handle to a twilight shard spawned from a cluster.
-    TwilightShard(Arc<TwilightShard>),
+    /// Handle to a map of twilight command senders.
+    Twilight(Arc<TwilightMap>, u64),
     /// Handle to a generic shard instance.
     Generic(#[derivative(Debug = "ignore")] Arc<dyn VoiceUpdate + Send + Sync>),
 }
@@ -161,17 +184,13 @@ impl VoiceUpdate for Shard {
                 Ok(())
             },
             #[cfg(feature = "twilight")]
-            Shard::TwilightCluster(handle, shard_id) => {
-                let channel_id = channel_id.map(|c| c.0).map(From::from);
-                let cmd = TwilightVoiceState::new(guild_id.0, channel_id, self_deaf, self_mute);
-                handle.command(*shard_id, &cmd).await?;
-                Ok(())
-            },
-            #[cfg(feature = "twilight")]
-            Shard::TwilightShard(handle) => {
+            Shard::Twilight(map, shard_id) => {
                 let channel_id = channel_id.map(|c| c.0).map(From::from);
                 let cmd = TwilightVoiceState::new(guild_id.0, channel_id, self_deaf, self_mute);
-                handle.command(&cmd).await?;
+                let sender = map
+                    .get(*shard_id)
+                    .ok_or(crate::error::JoinError::NoSender)?;
+                sender.command(&cmd)?;
                 Ok(())
             },
             Shard::Generic(g) =>