diff --git a/src/gateway/backend_tungstenite.rs b/src/gateway/backend_tungstenite.rs new file mode 100644 index 0000000..53b6982 --- /dev/null +++ b/src/gateway/backend_tungstenite.rs @@ -0,0 +1,65 @@ +use futures_util::{ + stream::{SplitSink, SplitStream}, + StreamExt, +}; +use tokio::net::TcpStream; +use tokio_tungstenite::{ + connect_async_tls_with_config, tungstenite, Connector, MaybeTlsStream, WebSocketStream, +}; + +use super::GatewayMessage; +use crate::errors::GatewayError; + +#[derive(Debug, Clone)] +pub struct WebSocketBackend; + +// These could be made into inherent associated types when that's stabilized +pub type WsSink = SplitSink>, tungstenite::Message>; +pub type WsStream = SplitStream>>; + +impl WebSocketBackend { + pub async fn connect( + websocket_url: &str, + ) -> Result<(WsSink, WsStream), crate::errors::GatewayError> { + let mut roots = rustls::RootCertStore::empty(); + for cert in rustls_native_certs::load_native_certs().expect("could not load platform certs") + { + roots.add(&rustls::Certificate(cert.0)).unwrap(); + } + let (websocket_stream, _) = match connect_async_tls_with_config( + websocket_url, + None, + false, + Some(Connector::Rustls( + rustls::ClientConfig::builder() + .with_safe_defaults() + .with_root_certificates(roots) + .with_no_client_auth() + .into(), + )), + ) + .await + { + Ok(websocket_stream) => websocket_stream, + Err(e) => { + return Err(GatewayError::CannotConnect { + error: e.to_string(), + }) + } + }; + + Ok(websocket_stream.split()) + } +} + +impl From for tungstenite::Message { + fn from(message: GatewayMessage) -> Self { + Self::Text(message.0) + } +} + +impl From for GatewayMessage { + fn from(value: tungstenite::Message) -> Self { + Self(value.to_string()) + } +} diff --git a/src/gateway/default/gateway.rs b/src/gateway/default/gateway.rs deleted file mode 100644 index 73178ba..0000000 --- a/src/gateway/default/gateway.rs +++ /dev/null @@ -1,189 +0,0 @@ -use futures_util::StreamExt; -use tokio_tungstenite::tungstenite::Message; - -use super::events::Events; -use super::*; -use crate::types::{self, WebSocketEvent}; - -#[derive(Debug)] -pub struct DefaultGateway { - events: Arc>, - heartbeat_handler: HeartbeatHandler>>, - websocket_send: Arc< - Mutex< - SplitSink< - WebSocketStream>, - tokio_tungstenite::tungstenite::Message, - >, - >, - >, - websocket_receive: SplitStream>>, - kill_send: tokio::sync::broadcast::Sender<()>, - store: GatewayStore, - url: String, -} - -#[async_trait] -impl - GatewayCapable< - tokio_tungstenite::tungstenite::Message, - WebSocketStream>, - > for DefaultGateway -{ - fn get_heartbeat_handler( - &self, - ) -> &HeartbeatHandler>> { - &self.heartbeat_handler - } - - async fn spawn>>>( - websocket_url: String, - ) -> Result { - let mut roots = rustls::RootCertStore::empty(); - for cert in rustls_native_certs::load_native_certs().expect("could not load platform certs") - { - roots.add(&rustls::Certificate(cert.0)).unwrap(); - } - let (websocket_stream, _) = match connect_async_tls_with_config( - &websocket_url, - None, - false, - Some(Connector::Rustls( - rustls::ClientConfig::builder() - .with_safe_defaults() - .with_root_certificates(roots) - .with_no_client_auth() - .into(), - )), - ) - .await - { - Ok(websocket_stream) => websocket_stream, - Err(e) => { - return Err(GatewayError::CannotConnect { - error: e.to_string(), - }) - } - }; - - let (websocket_send, mut websocket_receive) = websocket_stream.split(); - - let shared_websocket_send = Arc::new(Mutex::new(websocket_send)); - - // Create a shared broadcast channel for killing all gateway tasks - let (kill_send, mut _kill_receive) = tokio::sync::broadcast::channel::<()>(16); - - // Wait for the first hello and then spawn both tasks so we avoid nested tasks - // This automatically spawns the heartbeat task, but from the main thread - // TODO: We likely should not unwrap here. - let msg = websocket_receive.next().await.unwrap().unwrap(); - let gateway_payload: types::GatewayReceivePayload = - serde_json::from_str(msg.to_text().unwrap()).unwrap(); - - if gateway_payload.op_code != GATEWAY_HELLO { - return Err(GatewayError::NonHelloOnInitiate { - opcode: gateway_payload.op_code, - }); - } - - info!("GW: Received Hello"); - - let gateway_hello: types::HelloData = - serde_json::from_str(gateway_payload.event_data.unwrap().get()).unwrap(); - - let events = Events::default(); - let shared_events: Arc> = Arc::new(Mutex::new(events)); - - let store = Arc::new(Mutex::new(HashMap::new())); - - let mut gateway = DefaultGateway { - events: shared_events.clone(), - heartbeat_handler: HeartbeatHandler::new( - Duration::from_millis(gateway_hello.heartbeat_interval), - shared_websocket_send.clone(), - kill_send.subscribe(), - ), - websocket_send: shared_websocket_send.clone(), - websocket_receive, - kill_send: kill_send.clone(), - store: store.clone(), - url: websocket_url.clone(), - }; - - // Now we can continuously check for messages in a different task, since we aren't going to receive another hello - task::spawn(async move { - gateway.gateway_listen_task().await; - }); - - Ok(G::new( - websocket_url.clone(), - shared_events, - shared_websocket_send.clone(), - kill_send.clone(), - store, - )) - } - - /// Closes the websocket connection and stops all tasks - async fn close(&mut self) { - self.kill_send.send(()).unwrap(); - self.websocket_send.lock().await.close().await.unwrap(); - } - - fn get_events(&self) -> Arc> { - self.events.clone() - } - - fn get_websocket_send( - &self, - ) -> Arc>, Message>>> { - self.websocket_send.clone() - } - - fn get_store(&self) -> GatewayStore { - self.store.clone() - } - - fn get_url(&self) -> String { - self.url.clone() - } -} - -impl DefaultGateway { - /// The main gateway listener task; - /// - /// Can only be stopped by closing the websocket, cannot be made to listen for kill - pub async fn gateway_listen_task(&mut self) { - loop { - let msg = self.websocket_receive.next().await; - - // This if chain can be much better but if let is unstable on stable rust - if let Some(Ok(message)) = msg { - let _ = self - .handle_message(GatewayMessage::from_tungstenite_message(message)) - .await; - continue; - } - - // We couldn't receive the next message or it was an error, something is wrong with the websocket, close - warn!("GW: Websocket is broken, stopping gateway"); - break; - } - } - - /// Deserializes and updates a dispatched event, when we already know its type; - /// (Called for every event in handle_message) - async fn handle_event<'a, T: WebSocketEvent + serde::Deserialize<'a>>( - data: &'a str, - event: &mut GatewayEvent, - ) -> Result<(), serde_json::Error> { - let data_deserialize_result: Result = serde_json::from_str(data); - - if data_deserialize_result.is_err() { - return Err(data_deserialize_result.err().unwrap()); - } - - event.notify(data_deserialize_result.unwrap()).await; - Ok(()) - } -} diff --git a/src/gateway/default/handle.rs b/src/gateway/default/handle.rs deleted file mode 100644 index dc3fd4c..0000000 --- a/src/gateway/default/handle.rs +++ /dev/null @@ -1,131 +0,0 @@ -use super::{events::Events, *}; -use crate::types::{self, Composite}; - -#[async_trait(?Send)] -impl - GatewayHandleCapable< - tokio_tungstenite::tungstenite::Message, - WebSocketStream>, - > for DefaultGatewayHandle -{ - async fn send_json_event(&self, op_code: u8, to_send: serde_json::Value) { - self.send_json_event(op_code, to_send).await - } - - async fn observe>( - &self, - object: Arc>, - ) -> Arc> { - self.observe(object).await - } - - async fn close(&self) { - self.kill_send.send(()).unwrap(); - self.websocket_send.lock().await.close().await.unwrap(); - } - - fn new( - url: String, - events: Arc>, - websocket_send: Arc< - Mutex< - SplitSink< - WebSocketStream>, - tokio_tungstenite::tungstenite::Message, - >, - >, - >, - kill_send: tokio::sync::broadcast::Sender<()>, - store: GatewayStore, - ) -> Self { - Self { - url, - events, - websocket_send, - kill_send, - store, - } - } -} - -/// Represents a handle to a Gateway connection. A Gateway connection will create observable -/// [`GatewayEvents`](GatewayEvent), which you can subscribe to. Gateway events include all currently -/// implemented types with the trait [`WebSocketEvent`] -/// Using this handle you can also send Gateway Events directly. -#[derive(Debug, Clone)] -pub struct DefaultGatewayHandle { - pub url: String, - pub events: Arc>, - pub websocket_send: Arc< - Mutex< - SplitSink< - WebSocketStream>, - tokio_tungstenite::tungstenite::Message, - >, - >, - >, - /// Tells gateway tasks to close - pub(super) kill_send: tokio::sync::broadcast::Sender<()>, - pub(crate) store: GatewayStore, -} - -impl DefaultGatewayHandle { - async fn send_json_event(&self, op_code: u8, to_send: serde_json::Value) { - let gateway_payload = types::GatewaySendPayload { - op_code, - event_data: Some(to_send), - sequence_number: None, - }; - - let payload_json = serde_json::to_string(&gateway_payload).unwrap(); - - let message = tokio_tungstenite::tungstenite::Message::text(payload_json); - - self.websocket_send - .lock() - .await - .send(message) - .await - .unwrap(); - } - - pub async fn observe>( - &self, - object: Arc>, - ) -> Arc> { - let mut store = self.store.lock().await; - let id = object.read().unwrap().id(); - if let Some(channel) = store.get(&id) { - let object = channel.clone(); - drop(store); - object - .read() - .unwrap() - .downcast_ref::() - .unwrap_or_else(|| { - panic!( - "Snowflake {} already exists in the store, but it is not of type T.", - id - ) - }); - let ptr = Arc::into_raw(object.clone()); - // SAFETY: - // - We have just checked that the typeid of the `dyn Any ...` matches that of `T`. - // - This operation doesn't read or write any shared data, and thus cannot cause a data race - // - The reference count is not being modified - let downcasted = unsafe { Arc::from_raw(ptr as *const RwLock).clone() }; - let object = downcasted.read().unwrap().clone(); - - let watched_object = object.watch_whole(self).await; - *downcasted.write().unwrap() = watched_object; - downcasted - } else { - let id = object.read().unwrap().id(); - let object = object.read().unwrap().clone(); - let object = object.clone().watch_whole(self).await; - let wrapped = Arc::new(RwLock::new(object)); - store.insert(id, wrapped.clone()); - wrapped - } - } -} diff --git a/src/gateway/default/mod.rs b/src/gateway/default/mod.rs deleted file mode 100644 index d83b64e..0000000 --- a/src/gateway/default/mod.rs +++ /dev/null @@ -1,101 +0,0 @@ -pub mod gateway; -pub mod handle; - -use super::*; -pub use gateway::*; -pub use handle::*; -use tokio_tungstenite::tungstenite::Message; - -use crate::errors::GatewayError; - -use async_trait::async_trait; -use std::collections::HashMap; -use std::fmt::Debug; -use std::sync::{Arc, RwLock}; -use std::time::Duration; - -use futures_util::stream::{SplitSink, SplitStream}; -use log::{info, warn}; -use tokio::{net::TcpStream, sync::Mutex, task}; -use tokio_tungstenite::{ - connect_async_tls_with_config, Connector, MaybeTlsStream, WebSocketStream, -}; - -impl crate::gateway::MessageCapable for tokio_tungstenite::tungstenite::Message { - fn as_string(&self) -> Option { - match self { - Message::Text(text) => Some(text.clone()), - _ => None, - } - } - - fn is_empty(&self) -> bool { - match self { - Message::Text(text) => text.is_empty(), - Message::Binary(bytes) => bytes.is_empty(), - _ => false, - } - } - - fn as_bytes(&self) -> Option> { - match self { - Message::Binary(bytes) => Some(bytes.clone()), - _ => None, - } - } - - fn from_str(s: &str) -> Self { - Message::Text(s.to_string()) - } -} - -#[cfg(test)] -mod test { - use crate::types; - - use super::*; - use std::sync::atomic::{AtomicI32, Ordering::Relaxed}; - - #[derive(Debug)] - struct Consumer { - _name: String, - events_received: AtomicI32, - } - - #[async_trait] - impl Observer for Consumer { - async fn update(&self, _data: &types::GatewayResume) { - self.events_received.fetch_add(1, Relaxed); - } - } - - #[tokio::test] - async fn test_observer_behavior() { - let mut event = GatewayEvent::default(); - - let new_data = types::GatewayResume { - token: "token_3276ha37am3".to_string(), - session_id: "89346671230".to_string(), - seq: "3".to_string(), - }; - - let consumer = Arc::new(Consumer { - _name: "first".into(), - events_received: 0.into(), - }); - event.subscribe(consumer.clone()); - - let second_consumer = Arc::new(Consumer { - _name: "second".into(), - events_received: 0.into(), - }); - event.subscribe(second_consumer.clone()); - - event.notify(new_data.clone()).await; - event.unsubscribe(&*consumer); - event.notify(new_data).await; - - assert_eq!(consumer.events_received.load(Relaxed), 1); - assert_eq!(second_consumer.events_received.load(Relaxed), 2); - } -} diff --git a/src/gateway/gateway.rs b/src/gateway/gateway.rs new file mode 100644 index 0000000..38efe7c --- /dev/null +++ b/src/gateway/gateway.rs @@ -0,0 +1,537 @@ +use self::event::Events; +use super::handle::GatewayHandle; +use super::heartbeat::HeartbeatHandler; +use super::*; +use crate::types::{ + self, AutoModerationRule, AutoModerationRuleUpdate, Channel, ChannelCreate, ChannelDelete, + ChannelUpdate, Guild, GuildRoleCreate, GuildRoleUpdate, JsonField, RoleObject, SourceUrlField, + ThreadUpdate, UpdateMessage, WebSocketEvent, +}; + +#[derive(Debug)] +pub struct Gateway { + events: Arc>, + heartbeat_handler: HeartbeatHandler, + websocket_send: Arc>, + websocket_receive: WsStream, + kill_send: tokio::sync::broadcast::Sender<()>, + store: Arc>>>>, + url: String, +} + +impl Gateway { + #[allow(clippy::new_ret_no_self)] + pub async fn new(websocket_url: String) -> Result { + let (websocket_send, mut websocket_receive) = + WebSocketBackend::connect(&websocket_url).await?; + + let shared_websocket_send = Arc::new(Mutex::new(websocket_send)); + + // Create a shared broadcast channel for killing all gateway tasks + let (kill_send, mut _kill_receive) = tokio::sync::broadcast::channel::<()>(16); + + // Wait for the first hello and then spawn both tasks so we avoid nested tasks + // This automatically spawns the heartbeat task, but from the main thread + let msg: GatewayMessage = websocket_receive.next().await.unwrap().unwrap().into(); + let gateway_payload: types::GatewayReceivePayload = serde_json::from_str(&msg.0).unwrap(); + + if gateway_payload.op_code != GATEWAY_HELLO { + return Err(GatewayError::NonHelloOnInitiate { + opcode: gateway_payload.op_code, + }); + } + + info!("GW: Received Hello"); + + let gateway_hello: types::HelloData = + serde_json::from_str(gateway_payload.event_data.unwrap().get()).unwrap(); + + let events = Events::default(); + let shared_events = Arc::new(Mutex::new(events)); + + let store = Arc::new(Mutex::new(HashMap::new())); + + let mut gateway = Gateway { + events: shared_events.clone(), + heartbeat_handler: HeartbeatHandler::new( + Duration::from_millis(gateway_hello.heartbeat_interval), + shared_websocket_send.clone(), + kill_send.subscribe(), + ), + websocket_send: shared_websocket_send.clone(), + websocket_receive, + kill_send: kill_send.clone(), + store: store.clone(), + url: websocket_url.clone(), + }; + + // Now we can continuously check for messages in a different task, since we aren't going to receive another hello + task::spawn(async move { + gateway.gateway_listen_task().await; + }); + + Ok(GatewayHandle { + url: websocket_url.clone(), + events: shared_events, + websocket_send: shared_websocket_send.clone(), + kill_send: kill_send.clone(), + store, + }) + } + + /// The main gateway listener task; + /// + /// Can only be stopped by closing the websocket, cannot be made to listen for kill + pub async fn gateway_listen_task(&mut self) { + loop { + let msg = self.websocket_receive.next().await; + + // This if chain can be much better but if let is unstable on stable rust + if let Some(Ok(message)) = msg { + self.handle_message(message.into()).await; + continue; + } + + // We couldn't receive the next message or it was an error, something is wrong with the websocket, close + warn!("GW: Websocket is broken, stopping gateway"); + break; + } + } + + /// Closes the websocket connection and stops all tasks + async fn close(&mut self) { + self.kill_send.send(()).unwrap(); + self.websocket_send.lock().await.close().await.unwrap(); + } + + /// Deserializes and updates a dispatched event, when we already know its type; + /// (Called for every event in handle_message) + #[allow(dead_code)] // TODO: Remove this allow annotation + async fn handle_event<'a, T: WebSocketEvent + serde::Deserialize<'a>>( + data: &'a str, + event: &mut GatewayEvent, + ) -> Result<(), serde_json::Error> { + let data_deserialize_result: Result = serde_json::from_str(data); + + if data_deserialize_result.is_err() { + return Err(data_deserialize_result.err().unwrap()); + } + + event.notify(data_deserialize_result.unwrap()).await; + Ok(()) + } + + /// This handles a message as a websocket event and updates its events along with the events' observers + pub async fn handle_message(&mut self, msg: GatewayMessage) { + if msg.0.is_empty() { + return; + } + + let Ok(gateway_payload) = msg.payload() else { + if let Some(error) = msg.error() { + warn!("GW: Received error {:?}, connection will close..", error); + self.close().await; + self.events.lock().await.error.notify(error).await; + } else { + warn!( + "Message unrecognised: {:?}, please open an issue on the chorus github", + msg.0 + ); + } + return; + }; + + // See https://discord.com/developers/docs/topics/opcodes-and-status-codes#gateway-gateway-opcodes + match gateway_payload.op_code { + // An event was dispatched, we need to look at the gateway event name t + GATEWAY_DISPATCH => { + let Some(event_name) = gateway_payload.event_name else { + warn!("Gateway dispatch op without event_name"); + return; + }; + + trace!("Gateway: Received {event_name}"); + + macro_rules! handle { + ($($name:literal => $($path:ident).+ $( $message_type:ty: $update_type:ty)?),*) => { + match event_name.as_str() { + $($name => { + let event = &mut self.events.lock().await.$($path).+; + let json = gateway_payload.event_data.unwrap().get(); + match serde_json::from_str(json) { + Err(err) => warn!("Failed to parse gateway event {event_name} ({err})"), + Ok(message) => { + $( + let mut message: $message_type = message; + let store = self.store.lock().await; + let id = if message.id().is_some() { + message.id().unwrap() + } else { + event.notify(message).await; + return; + }; + if let Some(to_update) = store.get(&id) { + let object = to_update.clone(); + let inner_object = object.read().unwrap(); + if let Some(_) = inner_object.downcast_ref::<$update_type>() { + let ptr = Arc::into_raw(object.clone()); + // SAFETY: + // - We have just checked that the typeid of the `dyn Any ...` matches that of `T`. + // - This operation doesn't read or write any shared data, and thus cannot cause a data race + // - The reference count is not being modified + let downcasted = unsafe { Arc::from_raw(ptr as *const RwLock<$update_type>).clone() }; + drop(inner_object); + message.set_json(json.to_string()); + message.set_source_url(self.url.clone()); + message.update(downcasted.clone()); + } else { + warn!("Received {} for {}, but it has been observed to be a different type!", $name, id) + } + } + )? + event.notify(message).await; + } + } + },)* + "RESUMED" => (), + "SESSIONS_REPLACE" => { + let result: Result, serde_json::Error> = + serde_json::from_str(gateway_payload.event_data.unwrap().get()); + match result { + Err(err) => { + warn!( + "Failed to parse gateway event {} ({})", + event_name, + err + ); + return; + } + Ok(sessions) => { + self.events.lock().await.session.replace.notify( + types::SessionsReplace {sessions} + ).await; + } + } + }, + _ => { + warn!("Received unrecognized gateway event ({event_name})! Please open an issue on the chorus github so we can implement it"); + } + } + }; + } + + // See https://discord.com/developers/docs/topics/gateway-events#receive-events + // "Some" of these are undocumented + handle!( + "READY" => session.ready, + "READY_SUPPLEMENTAL" => session.ready_supplemental, + "APPLICATION_COMMAND_PERMISSIONS_UPDATE" => application.command_permissions_update, + "AUTO_MODERATION_RULE_CREATE" =>auto_moderation.rule_create, + "AUTO_MODERATION_RULE_UPDATE" =>auto_moderation.rule_update AutoModerationRuleUpdate: AutoModerationRule, + "AUTO_MODERATION_RULE_DELETE" => auto_moderation.rule_delete, + "AUTO_MODERATION_ACTION_EXECUTION" => auto_moderation.action_execution, + "CHANNEL_CREATE" => channel.create ChannelCreate: Guild, + "CHANNEL_UPDATE" => channel.update ChannelUpdate: Channel, + "CHANNEL_UNREAD_UPDATE" => channel.unread_update, + "CHANNEL_DELETE" => channel.delete ChannelDelete: Guild, + "CHANNEL_PINS_UPDATE" => channel.pins_update, + "CALL_CREATE" => call.create, + "CALL_UPDATE" => call.update, + "CALL_DELETE" => call.delete, + "THREAD_CREATE" => thread.create, // TODO + "THREAD_UPDATE" => thread.update ThreadUpdate: Channel, + "THREAD_DELETE" => thread.delete, // TODO + "THREAD_LIST_SYNC" => thread.list_sync, // TODO + "THREAD_MEMBER_UPDATE" => thread.member_update, // TODO + "THREAD_MEMBERS_UPDATE" => thread.members_update, // TODO + "GUILD_CREATE" => guild.create, // TODO + "GUILD_UPDATE" => guild.update, // TODO + "GUILD_DELETE" => guild.delete, // TODO + "GUILD_AUDIT_LOG_ENTRY_CREATE" => guild.audit_log_entry_create, + "GUILD_BAN_ADD" => guild.ban_add, // TODO + "GUILD_BAN_REMOVE" => guild.ban_remove, // TODO + "GUILD_EMOJIS_UPDATE" => guild.emojis_update, // TODO + "GUILD_STICKERS_UPDATE" => guild.stickers_update, // TODO + "GUILD_INTEGRATIONS_UPDATE" => guild.integrations_update, + "GUILD_MEMBER_ADD" => guild.member_add, + "GUILD_MEMBER_REMOVE" => guild.member_remove, + "GUILD_MEMBER_UPDATE" => guild.member_update, // TODO + "GUILD_MEMBERS_CHUNK" => guild.members_chunk, // TODO + "GUILD_ROLE_CREATE" => guild.role_create GuildRoleCreate: Guild, + "GUILD_ROLE_UPDATE" => guild.role_update GuildRoleUpdate: RoleObject, + "GUILD_ROLE_DELETE" => guild.role_delete, // TODO + "GUILD_SCHEDULED_EVENT_CREATE" => guild.role_scheduled_event_create, // TODO + "GUILD_SCHEDULED_EVENT_UPDATE" => guild.role_scheduled_event_update, // TODO + "GUILD_SCHEDULED_EVENT_DELETE" => guild.role_scheduled_event_delete, // TODO + "GUILD_SCHEDULED_EVENT_USER_ADD" => guild.role_scheduled_event_user_add, + "GUILD_SCHEDULED_EVENT_USER_REMOVE" => guild.role_scheduled_event_user_remove, + "PASSIVE_UPDATE_V1" => guild.passive_update_v1, // TODO + "INTEGRATION_CREATE" => integration.create, // TODO + "INTEGRATION_UPDATE" => integration.update, // TODO + "INTEGRATION_DELETE" => integration.delete, // TODO + "INTERACTION_CREATE" => interaction.create, // TODO + "INVITE_CREATE" => invite.create, // TODO + "INVITE_DELETE" => invite.delete, // TODO + "MESSAGE_CREATE" => message.create, + "MESSAGE_UPDATE" => message.update, // TODO + "MESSAGE_DELETE" => message.delete, + "MESSAGE_DELETE_BULK" => message.delete_bulk, + "MESSAGE_REACTION_ADD" => message.reaction_add, // TODO + "MESSAGE_REACTION_REMOVE" => message.reaction_remove, // TODO + "MESSAGE_REACTION_REMOVE_ALL" => message.reaction_remove_all, // TODO + "MESSAGE_REACTION_REMOVE_EMOJI" => message.reaction_remove_emoji, // TODO + "MESSAGE_ACK" => message.ack, + "PRESENCE_UPDATE" => user.presence_update, // TODO + "RELATIONSHIP_ADD" => relationship.add, + "RELATIONSHIP_REMOVE" => relationship.remove, + "STAGE_INSTANCE_CREATE" => stage_instance.create, + "STAGE_INSTANCE_UPDATE" => stage_instance.update, // TODO + "STAGE_INSTANCE_DELETE" => stage_instance.delete, + "TYPING_START" => user.typing_start, + "USER_UPDATE" => user.update, // TODO + "USER_GUILD_SETTINGS_UPDATE" => user.guild_settings_update, + "VOICE_STATE_UPDATE" => voice.state_update, // TODO + "VOICE_SERVER_UPDATE" => voice.server_update, + "WEBHOOKS_UPDATE" => webhooks.update + ); + } + // We received a heartbeat from the server + // "Discord may send the app a Heartbeat (opcode 1) event, in which case the app should send a Heartbeat event immediately." + GATEWAY_HEARTBEAT => { + trace!("GW: Received Heartbeat // Heartbeat Request"); + + // Tell the heartbeat handler it should send a heartbeat right away + + let heartbeat_communication = HeartbeatThreadCommunication { + sequence_number: gateway_payload.sequence_number, + op_code: Some(GATEWAY_HEARTBEAT), + }; + + self.heartbeat_handler + .send + .send(heartbeat_communication) + .await + .unwrap(); + } + GATEWAY_RECONNECT => { + todo!() + } + GATEWAY_INVALID_SESSION => { + todo!() + } + // Starts our heartbeat + // We should have already handled this in gateway init + GATEWAY_HELLO => { + warn!("Received hello when it was unexpected"); + } + GATEWAY_HEARTBEAT_ACK => { + trace!("GW: Received Heartbeat ACK"); + + // Tell the heartbeat handler we received an ack + + let heartbeat_communication = HeartbeatThreadCommunication { + sequence_number: gateway_payload.sequence_number, + op_code: Some(GATEWAY_HEARTBEAT_ACK), + }; + + self.heartbeat_handler + .send + .send(heartbeat_communication) + .await + .unwrap(); + } + GATEWAY_IDENTIFY + | GATEWAY_UPDATE_PRESENCE + | GATEWAY_UPDATE_VOICE_STATE + | GATEWAY_RESUME + | GATEWAY_REQUEST_GUILD_MEMBERS + | GATEWAY_CALL_SYNC + | GATEWAY_LAZY_REQUEST => { + info!( + "Received unexpected opcode ({}) for current state. This might be due to a faulty server implementation and is likely not the fault of chorus.", + gateway_payload.op_code + ); + } + _ => { + warn!("Received unrecognized gateway op code ({})! Please open an issue on the chorus github so we can implement it", gateway_payload.op_code); + } + } + + // If we we received a seq number we should let it know + if let Some(seq_num) = gateway_payload.sequence_number { + let heartbeat_communication = HeartbeatThreadCommunication { + sequence_number: Some(seq_num), + // Op code is irrelevant here + op_code: None, + }; + + self.heartbeat_handler + .send + .send(heartbeat_communication) + .await + .unwrap(); + } + } +} + +pub mod event { + use super::*; + + #[derive(Default, Debug)] + pub struct Events { + pub application: Application, + pub auto_moderation: AutoModeration, + pub session: Session, + pub message: Message, + pub user: User, + pub relationship: Relationship, + pub channel: Channel, + pub thread: Thread, + pub guild: Guild, + pub invite: Invite, + pub integration: Integration, + pub interaction: Interaction, + pub stage_instance: StageInstance, + pub call: Call, + pub voice: Voice, + pub webhooks: Webhooks, + pub gateway_identify_payload: GatewayEvent, + pub gateway_resume: GatewayEvent, + pub error: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Application { + pub command_permissions_update: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct AutoModeration { + pub rule_create: GatewayEvent, + pub rule_update: GatewayEvent, + pub rule_delete: GatewayEvent, + pub action_execution: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Session { + pub ready: GatewayEvent, + pub ready_supplemental: GatewayEvent, + pub replace: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct StageInstance { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub delete: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Message { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub delete: GatewayEvent, + pub delete_bulk: GatewayEvent, + pub reaction_add: GatewayEvent, + pub reaction_remove: GatewayEvent, + pub reaction_remove_all: GatewayEvent, + pub reaction_remove_emoji: GatewayEvent, + pub ack: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct User { + pub update: GatewayEvent, + pub guild_settings_update: GatewayEvent, + pub presence_update: GatewayEvent, + pub typing_start: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Relationship { + pub add: GatewayEvent, + pub remove: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Channel { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub unread_update: GatewayEvent, + pub delete: GatewayEvent, + pub pins_update: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Thread { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub delete: GatewayEvent, + pub list_sync: GatewayEvent, + pub member_update: GatewayEvent, + pub members_update: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Guild { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub delete: GatewayEvent, + pub audit_log_entry_create: GatewayEvent, + pub ban_add: GatewayEvent, + pub ban_remove: GatewayEvent, + pub emojis_update: GatewayEvent, + pub stickers_update: GatewayEvent, + pub integrations_update: GatewayEvent, + pub member_add: GatewayEvent, + pub member_remove: GatewayEvent, + pub member_update: GatewayEvent, + pub members_chunk: GatewayEvent, + pub role_create: GatewayEvent, + pub role_update: GatewayEvent, + pub role_delete: GatewayEvent, + pub role_scheduled_event_create: GatewayEvent, + pub role_scheduled_event_update: GatewayEvent, + pub role_scheduled_event_delete: GatewayEvent, + pub role_scheduled_event_user_add: GatewayEvent, + pub role_scheduled_event_user_remove: GatewayEvent, + pub passive_update_v1: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Invite { + pub create: GatewayEvent, + pub delete: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Integration { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub delete: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Interaction { + pub create: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Call { + pub create: GatewayEvent, + pub update: GatewayEvent, + pub delete: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Voice { + pub state_update: GatewayEvent, + pub server_update: GatewayEvent, + } + + #[derive(Default, Debug)] + pub struct Webhooks { + pub update: GatewayEvent, + } +} diff --git a/src/gateway/heartbeat.rs b/src/gateway/heartbeat.rs new file mode 100644 index 0000000..aeb3dae --- /dev/null +++ b/src/gateway/heartbeat.rs @@ -0,0 +1,128 @@ +use super::*; +use crate::types; + +/// The amount of time we wait for a heartbeat ack before resending our heartbeat in ms +const HEARTBEAT_ACK_TIMEOUT: u64 = 2000; + +/// Handles sending heartbeats to the gateway in another thread +#[allow(dead_code)] // FIXME: Remove this, once HeartbeatHandler is used +#[derive(Debug)] +pub(super) struct HeartbeatHandler { + /// How ofter heartbeats need to be sent at a minimum + pub heartbeat_interval: Duration, + /// The send channel for the heartbeat thread + pub send: Sender, + /// The handle of the thread + handle: JoinHandle<()>, +} + +impl HeartbeatHandler { + pub fn new( + heartbeat_interval: Duration, + websocket_tx: Arc>, + kill_rc: tokio::sync::broadcast::Receiver<()>, + ) -> Self { + let (send, receive) = tokio::sync::mpsc::channel(32); + let kill_receive = kill_rc.resubscribe(); + + let handle: JoinHandle<()> = task::spawn(async move { + Self::heartbeat_task(websocket_tx, heartbeat_interval, receive, kill_receive).await; + }); + + Self { + heartbeat_interval, + send, + handle, + } + } + + /// The main heartbeat task; + /// + /// Can be killed by the kill broadcast; + /// If the websocket is closed, will die out next time it tries to send a heartbeat; + pub async fn heartbeat_task( + websocket_tx: Arc>, + heartbeat_interval: Duration, + mut receive: tokio::sync::mpsc::Receiver, + mut kill_receive: tokio::sync::broadcast::Receiver<()>, + ) { + let mut last_heartbeat_timestamp: Instant = time::Instant::now(); + let mut last_heartbeat_acknowledged = true; + let mut last_seq_number: Option = None; + + loop { + if kill_receive.try_recv().is_ok() { + trace!("GW: Closing heartbeat task"); + break; + } + + let timeout = if last_heartbeat_acknowledged { + heartbeat_interval + } else { + // If the server hasn't acknowledged our heartbeat we should resend it + Duration::from_millis(HEARTBEAT_ACK_TIMEOUT) + }; + + let mut should_send = false; + + tokio::select! { + () = sleep_until(last_heartbeat_timestamp + timeout) => { + should_send = true; + } + Some(communication) = receive.recv() => { + // If we received a seq number update, use that as the last seq number + if communication.sequence_number.is_some() { + last_seq_number = communication.sequence_number; + } + + if let Some(op_code) = communication.op_code { + match op_code { + GATEWAY_HEARTBEAT => { + // As per the api docs, if the server sends us a Heartbeat, that means we need to respond with a heartbeat immediately + should_send = true; + } + GATEWAY_HEARTBEAT_ACK => { + // The server received our heartbeat + last_heartbeat_acknowledged = true; + } + _ => {} + } + } + } + } + + if should_send { + trace!("GW: Sending Heartbeat.."); + + let heartbeat = types::GatewayHeartbeat { + op: GATEWAY_HEARTBEAT, + d: last_seq_number, + }; + + let heartbeat_json = serde_json::to_string(&heartbeat).unwrap(); + + let msg = GatewayMessage(heartbeat_json); + + let send_result = websocket_tx.lock().await.send(msg.into()).await; + if send_result.is_err() { + // We couldn't send, the websocket is broken + warn!("GW: Couldnt send heartbeat, websocket seems broken"); + break; + } + + last_heartbeat_timestamp = time::Instant::now(); + last_heartbeat_acknowledged = false; + } + } + } +} + +/// Used for communications between the heartbeat and gateway thread. +/// Either signifies a sequence number update, a heartbeat ACK or a Heartbeat request by the server +#[derive(Clone, Copy, Debug)] +pub(super) struct HeartbeatThreadCommunication { + /// The opcode for the communication we received, if relevant + pub(super) op_code: Option, + /// The sequence number we got from discord, if any + pub(super) sequence_number: Option, +} diff --git a/src/gateway/message.rs b/src/gateway/message.rs index 12d0e1a..2c12e48 100644 --- a/src/gateway/message.rs +++ b/src/gateway/message.rs @@ -5,24 +5,14 @@ use super::*; /// Represents a messsage received from the gateway. This will be either a [types::GatewayReceivePayload], containing events, or a [GatewayError]. /// This struct is used internally when handling messages. #[derive(Clone, Debug)] -pub struct GatewayMessage { - /// The message we received from the server - pub(crate) message: tokio_tungstenite::tungstenite::Message, -} +pub struct GatewayMessage(pub String); impl GatewayMessage { - /// Creates self from a tungstenite message - pub fn from_tungstenite_message(message: tokio_tungstenite::tungstenite::Message) -> Self { - Self { message } - } - /// Parses the message as an error; /// Returns the error if succesfully parsed, None if the message isn't an error pub fn error(&self) -> Option { - let content = self.message.to_string(); - // Some error strings have dots on the end, which we don't care about - let processed_content = content.to_lowercase().replace('.', ""); + let processed_content = self.0.to_lowercase().replace('.', ""); match processed_content.as_str() { "unknown error" | "4000" => Some(GatewayError::Unknown), @@ -45,29 +35,9 @@ impl GatewayMessage { } } - /// Returns whether or not the message is an error - pub fn is_error(&self) -> bool { - self.error().is_some() - } - /// Parses the message as a payload; /// Returns a result of deserializing pub fn payload(&self) -> Result { - return serde_json::from_str(self.message.to_text().unwrap()); - } - - /// Returns whether or not the message is a payload - pub fn is_payload(&self) -> bool { - // close messages are never payloads, payloads are only text messages - if self.message.is_close() | !self.message.is_text() { - return false; - } - - return self.payload().is_ok(); - } - - /// Returns whether or not the message is empty - pub fn is_empty(&self) -> bool { - self.message.is_empty() + serde_json::from_str(&self.0) } } diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index c2965c3..9ab4f9d 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -1,19 +1,14 @@ -#[cfg(all(not(target_arch = "wasm32"), feature = "client"))] -pub mod default; pub mod events; pub mod message; -// TODO: Uncomment for Prod! -// #[cfg(all(target_arch = "wasm32", feature = "client"))] -pub mod wasm; -#[cfg(all(not(target_arch = "wasm32"), feature = "client"))] -pub use default::*; +#[cfg(not(wasm))] +pub mod backend_tungstenite; +#[cfg(not(wasm))] +use backend_tungstenite::*; + pub use message::*; use safina_timer::sleep_until; use tokio::task::{self, JoinHandle}; -// TODO: Uncomment for Prod! -// #[cfg(all(target_arch = "wasm32", feature = "client"))] -pub use wasm::*; use self::events::Events; use crate::errors::GatewayError; @@ -29,9 +24,6 @@ use std::marker::PhantomData; use std::sync::{Arc, RwLock}; use std::time::{self, Duration, Instant}; -use async_trait::async_trait; -use futures_util::stream::SplitSink; -use futures_util::Sink; use futures_util::SinkExt; use log::{info, trace, warn}; use tokio::sync::mpsc::Sender; diff --git a/src/gateway/wasm/gateway.rs b/src/gateway/wasm/gateway.rs deleted file mode 100644 index 516ac9e..0000000 --- a/src/gateway/wasm/gateway.rs +++ /dev/null @@ -1,127 +0,0 @@ -use std::sync::Arc; -use std::u8; - -use super::events::Events; -use super::*; -use futures_util::stream::SplitStream; -use futures_util::StreamExt; -use tokio::task; -use ws_stream_wasm::*; - -use crate::types::{self, GatewayReceivePayload}; -#[derive(Debug)] -pub struct WasmGateway { - events: Arc>, - heartbeat_handler: HeartbeatHandler, - websocket_send: Arc>>, - websocket_receive: SplitStream, - kill_send: tokio::sync::broadcast::Sender<()>, - store: GatewayStore, - url: String, -} - -#[async_trait] -impl GatewayCapable for WasmGateway { - fn get_events(&self) -> Arc> { - self.events.clone() - } - - fn get_websocket_send(&self) -> Arc>> { - self.websocket_send.clone() - } - - fn get_store(&self) -> GatewayStore { - self.store.clone() - } - - fn get_url(&self) -> String { - self.url.clone() - } - - async fn spawn>( - websocket_url: String, - ) -> Result { - let (_, websocket_stream) = match WsMeta::connect(websocket_url.clone(), None).await { - Ok(ws) => Ok(ws), - Err(e) => Err(GatewayError::CannotConnect { - error: e.to_string(), - }), - }?; - - let (kill_send, mut _kill_receive) = tokio::sync::broadcast::channel::<()>(16); - let (websocket_send, mut websocket_receive) = websocket_stream.split(); - let shared_websocket_send = Arc::new(Mutex::new(websocket_send)); - - let msg = match websocket_receive.next().await { - Some(msg) => match msg { - WsMessage::Text(text) => Ok(text), - WsMessage::Binary(vec) => Err(GatewayError::NonHelloOnInitiate { - opcode: vec.into_iter().next().unwrap_or(u8::MIN), - }), - }, - None => Err(GatewayError::CannotConnect { - error: "No 'Hello' message received!".to_string(), - }), - }?; - - let payload: GatewayReceivePayload = match serde_json::from_str(msg.as_str()) { - Ok(msg) => Ok(msg), - Err(_) => Err(GatewayError::Decode), - }?; - if payload.op_code != GATEWAY_HELLO { - return Err(GatewayError::NonHelloOnInitiate { - opcode: payload.op_code, - }); - }; - - info!("GW: Received Hello"); - - let gateway_hello: types::HelloData = - serde_json::from_str(payload.event_data.unwrap().get()).unwrap(); - - let events = Events::default(); - let shared_events: Arc> = Arc::new(Mutex::new(events)); - let store: GatewayStore = Arc::new(Mutex::new(HashMap::new())); - - let mut gateway = WasmGateway { - events: shared_events.clone(), - heartbeat_handler: HeartbeatHandler::new( - Duration::from_millis(gateway_hello.heartbeat_interval), - shared_websocket_send.clone(), - kill_send.subscribe(), - ), - websocket_send: shared_websocket_send.clone(), - websocket_receive, - store: store.clone(), - kill_send: kill_send.clone(), - url: websocket_url.clone(), - }; - - task::spawn_local(async move { - gateway.gateway_listen_task().await; - }); - - Ok(G::new( - websocket_url.clone(), - shared_events.clone(), - shared_websocket_send.clone(), - kill_send.clone(), - store, - )) - } - - fn get_heartbeat_handler(&self) -> &HeartbeatHandler { - &self.heartbeat_handler - } - - async fn close(&mut self) { - self.kill_send.send(()).unwrap(); - self.websocket_send.lock().await.close().await.unwrap(); - } -} - -impl WasmGateway { - async fn gateway_listen_task(&mut self) { - todo!() - } -} diff --git a/src/gateway/wasm/handle.rs b/src/gateway/wasm/handle.rs deleted file mode 100644 index 3e77ed3..0000000 --- a/src/gateway/wasm/handle.rs +++ /dev/null @@ -1,50 +0,0 @@ -use super::*; -use std::sync::Arc; - -use crate::gateway::GatewayHandleCapable; -use ws_stream_wasm::*; - -#[derive(Debug, Clone)] -pub struct WasmGatewayHandle { - pub url: String, - pub events: Arc>, - pub websocket_send: Arc>>, - /// Tells gateway tasks to close - pub(super) kill_send: tokio::sync::broadcast::Sender<()>, - pub(crate) store: GatewayStore, -} - -#[async_trait(?Send)] -impl GatewayHandleCapable for WasmGatewayHandle { - fn new( - url: String, - events: Arc>, - websocket_send: Arc>>, - kill_send: tokio::sync::broadcast::Sender<()>, - store: GatewayStore, - ) -> Self { - Self { - url, - events, - websocket_send, - kill_send, - store, - } - } - - async fn send_json_event(&self, op_code: u8, to_send: serde_json::Value) { - self.send_json_event(op_code, to_send).await - } - - async fn observe + Send + Sync>( - &self, - object: Arc>, - ) -> Arc> { - self.observe(object).await - } - - async fn close(&self) { - self.kill_send.send(()).unwrap(); - self.websocket_send.lock().await.close().await.unwrap(); - } -} diff --git a/src/gateway/wasm/mod.rs b/src/gateway/wasm/mod.rs deleted file mode 100644 index e664ad2..0000000 --- a/src/gateway/wasm/mod.rs +++ /dev/null @@ -1,34 +0,0 @@ -pub mod gateway; -pub mod handle; -use super::*; -pub use gateway::*; -pub use handle::*; -use ws_stream_wasm::WsMessage; - -impl crate::gateway::MessageCapable for WsMessage { - fn as_string(&self) -> Option { - match self { - WsMessage::Text(text) => Some(text.clone()), - _ => None, - } - } - - fn as_bytes(&self) -> Option> { - match self { - WsMessage::Binary(bytes) => Some(bytes.clone()), - _ => None, - } - } - - fn is_empty(&self) -> bool { - match self { - WsMessage::Text(text) => text.is_empty(), - WsMessage::Binary(bytes) => bytes.is_empty(), - _ => false, - } - } - - fn from_str(s: &str) -> Self { - WsMessage::Text(s.to_string()) - } -}