diff --git a/src/gateway/default/gateway.rs b/src/gateway/default/gateway.rs new file mode 100644 index 0000000..a90a2e9 --- /dev/null +++ b/src/gateway/default/gateway.rs @@ -0,0 +1,349 @@ +use futures_util::StreamExt; + +use self::event::Events; +use super::*; +use crate::types::{self, WebSocketEvent}; + +#[derive(Debug)] +pub struct DefaultGateway { + events: Arc>, + heartbeat_handler: DefaultHeartbeatHandler, + 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< + WebSocketStream>, + WebSocketStream>, + DefaultGatewayHandle, + DefaultHeartbeatHandler, + > for DefaultGateway +{ + fn get_heartbeat_handler(&self) -> &DefaultHeartbeatHandler { + &self.heartbeat_handler + } + + #[allow(clippy::new_ret_no_self)] + async fn get_handle(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 + 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::new(Mutex::new(events)); + + let store = Arc::new(Mutex::new(HashMap::new())); + + let mut gateway = DefaultGateway { + events: shared_events.clone(), + heartbeat_handler: DefaultHeartbeatHandler::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(DefaultGatewayHandle { + url: websocket_url.clone(), + events: shared_events, + websocket_send: shared_websocket_send.clone(), + kill_send: 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) + #[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(()) + } +} + +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/handle.rs b/src/gateway/default/handle.rs similarity index 52% rename from src/gateway/handle.rs rename to src/gateway/default/handle.rs index 6d018a9..14eb9cb 100644 --- a/src/gateway/handle.rs +++ b/src/gateway/default/handle.rs @@ -1,106 +1,6 @@ use super::{event::Events, *}; use crate::types::{self, Composite}; -#[async_trait(?Send)] -pub trait GatewayHandleCapable -where - R: Stream, - S: Sink, -{ - /// Sends json to the gateway with an opcode - async fn send_json_event(&self, op_code: u8, to_send: serde_json::Value); - - /// Observes an Item ``, which will update itself, if new information about this - /// item arrives on the corresponding Gateway Thread - async fn observe + Send + Sync>( - &self, - object: Arc>, - ) -> Arc>; - - /// Recursively observes and updates all updateable fields on the struct T. Returns an object `T` - /// with all of its observable fields being observed. - async fn observe_and_into_inner>( - &self, - object: Arc>, - ) -> T { - let channel = self.observe(object.clone()).await; - let object = channel.read().unwrap().clone(); - object - } - - /// Sends an identify event to the gateway - async fn send_identify(&self, to_send: types::GatewayIdentifyPayload) { - let to_send_value = serde_json::to_value(&to_send).unwrap(); - - trace!("GW: Sending Identify.."); - - self.send_json_event(GATEWAY_IDENTIFY, to_send_value).await; - } - - /// Sends an update presence event to the gateway - async fn send_update_presence(&self, to_send: types::UpdatePresence) { - let to_send_value = serde_json::to_value(&to_send).unwrap(); - - trace!("GW: Sending Update Presence.."); - - self.send_json_event(GATEWAY_UPDATE_PRESENCE, to_send_value) - .await; - } - - /// Sends a resume event to the gateway - async fn send_resume(&self, to_send: types::GatewayResume) { - let to_send_value = serde_json::to_value(&to_send).unwrap(); - - trace!("GW: Sending Resume.."); - - self.send_json_event(GATEWAY_RESUME, to_send_value).await; - } - - /// Sends a request guild members to the server - async fn send_request_guild_members(&self, to_send: types::GatewayRequestGuildMembers) { - let to_send_value = serde_json::to_value(&to_send).unwrap(); - - trace!("GW: Sending Request Guild Members.."); - - self.send_json_event(GATEWAY_REQUEST_GUILD_MEMBERS, to_send_value) - .await; - } - - /// Sends an update voice state to the server - async fn send_update_voice_state(&self, to_send: types::UpdateVoiceState) { - let to_send_value = serde_json::to_value(to_send).unwrap(); - - trace!("GW: Sending Update Voice State.."); - - self.send_json_event(GATEWAY_UPDATE_VOICE_STATE, to_send_value) - .await; - } - - /// Sends a call sync to the server - async fn send_call_sync(&self, to_send: types::CallSync) { - let to_send_value = serde_json::to_value(&to_send).unwrap(); - - trace!("GW: Sending Call Sync.."); - - self.send_json_event(GATEWAY_CALL_SYNC, to_send_value).await; - } - - /// Sends a Lazy Request - async fn send_lazy_request(&self, to_send: types::LazyRequest) { - let to_send_value = serde_json::to_value(&to_send).unwrap(); - - trace!("GW: Sending Lazy Request.."); - - self.send_json_event(GATEWAY_LAZY_REQUEST, to_send_value) - .await; - } - - /// Closes the websocket connection and stops all gateway tasks; - /// - /// Esentially pulls the plug on the gateway, leaving it possible to resume; - async fn close(&self); -} - #[async_trait(?Send)] impl GatewayHandleCapable< diff --git a/src/gateway/heartbeat.rs b/src/gateway/default/heartbeat.rs similarity index 94% rename from src/gateway/heartbeat.rs rename to src/gateway/default/heartbeat.rs index 6906e33..b737366 100644 --- a/src/gateway/heartbeat.rs +++ b/src/gateway/default/heartbeat.rs @@ -5,17 +5,6 @@ use super::*; /// The amount of time we wait for a heartbeat ack before resending our heartbeat in ms const HEARTBEAT_ACK_TIMEOUT: u64 = 2000; -pub trait HeartbeatHandlerCapable> { - fn new( - heartbeat_interval: Duration, - websocket_tx: Arc>>, - kill_rc: tokio::sync::broadcast::Receiver<()>, - ) -> Self; - - fn get_send(&self) -> &Sender; - fn get_heartbeat_interval(&self) -> Duration; -} - /// Handles sending heartbeats to the gateway in another thread #[allow(dead_code)] // FIXME: Remove this, once HeartbeatHandler is used #[derive(Debug)] diff --git a/src/gateway/message.rs b/src/gateway/default/message.rs similarity index 97% rename from src/gateway/message.rs rename to src/gateway/default/message.rs index edee9dd..12d0e1a 100644 --- a/src/gateway/message.rs +++ b/src/gateway/default/message.rs @@ -7,7 +7,7 @@ use super::*; #[derive(Clone, Debug)] pub struct GatewayMessage { /// The message we received from the server - pub(super) message: tokio_tungstenite::tungstenite::Message, + pub(crate) message: tokio_tungstenite::tungstenite::Message, } impl GatewayMessage { diff --git a/src/gateway/default/mod.rs b/src/gateway/default/mod.rs new file mode 100644 index 0000000..4843b96 --- /dev/null +++ b/src/gateway/default/mod.rs @@ -0,0 +1,84 @@ +pub mod gateway; +pub mod handle; +pub mod heartbeat; +pub mod message; + +use super::*; +pub use gateway::*; +pub use handle::*; +use heartbeat::*; +pub use message::*; +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 tokio::time::sleep_until; + +use futures_util::stream::SplitSink; +use futures_util::stream::SplitStream; +use log::{info, trace, warn}; +use tokio::net::TcpStream; +use tokio::sync::mpsc::Sender; +use tokio::sync::Mutex; +use tokio::task; +use tokio::task::JoinHandle; +use tokio::time; +use tokio::time::Instant; +use tokio_tungstenite::MaybeTlsStream; +use tokio_tungstenite::{connect_async_tls_with_config, Connector, WebSocketStream}; + +#[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 deleted file mode 100644 index 9f8ec38..0000000 --- a/src/gateway/gateway.rs +++ /dev/null @@ -1,645 +0,0 @@ -use self::event::Events; -use super::*; -use crate::types::{ - self, AutoModerationRule, AutoModerationRuleUpdate, Channel, ChannelCreate, ChannelDelete, - ChannelUpdate, Guild, GuildRoleCreate, GuildRoleUpdate, JsonField, RoleObject, SourceUrlField, - ThreadUpdate, UpdateMessage, WebSocketEvent, -}; - -pub type GatewayStore = Arc>>>>; - -#[allow(clippy::type_complexity)] -#[async_trait] -pub trait GatewayCapable -where - R: Stream, - S: Sink, - G: GatewayHandleCapable, - H: HeartbeatHandlerCapable + Send + Sync, -{ - fn get_events(&self) -> Arc>; - fn get_websocket_send(&self) -> Arc>>; - fn get_store(&self) -> GatewayStore; - fn get_url(&self) -> String; - fn get_heartbeat_handler(&self) -> &H; - /// Returns a Result with a matching impl of [`GatewayHandleCapable`], or a [`GatewayError`] - /// - /// DOCUMENTME: Explain what this method has to do to be a good get_handle() impl, or link to such documentation - async fn get_handle(websocket_url: String) -> Result; - async fn close(&mut self); - /// This handles a message as a websocket event and updates its events along with the events' observers - async fn handle_message(&mut self, msg: GatewayMessage) { - if msg.is_empty() { - return; - } - - if !msg.is_error() && !msg.is_payload() { - warn!( - "Message unrecognised: {:?}, please open an issue on the chorus github", - msg.message.to_string() - ); - return; - } - - if msg.is_error() { - let error = msg.error().unwrap(); - - warn!("GW: Received error {:?}, connection will close..", error); - - self.close().await; - - let events = self.get_events(); - let events = events.lock().await; - - events.error.notify(error).await; - - return; - } - - let gateway_payload = msg.payload().unwrap(); - println!("gateway payload: {:#?}", &gateway_payload); - - // 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 events = self.get_events(); - let event = &mut 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.get_store(); - let store = 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.get_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) => { - let events = self.get_events(); - let events = events.lock().await; - events.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), - }; - - let heartbeat_thread_communicator = self.get_heartbeat_handler().get_send(); - - heartbeat_thread_communicator - .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), - }; - - let heartbeat_handler = self.get_heartbeat_handler(); - let heartbeat_thread_communicator = heartbeat_handler.get_send(); - - heartbeat_thread_communicator - .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, - }; - - let heartbeat_handler = self.get_heartbeat_handler(); - let heartbeat_thread_communicator = heartbeat_handler.get_send(); - heartbeat_thread_communicator - .send(heartbeat_communication) - .await - .unwrap(); - } - } -} - -#[derive(Debug)] -pub struct DefaultGateway { - events: Arc>, - heartbeat_handler: DefaultHeartbeatHandler, - 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< - WebSocketStream>, - WebSocketStream>, - DefaultGatewayHandle, - DefaultHeartbeatHandler, - > for DefaultGateway -{ - fn get_heartbeat_handler(&self) -> &DefaultHeartbeatHandler { - &self.heartbeat_handler - } - - #[allow(clippy::new_ret_no_self)] - async fn get_handle(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 - 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::new(Mutex::new(events)); - - let store = Arc::new(Mutex::new(HashMap::new())); - - let mut gateway = DefaultGateway { - events: shared_events.clone(), - heartbeat_handler: DefaultHeartbeatHandler::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(DefaultGatewayHandle { - url: websocket_url.clone(), - events: shared_events, - websocket_send: shared_websocket_send.clone(), - kill_send: 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) - #[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(()) - } -} - -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/mod.rs b/src/gateway/mod.rs index b5888b4..23dda75 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -1,39 +1,31 @@ -pub mod gateway; -pub mod handle; -pub mod heartbeat; -pub mod message; +pub mod default; -pub use gateway::*; -pub use handle::*; -use heartbeat::*; -pub use message::*; +use self::event::Events; +use crate::types::{ + self, AutoModerationRule, AutoModerationRuleUpdate, Channel, ChannelCreate, ChannelDelete, + ChannelUpdate, Composite, Guild, GuildRoleCreate, GuildRoleUpdate, JsonField, RoleObject, + Snowflake, SourceUrlField, ThreadUpdate, UpdateMessage, WebSocketEvent, +}; +use default::heartbeat::HeartbeatThreadCommunication; use tokio_tungstenite::tungstenite::Message; use crate::errors::GatewayError; -use crate::types::{Snowflake, WebSocketEvent}; use async_trait::async_trait; use std::any::Any; use std::collections::HashMap; -use std::fmt::Debug; use std::sync::{Arc, RwLock}; use std::time::Duration; -use tokio::time::sleep_until; +pub use default::*; use futures_util::stream::SplitSink; -use futures_util::stream::SplitStream; -use futures_util::{Sink, StreamExt}; +use futures_util::Sink; use futures_util::{SinkExt, Stream}; use log::{info, trace, warn}; -use tokio::net::TcpStream; use tokio::sync::mpsc::Sender; use tokio::sync::Mutex; -use tokio::task; -use tokio::task::JoinHandle; -use tokio::time; -use tokio::time::Instant; -use tokio_tungstenite::MaybeTlsStream; -use tokio_tungstenite::{connect_async_tls_with_config, Connector, WebSocketStream}; + +pub type GatewayStore = Arc>>>>; // Gateway opcodes /// Opcode received when the server dispatches a [crate::types::WebSocketEvent] @@ -136,53 +128,405 @@ impl GatewayEvent { } } -#[cfg(test)] -mod test { - use crate::types; +#[allow(clippy::type_complexity)] +#[async_trait] +pub trait GatewayCapable +where + R: Stream, + S: Sink, + G: GatewayHandleCapable, + H: HeartbeatHandlerCapable + Send + Sync, +{ + fn get_events(&self) -> Arc>; + fn get_websocket_send(&self) -> Arc>>; + fn get_store(&self) -> GatewayStore; + fn get_url(&self) -> String; + fn get_heartbeat_handler(&self) -> &H; + /// Returns a Result with a matching impl of [`GatewayHandleCapable`], or a [`GatewayError`] + /// + /// DOCUMENTME: Explain what this method has to do to be a good get_handle() impl, or link to such documentation + async fn get_handle(websocket_url: String) -> Result; + async fn close(&mut self); + /// This handles a message as a websocket event and updates its events along with the events' observers + async fn handle_message(&mut self, msg: GatewayMessage) { + if msg.is_empty() { + return; + } - use super::*; - use std::sync::atomic::{AtomicI32, Ordering::Relaxed}; + if !msg.is_error() && !msg.is_payload() { + warn!( + "Message unrecognised: {:?}, please open an issue on the chorus github", + msg.message.to_string() + ); + return; + } - #[derive(Debug)] - struct Consumer { - _name: String, - events_received: AtomicI32, - } + if msg.is_error() { + let error = msg.error().unwrap(); - #[async_trait] - impl Observer for Consumer { - async fn update(&self, _data: &types::GatewayResume) { - self.events_received.fetch_add(1, Relaxed); + warn!("GW: Received error {:?}, connection will close..", error); + + self.close().await; + + let events = self.get_events(); + let events = events.lock().await; + + events.error.notify(error).await; + + return; + } + + let gateway_payload = msg.payload().unwrap(); + println!("gateway payload: {:#?}", &gateway_payload); + + // 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 events = self.get_events(); + let event = &mut 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.get_store(); + let store = 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.get_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) => { + let events = self.get_events(); + let events = events.lock().await; + events.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), + }; + + let heartbeat_thread_communicator = self.get_heartbeat_handler().get_send(); + + heartbeat_thread_communicator + .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), + }; + + let heartbeat_handler = self.get_heartbeat_handler(); + let heartbeat_thread_communicator = heartbeat_handler.get_send(); + + heartbeat_thread_communicator + .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, + }; + + let heartbeat_handler = self.get_heartbeat_handler(); + let heartbeat_thread_communicator = heartbeat_handler.get_send(); + heartbeat_thread_communicator + .send(heartbeat_communication) + .await + .unwrap(); } } - - #[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); - } +} + +#[async_trait(?Send)] +pub trait GatewayHandleCapable +where + R: Stream, + S: Sink, +{ + /// Sends json to the gateway with an opcode + async fn send_json_event(&self, op_code: u8, to_send: serde_json::Value); + + /// Observes an Item ``, which will update itself, if new information about this + /// item arrives on the corresponding Gateway Thread + async fn observe + Send + Sync>( + &self, + object: Arc>, + ) -> Arc>; + + /// Recursively observes and updates all updateable fields on the struct T. Returns an object `T` + /// with all of its observable fields being observed. + async fn observe_and_into_inner>( + &self, + object: Arc>, + ) -> T { + let channel = self.observe(object.clone()).await; + let object = channel.read().unwrap().clone(); + object + } + + /// Sends an identify event to the gateway + async fn send_identify(&self, to_send: types::GatewayIdentifyPayload) { + let to_send_value = serde_json::to_value(&to_send).unwrap(); + + trace!("GW: Sending Identify.."); + + self.send_json_event(GATEWAY_IDENTIFY, to_send_value).await; + } + + /// Sends an update presence event to the gateway + async fn send_update_presence(&self, to_send: types::UpdatePresence) { + let to_send_value = serde_json::to_value(&to_send).unwrap(); + + trace!("GW: Sending Update Presence.."); + + self.send_json_event(GATEWAY_UPDATE_PRESENCE, to_send_value) + .await; + } + + /// Sends a resume event to the gateway + async fn send_resume(&self, to_send: types::GatewayResume) { + let to_send_value = serde_json::to_value(&to_send).unwrap(); + + trace!("GW: Sending Resume.."); + + self.send_json_event(GATEWAY_RESUME, to_send_value).await; + } + + /// Sends a request guild members to the server + async fn send_request_guild_members(&self, to_send: types::GatewayRequestGuildMembers) { + let to_send_value = serde_json::to_value(&to_send).unwrap(); + + trace!("GW: Sending Request Guild Members.."); + + self.send_json_event(GATEWAY_REQUEST_GUILD_MEMBERS, to_send_value) + .await; + } + + /// Sends an update voice state to the server + async fn send_update_voice_state(&self, to_send: types::UpdateVoiceState) { + let to_send_value = serde_json::to_value(to_send).unwrap(); + + trace!("GW: Sending Update Voice State.."); + + self.send_json_event(GATEWAY_UPDATE_VOICE_STATE, to_send_value) + .await; + } + + /// Sends a call sync to the server + async fn send_call_sync(&self, to_send: types::CallSync) { + let to_send_value = serde_json::to_value(&to_send).unwrap(); + + trace!("GW: Sending Call Sync.."); + + self.send_json_event(GATEWAY_CALL_SYNC, to_send_value).await; + } + + /// Sends a Lazy Request + async fn send_lazy_request(&self, to_send: types::LazyRequest) { + let to_send_value = serde_json::to_value(&to_send).unwrap(); + + trace!("GW: Sending Lazy Request.."); + + self.send_json_event(GATEWAY_LAZY_REQUEST, to_send_value) + .await; + } + + /// Closes the websocket connection and stops all gateway tasks; + /// + /// Esentially pulls the plug on the gateway, leaving it possible to resume; + async fn close(&self); +} + +pub trait HeartbeatHandlerCapable> { + fn new( + heartbeat_interval: Duration, + websocket_tx: Arc>>, + kill_rc: tokio::sync::broadcast::Receiver<()>, + ) -> Self; + + fn get_send(&self) -> &Sender; + fn get_heartbeat_interval(&self) -> Duration; }