Resolve merge conflicts
This commit is contained in:
parent
79e70c43a2
commit
a4d5ebb689
|
@ -1,11 +1,9 @@
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use chorus::gateway::{GatewayCapable, GatewayHandleCapable};
|
use chorus::gateway::Gateway;
|
||||||
use chorus::GatewayHandle;
|
|
||||||
use chorus::{
|
use chorus::{
|
||||||
self,
|
self,
|
||||||
gateway::Observer,
|
gateway::Observer,
|
||||||
types::{GatewayIdentifyPayload, GatewayReady},
|
types::{GatewayIdentifyPayload, GatewayReady},
|
||||||
Gateway,
|
|
||||||
};
|
};
|
||||||
use std::{sync::Arc, time::Duration};
|
use std::{sync::Arc, time::Duration};
|
||||||
use tokio::{self, time::sleep};
|
use tokio::{self, time::sleep};
|
||||||
|
@ -33,7 +31,7 @@ async fn main() {
|
||||||
let websocket_url_spacebar = "wss://gateway.old.server.spacebar.chat/".to_string();
|
let websocket_url_spacebar = "wss://gateway.old.server.spacebar.chat/".to_string();
|
||||||
|
|
||||||
// Initiate the gateway connection
|
// Initiate the gateway connection
|
||||||
let gateway: GatewayHandle = Gateway::spawn(websocket_url_spacebar).await.unwrap();
|
let gateway = Gateway::spawn(websocket_url_spacebar).await.unwrap();
|
||||||
|
|
||||||
// Create an instance of our observer
|
// Create an instance of our observer
|
||||||
let observer = ExampleObserver {};
|
let observer = ExampleObserver {};
|
||||||
|
|
|
@ -1,8 +1,7 @@
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use chorus::gateway::{GatewayCapable, GatewayHandleCapable};
|
use chorus::gateway::Gateway;
|
||||||
use chorus::GatewayHandle;
|
use chorus::{self, types::GatewayIdentifyPayload};
|
||||||
use chorus::{self, types::GatewayIdentifyPayload, Gateway};
|
|
||||||
use tokio::time::sleep;
|
use tokio::time::sleep;
|
||||||
|
|
||||||
/// This example creates a simple gateway connection and a session with an Identify event
|
/// This example creates a simple gateway connection and a session with an Identify event
|
||||||
|
@ -12,7 +11,7 @@ async fn main() {
|
||||||
let websocket_url_spacebar = "wss://gateway.old.server.spacebar.chat/".to_string();
|
let websocket_url_spacebar = "wss://gateway.old.server.spacebar.chat/".to_string();
|
||||||
|
|
||||||
// Initiate the gateway connection, starting a listener in one thread and a heartbeat handler in another
|
// Initiate the gateway connection, starting a listener in one thread and a heartbeat handler in another
|
||||||
let gateway: GatewayHandle = Gateway::spawn(websocket_url_spacebar).await.unwrap();
|
let gateway = Gateway::spawn(websocket_url_spacebar).await.unwrap();
|
||||||
|
|
||||||
// At this point, we are connected to the server and are sending heartbeats, however we still haven't authenticated
|
// At this point, we are connected to the server and are sending heartbeats, however we still haven't authenticated
|
||||||
|
|
||||||
|
|
|
@ -4,11 +4,10 @@ use reqwest::Client;
|
||||||
use serde_json::to_string;
|
use serde_json::to_string;
|
||||||
|
|
||||||
use crate::errors::ChorusResult;
|
use crate::errors::ChorusResult;
|
||||||
use crate::gateway::{DefaultGatewayHandle, GatewayCapable, GatewayHandleCapable};
|
use crate::gateway::Gateway;
|
||||||
use crate::instance::{ChorusUser, Instance};
|
use crate::instance::{ChorusUser, Instance};
|
||||||
use crate::ratelimiter::ChorusRequest;
|
use crate::ratelimiter::ChorusRequest;
|
||||||
use crate::types::{GatewayIdentifyPayload, LimitType, LoginResult, LoginSchema};
|
use crate::types::{GatewayIdentifyPayload, LimitType, LoginResult, LoginSchema};
|
||||||
use crate::Gateway;
|
|
||||||
|
|
||||||
impl Instance {
|
impl Instance {
|
||||||
/// Logs into an existing account on the spacebar server.
|
/// Logs into an existing account on the spacebar server.
|
||||||
|
@ -37,7 +36,7 @@ impl Instance {
|
||||||
self.limits_information.as_mut().unwrap().ratelimits = shell.limits.clone().unwrap();
|
self.limits_information.as_mut().unwrap().ratelimits = shell.limits.clone().unwrap();
|
||||||
}
|
}
|
||||||
let mut identify = GatewayIdentifyPayload::common();
|
let mut identify = GatewayIdentifyPayload::common();
|
||||||
let gateway: DefaultGatewayHandle = Gateway::spawn(self.urls.wss.clone()).await.unwrap();
|
let gateway = Gateway::spawn(self.urls.wss.clone()).await.unwrap();
|
||||||
identify.token = login_result.token.clone();
|
identify.token = login_result.token.clone();
|
||||||
gateway.send_identify(identify).await;
|
gateway.send_identify(identify).await;
|
||||||
let user = ChorusUser::new(
|
let user = ChorusUser::new(
|
||||||
|
|
|
@ -3,10 +3,9 @@ use std::sync::{Arc, RwLock};
|
||||||
pub use login::*;
|
pub use login::*;
|
||||||
pub use register::*;
|
pub use register::*;
|
||||||
|
|
||||||
use crate::gateway::{DefaultGatewayHandle, GatewayCapable, GatewayHandleCapable};
|
use crate::gateway::Gateway;
|
||||||
use crate::{
|
use crate::{
|
||||||
errors::ChorusResult,
|
errors::ChorusResult,
|
||||||
gateway::DefaultGateway,
|
|
||||||
instance::{ChorusUser, Instance},
|
instance::{ChorusUser, Instance},
|
||||||
types::{GatewayIdentifyPayload, User},
|
types::{GatewayIdentifyPayload, User},
|
||||||
};
|
};
|
||||||
|
@ -26,8 +25,7 @@ impl Instance {
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let mut identify = GatewayIdentifyPayload::common();
|
let mut identify = GatewayIdentifyPayload::common();
|
||||||
let gateway: DefaultGatewayHandle =
|
let gateway = Gateway::spawn(self.urls.wss.clone()).await.unwrap();
|
||||||
DefaultGateway::spawn(self.urls.wss.clone()).await.unwrap();
|
|
||||||
identify.token = token.clone();
|
identify.token = token.clone();
|
||||||
gateway.send_identify(identify).await;
|
gateway.send_identify(identify).await;
|
||||||
let user = ChorusUser::new(
|
let user = ChorusUser::new(
|
||||||
|
|
|
@ -3,7 +3,7 @@ use std::sync::{Arc, RwLock};
|
||||||
use reqwest::Client;
|
use reqwest::Client;
|
||||||
use serde_json::to_string;
|
use serde_json::to_string;
|
||||||
|
|
||||||
use crate::gateway::{GatewayCapable, GatewayHandleCapable};
|
use crate::gateway::{Gateway, GatewayHandle};
|
||||||
use crate::types::GatewayIdentifyPayload;
|
use crate::types::GatewayIdentifyPayload;
|
||||||
use crate::{
|
use crate::{
|
||||||
errors::ChorusResult,
|
errors::ChorusResult,
|
||||||
|
@ -12,7 +12,6 @@ use crate::{
|
||||||
types::LimitType,
|
types::LimitType,
|
||||||
types::RegisterSchema,
|
types::RegisterSchema,
|
||||||
};
|
};
|
||||||
use crate::{Gateway, GatewayHandle};
|
|
||||||
|
|
||||||
impl Instance {
|
impl Instance {
|
||||||
/// Registers a new user on the server.
|
/// Registers a new user on the server.
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::types;
|
||||||
|
|
||||||
#[derive(Default, Debug)]
|
#[derive(Default, Debug)]
|
||||||
pub struct Events {
|
pub struct Events {
|
||||||
|
|
|
@ -1,7 +1,12 @@
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use futures_util::{SinkExt, StreamExt};
|
||||||
|
use log::*;
|
||||||
|
use tokio::task;
|
||||||
|
|
||||||
use self::event::Events;
|
use self::event::Events;
|
||||||
use super::handle::GatewayHandle;
|
|
||||||
use super::heartbeat::HeartbeatHandler;
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use super::{WsSink, WsStream};
|
||||||
use crate::types::{
|
use crate::types::{
|
||||||
self, AutoModerationRule, AutoModerationRuleUpdate, Channel, ChannelCreate, ChannelDelete,
|
self, AutoModerationRule, AutoModerationRuleUpdate, Channel, ChannelCreate, ChannelDelete,
|
||||||
ChannelUpdate, Guild, GuildRoleCreate, GuildRoleUpdate, JsonField, RoleObject, SourceUrlField,
|
ChannelUpdate, Guild, GuildRoleCreate, GuildRoleUpdate, JsonField, RoleObject, SourceUrlField,
|
||||||
|
@ -21,7 +26,7 @@ pub struct Gateway {
|
||||||
|
|
||||||
impl Gateway {
|
impl Gateway {
|
||||||
#[allow(clippy::new_ret_no_self)]
|
#[allow(clippy::new_ret_no_self)]
|
||||||
pub async fn new(websocket_url: String) -> Result<GatewayHandle, GatewayError> {
|
pub async fn spawn(websocket_url: String) -> Result<GatewayHandle, GatewayError> {
|
||||||
let (websocket_send, mut websocket_receive) =
|
let (websocket_send, mut websocket_receive) =
|
||||||
WebSocketBackend::connect(&websocket_url).await?;
|
WebSocketBackend::connect(&websocket_url).await?;
|
||||||
|
|
||||||
|
|
|
@ -0,0 +1,168 @@
|
||||||
|
use futures_util::SinkExt;
|
||||||
|
use log::*;
|
||||||
|
|
||||||
|
use std::fmt::Debug;
|
||||||
|
|
||||||
|
use super::{event::Events, *};
|
||||||
|
use crate::types::{self, Composite};
|
||||||
|
|
||||||
|
/// 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 GatewayHandle {
|
||||||
|
pub url: String,
|
||||||
|
pub events: Arc<Mutex<Events>>,
|
||||||
|
pub websocket_send: Arc<Mutex<WsSink>>,
|
||||||
|
/// Tells gateway tasks to close
|
||||||
|
pub(super) kill_send: tokio::sync::broadcast::Sender<()>,
|
||||||
|
pub(crate) store: Arc<Mutex<HashMap<Snowflake, Arc<RwLock<ObservableObject>>>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl GatewayHandle {
|
||||||
|
/// Sends json to the gateway with an opcode
|
||||||
|
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 = GatewayMessage(payload_json);
|
||||||
|
|
||||||
|
self.websocket_send
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.send(message.into())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn observe<T: Updateable + Clone + Debug + Composite<T>>(
|
||||||
|
&self,
|
||||||
|
object: Arc<RwLock<T>>,
|
||||||
|
) -> Arc<RwLock<T>> {
|
||||||
|
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::<T>()
|
||||||
|
.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<T>).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
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Recursively observes and updates all updateable fields on the struct T. Returns an object `T`
|
||||||
|
/// with all of its observable fields being observed.
|
||||||
|
pub async fn observe_and_into_inner<T: Updateable + Clone + Debug + Composite<T>>(
|
||||||
|
&self,
|
||||||
|
object: Arc<RwLock<T>>,
|
||||||
|
) -> T {
|
||||||
|
let channel = self.observe(object.clone()).await;
|
||||||
|
let object = channel.read().unwrap().clone();
|
||||||
|
object
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sends an identify event to the gateway
|
||||||
|
pub 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 a resume event to the gateway
|
||||||
|
pub 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 an update presence event to the gateway
|
||||||
|
pub 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 request guild members to the server
|
||||||
|
pub 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
|
||||||
|
pub 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
|
||||||
|
pub 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
|
||||||
|
pub 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;
|
||||||
|
pub async fn close(&self) {
|
||||||
|
self.kill_send.send(()).unwrap();
|
||||||
|
self.websocket_send.lock().await.close().await.unwrap();
|
||||||
|
}
|
||||||
|
}
|
|
@ -1,3 +1,11 @@
|
||||||
|
use futures_util::SinkExt;
|
||||||
|
use log::*;
|
||||||
|
use std::time::{self, Duration, Instant};
|
||||||
|
use tokio::sync::mpsc::{Receiver, Sender};
|
||||||
|
|
||||||
|
use safina_timer::sleep_until;
|
||||||
|
use tokio::task::{self, JoinHandle};
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::types;
|
use crate::types;
|
||||||
|
|
||||||
|
@ -43,13 +51,15 @@ impl HeartbeatHandler {
|
||||||
pub async fn heartbeat_task(
|
pub async fn heartbeat_task(
|
||||||
websocket_tx: Arc<Mutex<WsSink>>,
|
websocket_tx: Arc<Mutex<WsSink>>,
|
||||||
heartbeat_interval: Duration,
|
heartbeat_interval: Duration,
|
||||||
mut receive: tokio::sync::mpsc::Receiver<HeartbeatThreadCommunication>,
|
mut receive: Receiver<HeartbeatThreadCommunication>,
|
||||||
mut kill_receive: tokio::sync::broadcast::Receiver<()>,
|
mut kill_receive: tokio::sync::broadcast::Receiver<()>,
|
||||||
) {
|
) {
|
||||||
let mut last_heartbeat_timestamp: Instant = time::Instant::now();
|
let mut last_heartbeat_timestamp: Instant = time::Instant::now();
|
||||||
let mut last_heartbeat_acknowledged = true;
|
let mut last_heartbeat_acknowledged = true;
|
||||||
let mut last_seq_number: Option<u64> = None;
|
let mut last_seq_number: Option<u64> = None;
|
||||||
|
|
||||||
|
safina_timer::start_timer_thread();
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
if kill_receive.try_recv().is_ok() {
|
if kill_receive.try_recv().is_ok() {
|
||||||
trace!("GW: Closing heartbeat task");
|
trace!("GW: Closing heartbeat task");
|
||||||
|
|
|
@ -1,38 +1,32 @@
|
||||||
|
use async_trait::async_trait;
|
||||||
|
|
||||||
|
pub mod backend_tungstenite;
|
||||||
pub mod events;
|
pub mod events;
|
||||||
|
pub mod gateway;
|
||||||
|
pub mod handle;
|
||||||
|
pub mod heartbeat;
|
||||||
pub mod message;
|
pub mod message;
|
||||||
|
|
||||||
#[cfg(not(wasm))]
|
pub use gateway::*;
|
||||||
pub mod backend_tungstenite;
|
pub use handle::*;
|
||||||
#[cfg(not(wasm))]
|
use heartbeat::*;
|
||||||
use backend_tungstenite::*;
|
|
||||||
|
|
||||||
pub use message::*;
|
pub use message::*;
|
||||||
use safina_timer::sleep_until;
|
|
||||||
use tokio::task::{self, JoinHandle};
|
|
||||||
|
|
||||||
use self::events::Events;
|
|
||||||
use crate::errors::GatewayError;
|
use crate::errors::GatewayError;
|
||||||
use crate::types::{
|
use crate::types::{Snowflake, WebSocketEvent};
|
||||||
self, AutoModerationRule, AutoModerationRuleUpdate, Channel, ChannelCreate, ChannelDelete,
|
|
||||||
ChannelUpdate, Composite, Guild, GuildRoleCreate, GuildRoleUpdate, JsonField, RoleObject,
|
|
||||||
Snowflake, SourceUrlField, ThreadUpdate, UpdateMessage, WebSocketEvent,
|
|
||||||
};
|
|
||||||
|
|
||||||
use std::any::Any;
|
use std::any::Any;
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::marker::PhantomData;
|
|
||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, RwLock};
|
||||||
use std::time::{self, Duration, Instant};
|
|
||||||
|
|
||||||
use futures_util::SinkExt;
|
|
||||||
use log::{info, trace, warn};
|
|
||||||
use tokio::sync::mpsc::Sender;
|
|
||||||
use tokio::sync::Mutex;
|
use tokio::sync::Mutex;
|
||||||
|
|
||||||
pub type GatewayStore = Arc<Mutex<HashMap<Snowflake, Arc<RwLock<ObservableObject>>>>>;
|
#[cfg(not(target_arch = "wasm32"))]
|
||||||
|
pub type WsSink = backend_tungstenite::WsSink;
|
||||||
/// The amount of time we wait for a heartbeat ack before resending our heartbeat in ms
|
#[cfg(not(target_arch = "wasm32"))]
|
||||||
const HEARTBEAT_ACK_TIMEOUT: u64 = 2000;
|
pub type WsStream = backend_tungstenite::WsStream;
|
||||||
|
#[cfg(not(target_arch = "wasm32"))]
|
||||||
|
pub type WebSocketBackend = backend_tungstenite::WebSocketBackend;
|
||||||
|
|
||||||
// Gateway opcodes
|
// Gateway opcodes
|
||||||
/// Opcode received when the server dispatches a [crate::types::WebSocketEvent]
|
/// Opcode received when the server dispatches a [crate::types::WebSocketEvent]
|
||||||
|
@ -82,25 +76,8 @@ const GATEWAY_CALL_SYNC: u8 = 13;
|
||||||
/// See [types::LazyRequest]
|
/// See [types::LazyRequest]
|
||||||
const GATEWAY_LAZY_REQUEST: u8 = 14;
|
const GATEWAY_LAZY_REQUEST: u8 = 14;
|
||||||
|
|
||||||
pub trait MessageCapable {
|
|
||||||
fn as_string(&self) -> Option<String>;
|
|
||||||
fn as_bytes(&self) -> Option<Vec<u8>>;
|
|
||||||
fn is_empty(&self) -> bool;
|
|
||||||
fn from_str(s: &str) -> Self;
|
|
||||||
}
|
|
||||||
|
|
||||||
pub type ObservableObject = dyn Send + Sync + Any;
|
pub type ObservableObject = dyn Send + Sync + Any;
|
||||||
|
|
||||||
/// 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 struct HeartbeatThreadCommunication {
|
|
||||||
/// The opcode for the communication we received, if relevant
|
|
||||||
pub op_code: Option<u8>,
|
|
||||||
/// The sequence number we got from discord, if any
|
|
||||||
pub sequence_number: Option<u64>,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// An entity type which is supposed to be updateable via the Gateway. This is implemented for all such types chorus supports, implementing it for your own types is likely a mistake.
|
/// An entity type which is supposed to be updateable via the Gateway. This is implemented for all such types chorus supports, implementing it for your own types is likely a mistake.
|
||||||
pub trait Updateable: 'static + Send + Sync {
|
pub trait Updateable: 'static + Send + Sync {
|
||||||
fn id(&self) -> Snowflake;
|
fn id(&self) -> Snowflake;
|
||||||
|
@ -151,523 +128,3 @@ impl<T: WebSocketEvent> GatewayEvent<T> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
pub trait GatewayCapable<T, S>
|
|
||||||
where
|
|
||||||
T: MessageCapable + Send + 'static,
|
|
||||||
S: Sink<T> + Send,
|
|
||||||
{
|
|
||||||
fn get_events(&self) -> Arc<Mutex<Events>>;
|
|
||||||
fn get_websocket_send(&self) -> Arc<Mutex<SplitSink<S, T>>>;
|
|
||||||
fn get_store(&self) -> GatewayStore;
|
|
||||||
fn get_url(&self) -> String;
|
|
||||||
fn get_heartbeat_handler(&self) -> &HeartbeatHandler<T, S>;
|
|
||||||
/// 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
|
|
||||||
/// TODO: Give spawn a default trait impl, avoid code duplication
|
|
||||||
async fn spawn<G: GatewayHandleCapable<T, S>>(websocket_url: String)
|
|
||||||
-> Result<G, GatewayError>;
|
|
||||||
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<Vec<types::Session>, 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().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.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.send;
|
|
||||||
heartbeat_thread_communicator
|
|
||||||
.send(heartbeat_communication)
|
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait(?Send)]
|
|
||||||
pub trait GatewayHandleCapable<T, S>
|
|
||||||
where
|
|
||||||
T: MessageCapable + Send + 'static,
|
|
||||||
S: Sink<T>,
|
|
||||||
{
|
|
||||||
fn new(
|
|
||||||
url: String,
|
|
||||||
events: Arc<Mutex<Events>>,
|
|
||||||
websocket_send: Arc<Mutex<SplitSink<S, T>>>,
|
|
||||||
kill_send: tokio::sync::broadcast::Sender<()>,
|
|
||||||
store: GatewayStore,
|
|
||||||
) -> Self;
|
|
||||||
|
|
||||||
/// 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 `<T: Updateable>`, which will update itself, if new information about this
|
|
||||||
/// item arrives on the corresponding Gateway Thread
|
|
||||||
async fn observe<U: Updateable + Clone + std::fmt::Debug + Composite<U> + Send + Sync>(
|
|
||||||
&self,
|
|
||||||
object: Arc<RwLock<U>>,
|
|
||||||
) -> Arc<RwLock<U>>;
|
|
||||||
|
|
||||||
/// 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<U: Updateable + Clone + std::fmt::Debug + Composite<U>>(
|
|
||||||
&self,
|
|
||||||
object: Arc<RwLock<U>>,
|
|
||||||
) -> U {
|
|
||||||
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);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Handles sending heartbeats to the gateway in another thread
|
|
||||||
#[derive(Debug)]
|
|
||||||
pub struct HeartbeatHandler<T: MessageCapable + Send + 'static, S: Sink<T>> {
|
|
||||||
/// How ofter heartbeats need to be sent at a minimum
|
|
||||||
pub heartbeat_interval: Duration,
|
|
||||||
/// The send channel for the heartbeat thread
|
|
||||||
pub send: Sender<HeartbeatThreadCommunication>,
|
|
||||||
/// The handle of the thread
|
|
||||||
handle: JoinHandle<()>,
|
|
||||||
hb_type: (PhantomData<T>, PhantomData<S>),
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<T: MessageCapable + Send + 'static, S: Sink<T> + Send + 'static> HeartbeatHandler<T, S> {
|
|
||||||
pub async fn heartbeat_task(
|
|
||||||
websocket_tx: Arc<Mutex<SplitSink<S, T>>>,
|
|
||||||
heartbeat_interval: Duration,
|
|
||||||
mut receive: tokio::sync::mpsc::Receiver<HeartbeatThreadCommunication>,
|
|
||||||
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<u64> = None;
|
|
||||||
safina_timer::start_timer_thread();
|
|
||||||
|
|
||||||
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 = tokio_tungstenite::tungstenite::Message::text(heartbeat_json);
|
|
||||||
|
|
||||||
let send_result = websocket_tx
|
|
||||||
.lock()
|
|
||||||
.await
|
|
||||||
.send(MessageCapable::from_str(msg.to_string().as_str()))
|
|
||||||
.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;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn new(
|
|
||||||
heartbeat_interval: Duration,
|
|
||||||
websocket_tx: Arc<Mutex<SplitSink<S, T>>>,
|
|
||||||
kill_rc: tokio::sync::broadcast::Receiver<()>,
|
|
||||||
) -> HeartbeatHandler<T, S> {
|
|
||||||
let (send, receive) = tokio::sync::mpsc::channel(32);
|
|
||||||
let kill_receive = kill_rc.resubscribe();
|
|
||||||
|
|
||||||
let handle: JoinHandle<()> = task::spawn(async move {
|
|
||||||
HeartbeatHandler::heartbeat_task(
|
|
||||||
websocket_tx,
|
|
||||||
heartbeat_interval,
|
|
||||||
receive,
|
|
||||||
kill_receive,
|
|
||||||
)
|
|
||||||
.await;
|
|
||||||
});
|
|
||||||
|
|
||||||
Self {
|
|
||||||
heartbeat_interval,
|
|
||||||
send,
|
|
||||||
handle,
|
|
||||||
hb_type: (PhantomData::<T>, PhantomData::<S>),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
|
@ -9,11 +9,11 @@ use reqwest::Client;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
use crate::errors::ChorusResult;
|
use crate::errors::ChorusResult;
|
||||||
use crate::gateway::GatewayCapable;
|
use crate::gateway::{Gateway, GatewayHandle};
|
||||||
use crate::ratelimiter::ChorusRequest;
|
use crate::ratelimiter::ChorusRequest;
|
||||||
use crate::types::types::subconfigs::limits::rates::RateLimits;
|
use crate::types::types::subconfigs::limits::rates::RateLimits;
|
||||||
use crate::types::{GeneralConfiguration, Limit, LimitType, User, UserSettings};
|
use crate::types::{GeneralConfiguration, Limit, LimitType, User, UserSettings};
|
||||||
use crate::{Gateway, GatewayHandle, UrlBundle};
|
use crate::UrlBundle;
|
||||||
|
|
||||||
#[derive(Debug, Clone, Default)]
|
#[derive(Debug, Clone, Default)]
|
||||||
/// The [`Instance`]; what you will be using to perform all sorts of actions on the Spacebar server.
|
/// The [`Instance`]; what you will be using to perform all sorts of actions on the Spacebar server.
|
||||||
|
|
17
src/lib.rs
17
src/lib.rs
|
@ -17,23 +17,6 @@
|
||||||
#[cfg(all(feature = "rt", feature = "rt_multi_thread"))]
|
#[cfg(all(feature = "rt", feature = "rt_multi_thread"))]
|
||||||
compile_error!("feature \"rt\" and feature \"rt_multi_thread\" cannot be enabled at the same time");
|
compile_error!("feature \"rt\" and feature \"rt_multi_thread\" cannot be enabled at the same time");
|
||||||
|
|
||||||
#[cfg(all(not(target_arch = "wasm32"), feature = "client"))]
|
|
||||||
pub type Gateway = DefaultGateway;
|
|
||||||
#[cfg(all(target_arch = "wasm32", feature = "client"))]
|
|
||||||
pub type Gateway = WasmGateway;
|
|
||||||
#[cfg(all(not(target_arch = "wasm32"), feature = "client"))]
|
|
||||||
pub type GatewayHandle = DefaultGatewayHandle;
|
|
||||||
#[cfg(all(target_arch = "wasm32", feature = "client"))]
|
|
||||||
pub type GatewayHandle = WasmGatewayHandle;
|
|
||||||
|
|
||||||
#[cfg(all(not(target_arch = "wasm32"), feature = "client"))]
|
|
||||||
use gateway::DefaultGateway;
|
|
||||||
#[cfg(all(not(target_arch = "wasm32"), feature = "client"))]
|
|
||||||
use gateway::DefaultGatewayHandle;
|
|
||||||
#[cfg(all(target_arch = "wasm32", feature = "client"))]
|
|
||||||
use gateway::WasmGateway;
|
|
||||||
#[cfg(all(target_arch = "wasm32", feature = "client"))]
|
|
||||||
use gateway::WasmGatewayHandle;
|
|
||||||
use url::{ParseError, Url};
|
use url::{ParseError, Url};
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
|
|
|
@ -12,7 +12,10 @@ use crate::types::{
|
||||||
};
|
};
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{types::Composite, GatewayHandle};
|
use crate::types::Composite;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::gateway::Updateable;
|
use crate::gateway::Updateable;
|
||||||
|
|
|
@ -7,7 +7,10 @@ use crate::types::entities::User;
|
||||||
use crate::types::Snowflake;
|
use crate::types::Snowflake;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{types::Composite, GatewayHandle};
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::types::Composite;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::gateway::Updateable;
|
use crate::gateway::Updateable;
|
||||||
|
|
|
@ -16,7 +16,7 @@ use crate::types::{
|
||||||
use super::PublicUser;
|
use super::PublicUser;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{gateway::Updateable, GatewayHandle};
|
use crate::gateway::Updateable;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use chorus_macros::{observe_option_vec, observe_vec, Composite, Updateable};
|
use chorus_macros::{observe_option_vec, observe_vec, Composite, Updateable};
|
||||||
|
@ -24,6 +24,9 @@ use chorus_macros::{observe_option_vec, observe_vec, Composite, Updateable};
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::types::Composite;
|
use crate::types::Composite;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
/// See <https://discord.com/developers/docs/resources/guild>
|
/// See <https://discord.com/developers/docs/resources/guild>
|
||||||
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
|
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
|
||||||
#[cfg_attr(feature = "client", derive(Updateable, Composite))]
|
#[cfg_attr(feature = "client", derive(Updateable, Composite))]
|
||||||
|
|
|
@ -24,7 +24,10 @@ pub use voice_state::*;
|
||||||
pub use webhook::*;
|
pub use webhook::*;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{gateway::Updateable, GatewayHandle};
|
use crate::gateway::Updateable;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
|
|
|
@ -12,7 +12,10 @@ use chorus_macros::{Composite, Updateable};
|
||||||
use crate::gateway::Updateable;
|
use crate::gateway::Updateable;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{types::Composite, GatewayHandle};
|
use crate::types::Composite;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq)]
|
#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq)]
|
||||||
#[cfg_attr(feature = "client", derive(Updateable, Composite))]
|
#[cfg_attr(feature = "client", derive(Updateable, Composite))]
|
||||||
|
|
|
@ -8,7 +8,10 @@ use std::fmt::Debug;
|
||||||
use crate::gateway::Updateable;
|
use crate::gateway::Updateable;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{types::Composite, GatewayHandle};
|
use crate::types::Composite;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use chorus_macros::{Composite, Updateable};
|
use chorus_macros::{Composite, Updateable};
|
||||||
|
|
|
@ -4,7 +4,10 @@ use std::sync::{Arc, RwLock};
|
||||||
use chorus_macros::Composite;
|
use chorus_macros::Composite;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{types::Composite, GatewayHandle};
|
use crate::types::Composite;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::gateway::Updateable;
|
use crate::gateway::Updateable;
|
||||||
|
|
|
@ -10,7 +10,10 @@ use crate::gateway::Updateable;
|
||||||
use chorus_macros::{Composite, Updateable};
|
use chorus_macros::{Composite, Updateable};
|
||||||
|
|
||||||
#[cfg(feature = "client")]
|
#[cfg(feature = "client")]
|
||||||
use crate::{types::Composite, GatewayHandle};
|
use crate::types::Composite;
|
||||||
|
|
||||||
|
#[cfg(feature = "client")]
|
||||||
|
use crate::gateway::GatewayHandle;
|
||||||
|
|
||||||
use crate::types::{
|
use crate::types::{
|
||||||
entities::{Guild, User},
|
entities::{Guild, User},
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, RwLock};
|
||||||
|
|
||||||
use chorus::gateway::{DefaultGateway, GatewayCapable};
|
use chorus::gateway::Gateway;
|
||||||
use chorus::{
|
use chorus::{
|
||||||
instance::{ChorusUser, Instance},
|
instance::{ChorusUser, Instance},
|
||||||
types::{
|
types::{
|
||||||
|
@ -43,7 +43,7 @@ impl TestBundle {
|
||||||
limits: self.user.limits.clone(),
|
limits: self.user.limits.clone(),
|
||||||
settings: self.user.settings.clone(),
|
settings: self.user.settings.clone(),
|
||||||
object: self.user.object.clone(),
|
object: self.user.object.clone(),
|
||||||
gateway: DefaultGateway::spawn(self.instance.urls.wss.clone())
|
gateway: Gateway::spawn(self.instance.urls.wss.clone())
|
||||||
.await
|
.await
|
||||||
.unwrap(),
|
.unwrap(),
|
||||||
}
|
}
|
||||||
|
|
|
@ -2,17 +2,15 @@ mod common;
|
||||||
|
|
||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, RwLock};
|
||||||
|
|
||||||
|
use chorus::gateway::*;
|
||||||
use chorus::types::{self, ChannelModifySchema, RoleCreateModifySchema, RoleObject};
|
use chorus::types::{self, ChannelModifySchema, RoleCreateModifySchema, RoleObject};
|
||||||
use chorus::{gateway::*, GatewayHandle};
|
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
/// Tests establishing a connection (hello and heartbeats) on the local gateway;
|
/// Tests establishing a connection (hello and heartbeats) on the local gateway;
|
||||||
async fn test_gateway_establish() {
|
async fn test_gateway_establish() {
|
||||||
let bundle = common::setup().await;
|
let bundle = common::setup().await;
|
||||||
|
|
||||||
let _: GatewayHandle = DefaultGateway::spawn(bundle.urls.wss.clone())
|
let _: GatewayHandle = Gateway::spawn(bundle.urls.wss.clone()).await.unwrap();
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
common::teardown(bundle).await
|
common::teardown(bundle).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -21,9 +19,7 @@ async fn test_gateway_establish() {
|
||||||
async fn test_gateway_authenticate() {
|
async fn test_gateway_authenticate() {
|
||||||
let bundle = common::setup().await;
|
let bundle = common::setup().await;
|
||||||
|
|
||||||
let gateway: GatewayHandle = DefaultGateway::spawn(bundle.urls.wss.clone())
|
let gateway: GatewayHandle = Gateway::spawn(bundle.urls.wss.clone()).await.unwrap();
|
||||||
.await
|
|
||||||
.unwrap();
|
|
||||||
|
|
||||||
let mut identify = types::GatewayIdentifyPayload::common();
|
let mut identify = types::GatewayIdentifyPayload::common();
|
||||||
identify.token = bundle.user.token.clone();
|
identify.token = bundle.user.token.clone();
|
||||||
|
|
Loading…
Reference in New Issue