liminal-sdk 0.19.1

//! On-demand, transport-only quiet fences for participant drains.

use alloc::format;
use std::time::Instant;

use liminal::protocol::Frame;

use super::{
    ExchangeError, FrameCorrelation, IO_TIMEOUT, LinkRead, WsLink, decode_response, read_step,
    send_frame,
};
use crate::SdkError;

pub(in crate::remote::websocket::connection) fn try_receive_participant_frame(
    link: &mut WsLink,
) -> Result<Option<Frame>, ExchangeError> {
    let pending = link.pump_fences_pending.checked_add(1).ok_or_else(|| {
        ExchangeError::Local(SdkError::Protocol {
            description: "participant pump fence count exhausted".into(),
        })
    })?;
    link.pump_diagnostics.pumps_run = link.pump_diagnostics.pumps_run.saturating_add(1);
    let started = Instant::now();
    send_frame(link, &Frame::Ping { flags: 0 })?;
    link.pump_fences_pending = pending;
    loop {
        let remaining = IO_TIMEOUT.saturating_sub(started.elapsed());
        if remaining.is_zero() {
            return Err(ExchangeError::Local(SdkError::Connection {
                description: format!(
                    "participant pump liveness fence was unanswered after {:.3}s",
                    started.elapsed().as_secs_f64()
                ),
            }));
        }
        match read_step(link, Some(remaining))? {
            LinkRead::PumpFence if link.pump_fences_pending == 0 => return Ok(None),
            LinkRead::PumpFence => {}
            LinkRead::TimedOut => {
                link.pump_diagnostics.reads_timed_out =
                    link.pump_diagnostics.reads_timed_out.saturating_add(1);
            }
            LinkRead::Frame { bytes, correlation } => match correlation {
                FrameCorrelation::UnsolicitedDelivery => {}
                FrameCorrelation::CorrelatedResponse | FrameCorrelation::UnsolicitedFrame => {
                    return decode_response(link, &bytes).map(Some);
                }
            },
        }
    }
}