rtc_stun/

agent.rs

//! STUN transaction tracking.
//!
//! The agent remembers which requests are outstanding and when each should be considered lost. It
//! performs no I/O: the caller submits [`ClientAgent`](crate::agent::ClientAgent) commands — start a transaction, hand over
//! an inbound message, advance time, stop or close — and polls for the resulting [`Event`](crate::agent::Event)s.
//!
//! This split is what lets the retransmission schedule be tested without a network, and lets ICE
//! reuse the same transaction bookkeeping for its connectivity checks.
#[cfg(test)]
mod agent_test;

use shared::error::*;
use std::collections::{HashMap, VecDeque};
use std::time::Instant;

use crate::message::*;

/// Agent is low-level abstraction over transaction list that
/// handles concurrency and time outs (via Collect call).
#[derive(Default)]
pub struct Agent {
    /// transactions is map of transactions that are currently
    /// in progress. Event handling is done in such way when
    /// transaction is unregistered before AgentTransaction access,
    /// minimizing mux lock and protecting AgentTransaction from
    /// data races via unexpected concurrent access.
    transactions: HashMap<TransactionId, AgentTransaction>,
    /// all calls are invalid if true
    closed: bool,
    /// events queue
    events_queue: VecDeque<Event>,
}

/// Event is passed to Handler describing the transaction event.
/// Do not reuse outside Handler.
#[derive(Debug)] //Clone
pub struct Event {
    /// The transaction this event belongs to.
    pub id: TransactionId,
    /// What happened.
    pub evt: StunEvent,
}

#[derive(Debug)] //Clone
/// What became of a STUN transaction.
pub enum StunEvent {
    /// The agent was closed, abandoning this transaction.
    AgentClosed,
    /// The transaction was stopped by the caller.
    TransactionStopped,
    /// The transaction timed out with no response.
    TransactionTimeOut,
    /// A response arrived for this transaction.
    Message(Message),
}

/// AgentTransaction represents transaction in progress.
/// Concurrent access is invalid.
pub(crate) struct AgentTransaction {
    id: TransactionId,
    deadline: Instant,
}

/// AGENT_COLLECT_CAP is initial capacity for Agent.Collect slices,
/// sufficient to make function zero-alloc in most cases.
const AGENT_COLLECT_CAP: usize = 100;

/// ClientAgent is Agent implementation that is used by Client to
/// process transactions.
#[derive(Debug)]
pub enum ClientAgent {
    /// Hand an inbound message to the agent for matching against a transaction.
    Process(Message),
    /// Advance time so the agent can expire transactions.
    Collect(Instant),
    /// Register a new transaction with its deadline.
    Start(TransactionId, Instant),
    /// Abandon a transaction without waiting for its deadline.
    Stop(TransactionId),
    /// Close the agent, abandoning every outstanding transaction.
    Close,
}

impl Agent {
    /// new initializes and returns new Agent with provided handler.
    pub fn new() -> Self {
        Agent {
            transactions: HashMap::new(),
            closed: false,
            events_queue: VecDeque::new(),
        }
    }

    /// Applies an agent command: start, stop, process a message, collect timeouts, or close.
    ///
    /// # Errors
    ///
    /// Fails if the agent is closed, or a transaction id is already in use.
    pub fn handle_event(&mut self, client_agent: ClientAgent) -> Result<()> {
        match client_agent {
            ClientAgent::Process(message) => self.process(message),
            ClientAgent::Collect(deadline) => self.collect(deadline),
            ClientAgent::Start(tid, deadline) => self.start(tid, deadline),
            ClientAgent::Stop(tid) => self.stop(tid),
            ClientAgent::Close => self.close(),
        }
    }

    /// When the agent next needs [`ClientAgent::Collect`], or `None` with nothing outstanding.
    pub fn poll_timeout(&mut self) -> Option<Instant> {
        let mut deadline = None;
        for transaction in self.transactions.values() {
            if deadline.is_none() || transaction.deadline < *deadline.as_ref().unwrap() {
                deadline = Some(transaction.deadline);
            }
        }
        deadline
    }

    /// The next transaction event, or `None` when there is nothing to report.
    pub fn poll_event(&mut self) -> Option<Event> {
        self.events_queue.pop_front()
    }

    /// process incoming message, synchronously passing it to handler.
    fn process(&mut self, message: Message) -> Result<()> {
        if self.closed {
            return Err(Error::ErrAgentClosed);
        }

        self.transactions.remove(&message.transaction_id);

        self.events_queue.push_back(Event {
            id: message.transaction_id,
            evt: StunEvent::Message(message),
        });

        Ok(())
    }

    /// close terminates all transactions with ErrAgentClosed and renders Agent to
    /// closed state.
    fn close(&mut self) -> Result<()> {
        if self.closed {
            return Err(Error::ErrAgentClosed);
        }

        for id in self.transactions.keys() {
            self.events_queue.push_back(Event {
                id: *id,
                evt: StunEvent::AgentClosed,
            });
        }
        self.transactions.clear();
        self.closed = true;

        Ok(())
    }

    /// start registers transaction with provided id and deadline.
    /// Could return ErrAgentClosed, ErrTransactionExists.
    ///
    /// Agent handler is guaranteed to be eventually called.
    fn start(&mut self, id: TransactionId, deadline: Instant) -> Result<()> {
        if self.closed {
            return Err(Error::ErrAgentClosed);
        }
        if self.transactions.contains_key(&id) {
            return Err(Error::ErrTransactionExists);
        }

        self.transactions
            .insert(id, AgentTransaction { id, deadline });

        Ok(())
    }

    /// stop stops transaction by id with ErrTransactionStopped, blocking
    /// until handler returns.
    fn stop(&mut self, id: TransactionId) -> Result<()> {
        if self.closed {
            return Err(Error::ErrAgentClosed);
        }

        let v = self.transactions.remove(&id);
        if let Some(t) = v {
            self.events_queue.push_back(Event {
                id: t.id,
                evt: StunEvent::TransactionStopped,
            });
            Ok(())
        } else {
            Err(Error::ErrTransactionNotExists)
        }
    }

    /// collect terminates all transactions that have deadline before provided
    /// time, blocking until all handlers will process ErrTransactionTimeOut.
    /// Will return ErrAgentClosed if agent is already closed.
    ///
    /// It is safe to call Collect concurrently but makes no sense.
    fn collect(&mut self, deadline: Instant) -> Result<()> {
        if self.closed {
            // Doing nothing if agent is closed.
            // All transactions should be already closed
            // during Close() call.
            return Err(Error::ErrAgentClosed);
        }

        let mut to_remove: Vec<TransactionId> = Vec::with_capacity(AGENT_COLLECT_CAP);

        // Adding all transactions with deadline before gc_time
        // to toCall and to_remove slices.
        // No allocs if there are less than AGENT_COLLECT_CAP
        // timed out transactions.
        for (id, t) in &self.transactions {
            if t.deadline < deadline {
                to_remove.push(*id);
            }
        }
        // Un-registering timed out transactions.
        for id in &to_remove {
            self.transactions.remove(id);
        }

        for id in to_remove {
            self.events_queue.push_back(Event {
                id,
                evt: StunEvent::TransactionTimeOut,
            });
        }

        Ok(())
    }
}