Streams-R-Us
/
atem-connection-rs
3
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

use Instant over SystemTime

This commit is contained in:
Sam W 2025-06-19 00:25:56 +01:00
parent fe7243e359
commit 3e1ecb0df1
1 changed files with 22 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
atem-connection-rs/src/atem_lib/atem_socket.rs
View File

@ -4,7 +4,7 @@ use std::{
io,
net::SocketAddr,
sync::Arc,
time::{Duration, SystemTime},
time::{Duration, Instant},
};
use itertools::Itertools;
@ -25,10 +25,10 @@ use crate::{
use super::atem_packet::PacketFlag;
const IN_FLIGHT_TIMEOUT: u64 = 60;
const CONNECTION_TIMEOUT: u64 = 5000;
const CONNECTION_RETRY_INTERVAL: u64 = 1000;
const RETRANSMIT_CHECK_INTERVAL: u64 = 1000;
const IN_FLIGHT_TIMEOUT: Duration = Duration::from_millis(60);
const CONNECTION_TIMEOUT: Duration = Duration::from_millis(5000);
const CONNECTION_RETRY_INTERVAL: Duration = Duration::from_millis(1000);
const RETRANSMIT_CHECK_INTERVAL: Duration = Duration::from_millis(1000);
const MAX_PACKET_RETRIES: u16 = 10;
const MAX_PACKET_ID: u16 = 1 << 15;
const MAX_PACKET_PER_ACK: u16 = 16;
@ -93,8 +93,8 @@ impl AtemSocketCommand {
pub struct AtemSocket {
connection_state: ConnectionState,
reconnect_timer: Option<SystemTime>,
retransmit_timer: Option<SystemTime>,
reconnect_timer: Option<Instant>,
retransmit_timer: Option<Instant>,
next_tracking_id: u64,
@ -106,10 +106,10 @@ pub struct AtemSocket {
protocol_version: ProtocolVersion,
last_received_at: SystemTime,
last_received_at: Instant,
last_received_packed_id: u16,
in_flight: Vec<InFlightPacket>,
ack_timer: Option<SystemTime>,
ack_timer: Option<Instant>,
received_without_ack: u16,
atem_message_rx: tokio::sync::mpsc::Receiver<AtemSocketMessage>,
@ -142,7 +142,7 @@ struct InFlightPacket {
packet_id: u16,
tracking_id: u64,
payload: Vec<u8>,
pub last_sent: SystemTime,
pub last_sent: Instant,
pub resent: u16,
}
@ -171,7 +171,7 @@ impl AtemSocket {
protocol_version: ProtocolVersion::V7_2,
last_received_at: SystemTime::now(),
last_received_at: Instant::now(),
last_received_packed_id: 0,
in_flight: vec![],
ack_timer: None,
@ -321,7 +321,7 @@ impl AtemSocket {
packet_id,
tracking_id,
payload: buffer,
last_sent: SystemTime::now(),
last_sent: Instant::now(),
resent: 0,
})
}
@ -339,17 +339,15 @@ impl AtemSocket {
}
}
if let Some(ack_time) = self.ack_timer {
if ack_time <= SystemTime::now() {
if ack_time <= Instant::now() {
self.ack_timer = None;
self.received_without_ack = 0;
self.send_ack(self.last_received_packed_id).await;
}
}
if let Some(reconnect_time) = self.reconnect_timer {
if reconnect_time <= SystemTime::now() {
if self.last_received_at + Duration::from_millis(CONNECTION_TIMEOUT)
<= SystemTime::now()
{
if reconnect_time <= Instant::now() {
if self.last_received_at + CONNECTION_TIMEOUT <= Instant::now() {
log::debug!("{:?}", self.last_received_at);
log::debug!("Connection timed out, restarting");
self.restart_connection().await;
@ -358,7 +356,7 @@ impl AtemSocket {
}
}
if let Some(retransmit_time) = self.retransmit_timer {
if retransmit_time <= SystemTime::now() {
if retransmit_time <= Instant::now() {
self.check_for_retransmit().await;
self.start_retransmit_timer();
}
@ -397,7 +395,7 @@ impl AtemSocket {
atem_packet.raw_fields().map(|f| f.r#type).join(",")
);
self.last_received_at = SystemTime::now();
self.last_received_at = Instant::now();
self.session_id = atem_packet.session_id();
// TODO: naming seems rather off here
@ -476,7 +474,7 @@ impl AtemSocket {
self.ack_timer = None;
self.send_ack(self.last_received_packed_id).await;
} else if self.ack_timer.is_none() {
self.ack_timer = Some(SystemTime::now() + Duration::from_millis(5));
self.ack_timer = Some(Instant::now() + Duration::from_millis(5));
}
}
@ -509,7 +507,7 @@ impl AtemSocket {
if sent_packet.packet_id == from_id
|| !self.is_packet_covered_by_ack(from_id, sent_packet.packet_id)
{
sent_packet.last_sent = SystemTime::now();
sent_packet.last_sent = Instant::now();
sent_packet.resent += 1;
self.send_packet(&sent_packet.payload).await;
@ -523,8 +521,7 @@ impl AtemSocket {
async fn check_for_retransmit(&mut self) {
for sent_packet in self.in_flight.clone() {
if sent_packet.last_sent + Duration::from_millis(IN_FLIGHT_TIMEOUT) < SystemTime::now()
{
if sent_packet.last_sent + IN_FLIGHT_TIMEOUT < Instant::now() {
if sent_packet.resent <= MAX_PACKET_RETRIES
&& self
.is_packet_covered_by_ack(self.next_send_packet_id, sent_packet.packet_id)
@ -580,13 +577,11 @@ impl AtemSocket {
}
fn start_reconnect_timer(&mut self) {
self.reconnect_timer =
Some(SystemTime::now() + Duration::from_millis(CONNECTION_RETRY_INTERVAL));
self.reconnect_timer = Some(Instant::now() + CONNECTION_RETRY_INTERVAL);
}
fn start_retransmit_timer(&mut self) {
self.retransmit_timer =
Some(SystemTime::now() + Duration::from_millis(RETRANSMIT_CHECK_INTERVAL));
self.retransmit_timer = Some(Instant::now() + RETRANSMIT_CHECK_INTERVAL);
}
fn next_packet_tracking_id(&mut self) -> u64 {