Compare commits

..

13 Commits

Author SHA1 Message Date
kozabrada123 b3c1e37fa4 fix: unused import 2024-01-19 19:47:04 +01:00
kozabrada123 60c0c3c536 committed too much 2024-01-19 19:45:34 +01:00
kozabrada123 a787a989ef update voice heartbeat, fix the new test issue 2024-01-19 19:42:24 +01:00
kozabrada123 2bf022924b merge w/ dev 2024-01-19 19:24:59 +01:00
Flori 011b214ea1
Fix gateway heartbeat on WASM (#460)
It turns out `std::time::Instant::now()` panics WASM (see #459) and
breaks the heartbeat handler.

This pr attempts to fix that by replacing `std::time::Instant` with
`wasmtimer::std::Instant` and `safina_timer::sleep_until` with
`wasmtimer::tokio::sleep_until`.
2024-01-19 17:24:48 +01:00
kozabrada123 8a2bc8287e Revert "Mess w/ the tests to see if it really works"
This reverts commit 8243f103f9.
2024-01-19 16:05:34 +01:00
kozabrada123 8243f103f9 Mess w/ the tests to see if it really works 2024-01-19 15:53:24 +01:00
kozabrada123 34cc344c8d feat: switch safina_timer for tokio, fix sleep duration overflow in examples 2024-01-19 15:48:59 +01:00
kozabrada123 72936d4f21 right 2024-01-19 15:31:40 +01:00
kozabrada123 921a3ef9c0 fix: gateway simple example 2024-01-19 15:21:53 +01:00
kozabrada123 c3017df1c2 fix tests 2024-01-19 15:14:50 +01:00
kozabrada123 e2b69487aa fix error 2024-01-19 15:06:27 +01:00
kozabrada123 d37415fc13 feat: fix heartbeat time on WASM 2024-01-19 14:55:23 +01:00
7 changed files with 73 additions and 36 deletions

25
Cargo.lock generated
View File

@ -233,7 +233,6 @@ dependencies = [
"reqwest", "reqwest",
"rustls", "rustls",
"rustls-native-certs", "rustls-native-certs",
"safina-timer",
"serde", "serde",
"serde-aux", "serde-aux",
"serde_json", "serde_json",
@ -247,6 +246,7 @@ dependencies = [
"wasm-bindgen", "wasm-bindgen",
"wasm-bindgen-futures", "wasm-bindgen-futures",
"wasm-bindgen-test", "wasm-bindgen-test",
"wasmtimer",
"ws_stream_wasm", "ws_stream_wasm",
] ]
@ -1800,15 +1800,6 @@ version = "1.0.16"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f98d2aa92eebf49b69786be48e4477826b256916e84a57ff2a4f21923b48eb4c" checksum = "f98d2aa92eebf49b69786be48e4477826b256916e84a57ff2a4f21923b48eb4c"
[[package]]
name = "safina-timer"
version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1081a264d1a3e81b75c4bcd5696094fb6ce470c2ded14cbd47bcb5229079b9df"
dependencies = [
"once_cell",
]
[[package]] [[package]]
name = "salsa20" name = "salsa20"
version = "0.10.2" version = "0.10.2"
@ -2819,6 +2810,20 @@ dependencies = [
"syn 2.0.48", "syn 2.0.48",
] ]
[[package]]
name = "wasmtimer"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f656cd8858a5164932d8a90f936700860976ec21eb00e0fe2aa8cab13f6b4cf"
dependencies = [
"futures",
"js-sys",
"parking_lot",
"pin-utils",
"slab",
"wasm-bindgen",
]
[[package]] [[package]]
name = "web-sys" name = "web-sys"
version = "0.3.66" version = "0.3.66"

View File

@ -57,7 +57,6 @@ sqlx = { version = "0.7.3", features = [
], optional = true } ], optional = true }
discortp = { version = "0.5.0", optional = true, features = ["rtp", "discord", "demux"] } discortp = { version = "0.5.0", optional = true, features = ["rtp", "discord", "demux"] }
crypto_secretbox = {version = "0.1.1", optional = true} crypto_secretbox = {version = "0.1.1", optional = true}
safina-timer = "0.1.11"
rand = "0.8.5" rand = "0.8.5"
[target.'cfg(not(target_arch = "wasm32"))'.dependencies] [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
@ -75,6 +74,7 @@ getrandom = { version = "0.2.12" }
getrandom = { version = "0.2.12", features = ["js"] } getrandom = { version = "0.2.12", features = ["js"] }
ws_stream_wasm = "0.7.4" ws_stream_wasm = "0.7.4"
wasm-bindgen-futures = "0.4.39" wasm-bindgen-futures = "0.4.39"
wasmtimer = "0.2.0"
[dev-dependencies] [dev-dependencies]
lazy_static = "1.4.0" lazy_static = "1.4.0"

View File

@ -8,6 +8,11 @@ use chorus::{
use std::{sync::Arc, time::Duration}; use std::{sync::Arc, time::Duration};
use tokio::{self}; use tokio::{self};
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::sleep;
// This example creates a simple gateway connection and a basic observer struct // This example creates a simple gateway connection and a basic observer struct
// Due to certain limitations all observers must impl debug // Due to certain limitations all observers must impl debug
@ -54,10 +59,9 @@ async fn main() {
let mut identify = GatewayIdentifyPayload::common(); let mut identify = GatewayIdentifyPayload::common();
identify.token = token; identify.token = token;
gateway.send_identify(identify).await; gateway.send_identify(identify).await;
safina_timer::start_timer_thread();
// Do something on the main thread so we don't quit // Do something on the main thread so we don't quit
loop { loop {
safina_timer::sleep_for(Duration::MAX).await sleep(Duration::from_secs(3600)).await;
} }
} }

View File

@ -3,6 +3,11 @@ use std::time::Duration;
use chorus::gateway::Gateway; use chorus::gateway::Gateway;
use chorus::{self, types::GatewayIdentifyPayload}; use chorus::{self, types::GatewayIdentifyPayload};
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::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
#[tokio::main(flavor = "current_thread")] #[tokio::main(flavor = "current_thread")]
async fn main() { async fn main() {
@ -10,7 +15,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::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
@ -26,10 +31,10 @@ async fn main() {
identify.token = token; identify.token = token;
// Send off the event // Send off the event
safina_timer::start_timer_thread(); gateway.send_identify(identify).await;
// Do something on the main thread so we don't quit // Do something on the main thread so we don't quit
loop { loop {
safina_timer::sleep_for(Duration::MAX).await sleep(Duration::from_secs(3600)).await;
} }
} }

View File

@ -1,9 +1,20 @@
use futures_util::SinkExt; use futures_util::SinkExt;
use log::*; use log::*;
use std::time::{self, Duration, Instant};
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::Instant;
#[cfg(target_arch = "wasm32")]
use wasmtimer::std::Instant;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep_until;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::sleep_until;
use std::time::Duration;
use tokio::sync::mpsc::{Receiver, Sender}; use tokio::sync::mpsc::{Receiver, Sender};
use safina_timer::sleep_until;
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
use tokio::task; use tokio::task;
@ -57,12 +68,10 @@ impl HeartbeatHandler {
mut receive: 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 = 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");
@ -123,7 +132,7 @@ impl HeartbeatHandler {
break; break;
} }
last_heartbeat_timestamp = time::Instant::now(); last_heartbeat_timestamp = Instant::now();
last_heartbeat_acknowledged = false; last_heartbeat_acknowledged = false;
} }
} }

