feat(twilight): 0.16-rc.1 support

Cargo.toml

@@ -49,8 +49,9 @@ tokio-websockets = { optional = true, version = "0.5", features = ["client", "ge
 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.15.0" }
-twilight-model = { default-features = false, optional = true, version = "0.15.0" }
+twilight-gateway = { default-features = false, optional = true, version = "=0.16.0-rc.1" }
+twilight-model = { default-features = false, optional = true, version = "=0.16.0-rc.1" }
+typemap_rev = { optional = true, version = "0.3" }
 url = { optional = true, version = "2" }
 uuid = { features = ["v4"], optional = true, version = "1" }
 
@@ -125,7 +126,7 @@ native = [
     "serenity?/native_tls_backend",
     "tokio-tungstenite?/native-tls",
     "tokio-websockets?/native-tls",
-    "twilight-gateway?/native",
+    "twilight-gateway?/native-tls",
 ]
 tungstenite = ["dep:tokio-tungstenite"]
 tws = ["dep:tokio-websockets"]

examples/Cargo.toml

@@ -11,11 +11,11 @@ resolver = "2"
 [workspace.dependencies]
 reqwest = "0.11"
 serenity = { features = ["cache", "framework", "standard_framework", "voice", "http", "rustls_backend"], version = "0.12" }
-songbird = { path = "../", version = "0.4" }
+songbird = { path = "../", version = "0.4", default-features = false }
 symphonia = { features = ["aac", "mp3", "isomp4", "alac"], version = "0.5.2" }
 tokio = { features = ["macros", "rt-multi-thread", "signal", "sync"], version = "1" }
 tracing = "0.1"
-tracing-subscriber = "0.2"
+tracing-subscriber = "0.3"
 tracing-futures = "0.2"
 
 [profile.release]

examples/twilight/Cargo.toml

@@ -7,12 +7,12 @@ edition = "2021"
 [dependencies]
 futures = "0.3"
 reqwest = { workspace = true }
-songbird = { workspace = true, features = ["driver", "gateway", "twilight", "rustls", "tungstenite"] }
+songbird = { workspace = true, features = ["driver", "gateway", "twilight", "rustls", "tws"] }
 symphonia = { workspace = true }
 tracing = { workspace = true }
-tracing-subscriber = { workspace = true }
+tracing-subscriber = { workspace = true, default-features = true }
 tokio = { workspace = true }
-twilight-gateway = "0.15"
-twilight-http = "0.15"
-twilight-model = "0.15"
-twilight-standby = "0.15"
+twilight-gateway = "=0.16.0-rc.1"
+twilight-http = "=0.16.0-rc.1"
+twilight-model = "=0.16.0-rc.1"
+twilight-standby = "=0.16.0-rc.1"

examples/twilight/src/main.rs

@@ -20,7 +20,6 @@
 //!
 //! [basic lavalink bot]: https://github.com/twilight-rs/twilight/tree/main/examples/lavalink-basic-bot.rs
 
-use futures::StreamExt;
 use songbird::{
     input::{Compose, YoutubeDl},
     shards::TwilightMap,
@@ -29,12 +28,7 @@ use songbird::{
 };
 use std::{collections::HashMap, env, error::Error, future::Future, num::NonZeroU64, sync::Arc};
 use tokio::sync::RwLock;
-use twilight_gateway::{
-    stream::{self, ShardEventStream},
-    Event,
-    Intents,
-    Shard,
-};
+use twilight_gateway::{Event, EventTypeFlags, Intents, Shard, StreamExt as _};
 use twilight_http::Client as HttpClient;
 use twilight_model::{
     channel::Message,
@@ -68,7 +62,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     // Initialize the tracing subscriber.
     tracing_subscriber::fmt::init();
 
-    let (mut shards, state) = {
+    let (shards, state) = {
         let token = env::var("DISCORD_TOKEN")?;
 
         let http = HttpClient::new(token.clone());
@@ -79,7 +73,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         let config = twilight_gateway::Config::new(token.clone(), intents);
 
         let shards: Vec<Shard> =
-            stream::create_recommended(&http, config, |_, builder| builder.build())
+            twilight_gateway::create_recommended(&http, config, |_, builder| builder.build())
                 .await?
                 .collect();
 
@@ -103,51 +97,63 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         )
     };
 
-    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");
+    let mut set = tokio::task::JoinSet::new();
+    for shard in shards {
+        set.spawn(tokio::spawn(runner(shard, state.clone())));
+    }
 
-                if source.is_fatal() {
-                    break;
-                }
+    set.join_next().await;
+
+    Ok(())
+}
+
+async fn runner(mut shard: Shard, state: Arc<StateRef>) {
+    while let Some(item) = shard.next_event(EventTypeFlags::all()).await {
+        let event = match item {
+            Ok(event) => event,
+            Err(source) => {
+                tracing::warn!(?source, "error receiving event");
 
                 continue;
             },
-            None => break,
         };
 
-        state.standby.process(&event);
-        state.songbird.process(&event).await;
-
-        if let Event::MessageCreate(msg) = event {
-            if msg.guild_id.is_none() || !msg.content.starts_with('!') {
-                continue;
+        tokio::spawn({
+            let state = state.clone();
+            async move {
+                handle_event(event, state).await;
             }
+        });
+    }
+}
 
-            match msg.content.splitn(2, ' ').next() {
-                Some("!join") => spawn(join(msg.0, Arc::clone(&state))),
-                Some("!leave") => spawn(leave(msg.0, Arc::clone(&state))),
-                Some("!pause") => spawn(pause(msg.0, Arc::clone(&state))),
-                Some("!play") => spawn(play(msg.0, Arc::clone(&state))),
-                Some("!seek") => spawn(seek(msg.0, Arc::clone(&state))),
-                Some("!stop") => spawn(stop(msg.0, Arc::clone(&state))),
-                Some("!volume") => spawn(volume(msg.0, Arc::clone(&state))),
-                _ => continue,
-            }
+async fn handle_event(event: Event, state: Arc<StateRef>) {
+    state.standby.process(&event);
+    state.songbird.process(&event).await;
+
+    if let Event::MessageCreate(msg) = event {
+        if msg.guild_id.is_none() || !msg.content.starts_with('!') {
+            return;
         }
-    }
 
-    Ok(())
+        match msg.content.splitn(2, ' ').next() {
+            Some("!join") => spawn(join(msg.0, Arc::clone(&state))),
+            Some("!leave") => spawn(leave(msg.0, Arc::clone(&state))),
+            Some("!pause") => spawn(pause(msg.0, Arc::clone(&state))),
+            Some("!play") => spawn(play(msg.0, Arc::clone(&state))),
+            Some("!seek") => spawn(seek(msg.0, Arc::clone(&state))),
+            Some("!stop") => spawn(stop(msg.0, Arc::clone(&state))),
+            Some("!volume") => spawn(volume(msg.0, Arc::clone(&state))),
+            _ => return,
+        }
+    }
 }
 
 async fn join(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     state
         .http
         .create_message(msg.channel_id)
-        .content("What's the channel ID you want me to join?")?
+        .content("What's the channel ID you want me to join?")
         .await?;
 
     let author_id = msg.author.id;
@@ -171,7 +177,7 @@ async fn join(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
     state
         .http
         .create_message(msg.channel_id)
-        .content(&content)?
+        .content(&content)
         .await?;
 
     Ok(())
@@ -191,7 +197,7 @@ async fn leave(msg: Message, state: State) -> Result<(), Box<dyn Error + Send +
     state
         .http
         .create_message(msg.channel_id)
-        .content("Left the channel")?
+        .content("Left the channel")
         .await?;
 
     Ok(())
@@ -206,7 +212,7 @@ async fn play(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
     state
         .http
         .create_message(msg.channel_id)
-        .content("What's the URL of the audio to play?")?
+        .content("What's the URL of the audio to play?")
         .await?;
 
     let author_id = msg.author.id;
@@ -230,7 +236,7 @@ async fn play(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
         state
             .http
             .create_message(msg.channel_id)
-            .content(&content)?
+            .content(&content)
             .await?;
 
         if let Some(call_lock) = state.songbird.get(guild_id) {
@@ -244,7 +250,7 @@ async fn play(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
         state
             .http
             .create_message(msg.channel_id)
-            .content("Didn't find any results")?
+            .content("Didn't find any results")
             .await?;
     }
 
@@ -286,7 +292,7 @@ async fn pause(msg: Message, state: State) -> Result<(), Box<dyn Error + Send +
     state
         .http
         .create_message(msg.channel_id)
-        .content(&content)?
+        .content(&content)
         .await?;
 
     Ok(())
@@ -301,7 +307,7 @@ async fn seek(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
     state
         .http
         .create_message(msg.channel_id)
-        .content("Where in the track do you want to seek to (in seconds)?")?
+        .content("Where in the track do you want to seek to (in seconds)?")
         .await?;
 
     let author_id = msg.author.id;
@@ -326,7 +332,7 @@ async fn seek(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
     state
         .http
         .create_message(msg.channel_id)
-        .content(&content)?
+        .content(&content)
         .await?;
 
     Ok(())
@@ -349,7 +355,7 @@ async fn stop(msg: Message, state: State) -> Result<(), Box<dyn Error + Send + S
     state
         .http
         .create_message(msg.channel_id)
-        .content("Stopped the track")?
+        .content("Stopped the track")
         .await?;
 
     Ok(())
@@ -364,7 +370,7 @@ async fn volume(msg: Message, state: State) -> Result<(), Box<dyn Error + Send +
     state
         .http
         .create_message(msg.channel_id)
-        .content("What's the volume you want to set (0.0-10.0, 1.0 being the default)?")?
+        .content("What's the volume you want to set (0.0-10.0, 1.0 being the default)?")
         .await?;
 
     let author_id = msg.author.id;
@@ -381,7 +387,7 @@ async fn volume(msg: Message, state: State) -> Result<(), Box<dyn Error + Send +
         state
             .http
             .create_message(msg.channel_id)
-            .content("Invalid volume!")?
+            .content("Invalid volume!")
             .await?;
 
         return Ok(());
@@ -399,7 +405,7 @@ async fn volume(msg: Message, state: State) -> Result<(), Box<dyn Error + Send +
     state
         .http
         .create_message(msg.channel_id)
-        .content(&content)?
+        .content(&content)
         .await?;
 
     Ok(())

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::error::SendError;
+use twilight_gateway::error::ChannelError;
 
 #[cfg(feature = "gateway")]
 #[derive(Debug)]
@@ -51,7 +51,7 @@ pub enum JoinError {
     Serenity(Box<TrySendError<ShardRunnerMessage>>),
     #[cfg(feature = "twilight")]
     /// Twilight-specific WebSocket send error when a message fails to send over websocket.
-    Twilight(SendError),
+    Twilight(ChannelError),
 }
 
 #[cfg(feature = "gateway")]
@@ -124,8 +124,8 @@ impl From<Box<TrySendError<ShardRunnerMessage>>> for JoinError {
 }
 
 #[cfg(all(feature = "twilight", feature = "gateway"))]
-impl From<SendError> for JoinError {
-    fn from(e: SendError) -> Self {
+impl From<ChannelError> for JoinError {
+    fn from(e: ChannelError) -> Self {
         JoinError::Twilight(e)
     }
 }

src/shards.rs

@@ -29,7 +29,7 @@ use twilight_model::gateway::payload::outgoing::update_voice_state::UpdateVoiceS
 #[cfg(feature = "twilight")]
 #[derive(Debug)]
 pub struct TwilightMap {
-    map: std::collections::HashMap<u64, MessageSender>,
+    map: std::collections::HashMap<u32, MessageSender>,
 }
 
 #[cfg(feature = "twilight")]
@@ -38,13 +38,13 @@ impl TwilightMap {
     ///
     /// For correctness all shards should be in the map.
     #[must_use]
-    pub fn new(map: std::collections::HashMap<u64, MessageSender>) -> Self {
+    pub fn new(map: std::collections::HashMap<u32, MessageSender>) -> Self {
         TwilightMap { map }
     }
 
     /// Get the message sender for `shard_id`.
     #[must_use]
-    pub fn get(&self, shard_id: u64) -> Option<&MessageSender> {
+    pub fn get(&self, shard_id: u32) -> Option<&MessageSender> {
         self.map.get(&shard_id)
     }
 
@@ -90,7 +90,7 @@ impl Sharder {
                 s.get_or_insert_shard_handle(shard_id as u32),
             )),
             #[cfg(feature = "twilight")]
-            Sharder::Twilight(t) => Some(Shard::Twilight(t.clone(), shard_id)),
+            Sharder::Twilight(t) => Some(Shard::Twilight(t.clone(), shard_id as u32)),
             Sharder::Generic(src) => src.get_shard(shard_id).map(Shard::Generic),
         }
     }
@@ -156,7 +156,7 @@ pub enum Shard {
     Serenity(Arc<SerenityShardHandle>),
     #[cfg(feature = "twilight")]
     /// Handle to a map of twilight command senders.
-    Twilight(Arc<TwilightMap>, u64),
+    Twilight(Arc<TwilightMap>, u32),
     /// Handle to a generic shard instance.
     Generic(#[derivative(Debug = "ignore")] Arc<dyn VoiceUpdate + Send + Sync>),
 }