Replace tokio::time with safina-timer

This commit is contained in:
bitfl0wer 2023-11-13 19:20:38 +01:00
parent cafd645dd3
commit 1450546429
No known key found for this signature in database
GPG Key ID: 0ACD574FCF5226CF
3 changed files with 16 additions and 14 deletions

17
Cargo.lock generated
View File

@ -197,7 +197,6 @@ dependencies = [
"chorus-macros", "chorus-macros",
"chrono", "chrono",
"custom_error", "custom_error",
"futures-timer",
"futures-util", "futures-util",
"getrandom", "getrandom",
"hostname", "hostname",
@ -213,6 +212,7 @@ dependencies = [
"rustls", "rustls",
"rustls-native-certs", "rustls-native-certs",
"rusty-hook", "rusty-hook",
"safina-timer",
"serde", "serde",
"serde-aux", "serde-aux",
"serde_json", "serde_json",
@ -624,12 +624,6 @@ version = "0.3.29"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "efd193069b0ddadc69c46389b740bbccdd97203899b48d09c5f7969591d6bae2" checksum = "efd193069b0ddadc69c46389b740bbccdd97203899b48d09c5f7969591d6bae2"
[[package]]
name = "futures-timer"
version = "3.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e64b03909df88034c26dc1547e8970b91f98bdb65165d6a4e9110d94263dbb2c"
[[package]] [[package]]
name = "futures-util" name = "futures-util"
version = "0.3.29" version = "0.3.29"
@ -1714,6 +1708,15 @@ version = "1.0.15"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ad4cc8da4ef723ed60bced201181d83791ad433213d8c24efffda1eec85d741" checksum = "1ad4cc8da4ef723ed60bced201181d83791ad433213d8c24efffda1eec85d741"
[[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 = "schannel" name = "schannel"
version = "0.1.22" version = "0.1.22"

View File

@ -52,7 +52,7 @@ sqlx = { version = "0.7.1", features = [
"runtime-tokio-native-tls", "runtime-tokio-native-tls",
"any", "any",
], optional = true } ], optional = true }
futures-timer = "3.0.2" 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]

View File

@ -12,8 +12,7 @@ use std::any::Any;
use std::collections::HashMap; use std::collections::HashMap;
use std::fmt::Debug; use std::fmt::Debug;
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::time::Duration; use std::time::{self, Duration};
use tokio::time::sleep_until;
use futures_util::stream::SplitSink; use futures_util::stream::SplitSink;
use futures_util::stream::SplitStream; use futures_util::stream::SplitStream;
@ -25,8 +24,6 @@ use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex; use tokio::sync::Mutex;
use tokio::task; use tokio::task;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use tokio::time;
use tokio::time::Instant;
use tokio_tungstenite::MaybeTlsStream; use tokio_tungstenite::MaybeTlsStream;
use tokio_tungstenite::{connect_async_tls_with_config, Connector, WebSocketStream}; use tokio_tungstenite::{connect_async_tls_with_config, Connector, WebSocketStream};
@ -798,10 +795,12 @@ impl HeartbeatHandler {
mut receive: tokio::sync::mpsc::Receiver<HeartbeatThreadCommunication>, mut receive: tokio::sync::mpsc::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: time::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");
@ -818,7 +817,7 @@ impl HeartbeatHandler {
let mut should_send = false; let mut should_send = false;
tokio::select! { tokio::select! {
() = sleep_until(last_heartbeat_timestamp + timeout) => { () = safina_timer::sleep_until(last_heartbeat_timestamp + timeout) => {
should_send = true; should_send = true;
} }
Some(communication) = receive.recv() => { Some(communication) = receive.recv() => {