View File

@ -1,12 +1,25 @@
use std::{
sync::Arc,
time::{Duration, Instant},
};
use futures_util::SinkExt; use futures_util::SinkExt;
use log::*; use log::*;
use safina_timer::sleep_until;
use tokio::sync::{mpsc::Sender, Mutex}; #[cfg(not(target_arch = "wasm32"))]
use tokio::time::Instant;
#[cfg(target_arch = "wasm32")]
use wasmtimer::std::Instant;
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep_until;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::sleep_until;
use std::{sync::Arc, time::Duration};
use tokio::sync::{
mpsc::{Receiver, Sender},
Mutex,
};
#[cfg(not(target_arch = "wasm32"))]
use tokio::task;
use crate::{ use crate::{
gateway::heartbeat::HEARTBEAT_ACK_TIMEOUT, gateway::heartbeat::HEARTBEAT_ACK_TIMEOUT,
@ -37,7 +50,7 @@ impl VoiceHeartbeatHandler {
let kill_receive = kill_rc.resubscribe(); let kill_receive = kill_rc.resubscribe();
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
tokio::task::spawn(async move { task::spawn(async move {
Self::heartbeat_task( Self::heartbeat_task(
websocket_tx, websocket_tx,
heartbeat_interval, heartbeat_interval,
@ -73,15 +86,13 @@ impl VoiceHeartbeatHandler {
websocket_tx: Arc<Mutex<Sink>>, websocket_tx: Arc<Mutex<Sink>>,
heartbeat_interval: Duration, heartbeat_interval: Duration,
starting_nonce: u64, starting_nonce: u64,
mut receive: tokio::sync::mpsc::Receiver<VoiceHeartbeatThreadCommunication>, mut receive: Receiver<VoiceHeartbeatThreadCommunication>,
mut kill_receive: tokio::sync::broadcast::Receiver<()>, mut kill_receive: tokio::sync::broadcast::Receiver<()>,
) { ) {
let mut last_heartbeat_timestamp: Instant = Instant::now(); let mut last_heartbeat_timestamp: Instant = Instant::now();
let mut last_heartbeat_acknowledged = true; let mut last_heartbeat_acknowledged = true;
let mut nonce: u64 = starting_nonce; let mut nonce: u64 = starting_nonce;
safina_timer::start_timer_thread();
loop { loop {
if kill_receive.try_recv().is_ok() { if kill_receive.try_recv().is_ok() {
trace!("VGW: Closing heartbeat task"); trace!("VGW: Closing heartbeat task");

View File

@ -12,6 +12,11 @@ use wasm_bindgen_test::*;
#[cfg(target_arch = "wasm32")] #[cfg(target_arch = "wasm32")]
wasm_bindgen_test_configure!(run_in_browser); wasm_bindgen_test_configure!(run_in_browser);
#[cfg(not(target_arch = "wasm32"))]
use tokio::time::sleep;
#[cfg(target_arch = "wasm32")]
use wasmtimer::tokio::sleep;
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)] #[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
#[cfg_attr(not(target_arch = "wasm32"), tokio::test)] #[cfg_attr(not(target_arch = "wasm32"), tokio::test)]
/// Tests establishing a connection (hello and heartbeats) on the local gateway; /// Tests establishing a connection (hello and heartbeats) on the local gateway;
@ -61,11 +66,9 @@ async fn test_gateway_authenticate() {
gateway.send_identify(identify).await; gateway.send_identify(identify).await;
let current_time = std::time::Instant::now();
tokio::select! { tokio::select! {
// Fail, we timed out waiting for it // Fail, we timed out waiting for it
() = safina_timer::sleep_until(current_time + Duration::from_secs(20)) => { () = sleep(Duration::from_secs(20)) => {
println!("Timed out waiting for event, failing.."); println!("Timed out waiting for event, failing..");
assert!(false); assert!(false);
} }