pub(crate) mod correlation;
pub(crate) mod counters;
pub(crate) mod keepalive;
pub(crate) mod logs;
pub(crate) mod reader;
pub(crate) mod reconcile;
pub(crate) mod reconnect;
pub(crate) mod slot;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering};
use std::thread::JoinHandle;
use std::time::Duration;
use parking_lot::Mutex;
use crate::error::Result;
use crate::protocol::{FrameType, encode};
use crate::transport::Transport;
use crate::types::{CountersSnapshot, LogLine};
use correlation::PendingEntry;
use counters::Counters;
use reconcile::DesiredState;
use slot::TransportSlot;
pub const DEFAULT_QUERY_TIMEOUT: Duration = Duration::from_secs(1);
pub const DEFAULT_KEEPALIVE_CADENCE: Duration = Duration::from_millis(500);
pub(crate) struct LinkInner {
transport: Arc<TransportSlot>,
write_lock: Arc<Mutex<()>>,
seq: Arc<AtomicU8>,
query_gen: Arc<AtomicU64>,
pending: Arc<Mutex<HashMap<u8, PendingEntry>>>,
logs_rx: flume::Receiver<LogLine>,
desired: Arc<Mutex<DesiredState>>,
counters: Arc<Counters>,
stop: Arc<AtomicBool>,
reconnect_lock: Arc<Mutex<()>>,
query_timeout: Duration,
reader: Option<JoinHandle<()>>,
keepalive: Option<JoinHandle<()>>,
}
impl std::fmt::Debug for LinkInner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LinkInner")
.field("transport", &self.transport)
.field("seq", &self.seq.load(Ordering::Relaxed))
.field("pending", &self.pending.lock().len())
.field("counters", &self.counters.snapshot())
.field("stopped", &self.stop.load(Ordering::Relaxed))
.finish_non_exhaustive()
}
}
impl Drop for LinkInner {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
if let Some(h) = self.reader.take() {
let _ = h.join();
}
if let Some(h) = self.keepalive.take() {
let _ = h.join();
}
}
}
#[derive(Clone, Debug)]
pub(crate) struct Link {
inner: Arc<LinkInner>,
}
impl Link {
pub(crate) fn from_transport(transport: Arc<dyn Transport>) -> Link {
Self::from_transport_with_cadence(transport, DEFAULT_KEEPALIVE_CADENCE)
}
pub(crate) fn from_transport_with_cadence(
transport: Arc<dyn Transport>,
keepalive_cadence: Duration,
) -> Link {
let query_timeout = DEFAULT_QUERY_TIMEOUT;
let pending: Arc<Mutex<HashMap<u8, PendingEntry>>> = Arc::new(Mutex::new(HashMap::new()));
let (logs_tx, logs_rx) = flume::bounded(logs::LOGS_CAPACITY);
let counters = Arc::new(Counters::default());
let stop = Arc::new(AtomicBool::new(false));
let desired = Arc::new(Mutex::new(DesiredState::default()));
let write_lock = Arc::new(Mutex::new(()));
let seq = Arc::new(AtomicU8::new(0));
let query_gen = Arc::new(AtomicU64::new(0));
let transport = Arc::new(TransportSlot::new(transport));
let reconnect_lock = Arc::new(Mutex::new(()));
let reader = reader::spawn_reader(
Arc::clone(&transport),
Arc::clone(&pending),
logs_tx.clone(),
logs_rx.clone(),
Arc::clone(&counters),
Arc::clone(&stop),
reconnect::ReconnectCtx {
transport: Arc::clone(&transport),
write_lock: Arc::clone(&write_lock),
seq: Arc::clone(&seq),
counters: Arc::clone(&counters),
desired: Arc::clone(&desired),
reconnect_lock: Arc::clone(&reconnect_lock),
},
);
let keepalive = keepalive::spawn_keepalive(keepalive::KeepaliveCtx {
transport: Arc::clone(&transport),
write_lock: Arc::clone(&write_lock),
seq: Arc::clone(&seq),
counters: Arc::clone(&counters),
desired: Arc::clone(&desired),
stop: Arc::clone(&stop),
cadence: keepalive_cadence,
});
Link {
inner: Arc::new(LinkInner {
transport,
write_lock,
seq,
query_gen,
pending,
logs_rx,
desired,
counters,
stop,
reconnect_lock,
query_timeout,
reader: Some(reader),
keepalive: Some(keepalive),
}),
}
}
pub(crate) fn next_seq(&self) -> u8 {
self.inner.seq.fetch_add(1, Ordering::Relaxed)
}
pub(crate) fn send_with_seq(&self, seq: u8, ty: FrameType, payload: &[u8]) -> Result<()> {
write_frame(
&self.inner.transport,
&self.inner.write_lock,
&self.inner.counters,
seq,
ty,
payload,
)
}
pub(crate) fn send(&self, ty: FrameType, payload: &[u8]) -> Result<()> {
let seq = self.next_seq();
self.send_with_seq(seq, ty, payload)
}
pub(crate) fn counters(&self) -> CountersSnapshot {
self.inner.counters.snapshot()
}
pub(crate) fn query_timeout_default(&self) -> Duration {
self.inner.query_timeout
}
pub(crate) fn desired(&self) -> &Mutex<DesiredState> {
&self.inner.desired
}
pub(crate) fn logs_rx(&self) -> flume::Receiver<LogLine> {
self.inner.logs_rx.clone()
}
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn transport_slot(&self) -> &Arc<TransportSlot> {
&self.inner.transport
}
#[cfg(feature = "async")]
pub(crate) fn weak(&self) -> std::sync::Weak<LinkInner> {
Arc::downgrade(&self.inner)
}
}
fn write_frame(
transport: &TransportSlot,
write_lock: &Mutex<()>,
counters: &Counters,
seq: u8,
ty: FrameType,
payload: &[u8],
) -> Result<()> {
let frame = encode(ty, seq, payload)?;
let current = transport.current();
{
let _guard = write_lock.lock();
current.write_all(&frame)?;
}
counters.inc_tx();
trace_event!(
target: "medius::transport",
tracing::Level::TRACE,
dir = "tx",
opcode = u8::from(ty),
seq,
len = payload.len(),
);
Ok(())
}