pub mod conference;
mod loopback;
pub mod media;
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use freeswitch_types::variables::VariableName;
use freeswitch_types::{BridgeDialString, CallDirection, DialString};
use crate::line::parse_line;
use crate::message::{classify_message, MessageKind};
use crate::stream::{Block, LogEntry, LogStream, ParseStats, UnclassifiedLine};
use conference::{ConferenceEvent, ConferenceMembership, ConferenceRegistry};
use media::SessionMedia;
type SessionHook = Box<dyn Fn(&LogEntry, &mut SessionState) + Send>;
#[derive(Debug, Clone, Default)]
pub struct SessionState {
pub channel_name: Option<String>,
pub channel_state: Option<String>,
pub initial_context: Option<String>,
pub initial_destination: Option<String>,
pub dialplan_context: Option<String>,
pub dialplan_from: Option<String>,
pub dialplan_to: Option<String>,
pub call_direction: Option<CallDirection>,
pub caller_id_number: Option<String>,
pub caller_id_name: Option<String>,
pub destination_number: Option<String>,
pub hangup_cause: Option<String>,
pub answered_at: Option<String>,
pub other_leg_uuid: Option<String>,
pub conference: Option<ConferenceMembership>,
pub media: SessionMedia,
pub(crate) pending_bridge_target: Option<String>,
pub variables: HashMap<String, String>,
}
#[derive(Default)]
struct IndexedFieldChanges {
channel_name: Option<(Option<String>, Option<String>)>,
pending_bridge_target: Option<(Option<String>, Option<String>)>,
other_leg_uuid: Option<(Option<String>, Option<String>)>,
conference: Option<(Option<ConferenceMembership>, Option<ConferenceMembership>)>,
}
#[derive(Default)]
struct IndexedFields {
channel_name: Option<String>,
pending_bridge_target: Option<String>,
other_leg_uuid: Option<String>,
conference: Option<ConferenceMembership>,
}
impl IndexedFields {
fn of(state: &SessionState) -> Self {
IndexedFields {
channel_name: state.channel_name.clone(),
pending_bridge_target: state.pending_bridge_target.clone(),
other_leg_uuid: state.other_leg_uuid.clone(),
conference: state.conference.clone(),
}
}
}
impl IndexedFieldChanges {
fn diff(old: IndexedFields, state: &SessionState) -> Self {
let IndexedFields {
channel_name: old_channel_name,
pending_bridge_target: old_pending_bridge_target,
other_leg_uuid: old_other_leg_uuid,
conference: old_conference,
} = old;
let mut changes = IndexedFieldChanges::default();
if state.conference != old_conference {
changes.conference = Some((old_conference, state.conference.clone()));
}
if state.channel_name != old_channel_name {
changes.channel_name = Some((old_channel_name, state.channel_name.clone()));
}
if state.pending_bridge_target != old_pending_bridge_target {
changes.pending_bridge_target = Some((
old_pending_bridge_target,
state.pending_bridge_target.clone(),
));
}
if state.other_leg_uuid != old_other_leg_uuid {
changes.other_leg_uuid = Some((old_other_leg_uuid, state.other_leg_uuid.clone()));
}
changes
}
}
#[derive(Debug, Clone)]
pub struct SessionSnapshot {
pub channel_name: Option<String>,
pub channel_state: Option<String>,
pub initial_context: Option<String>,
pub initial_destination: Option<String>,
pub dialplan_context: Option<String>,
pub dialplan_from: Option<String>,
pub dialplan_to: Option<String>,
pub call_direction: Option<CallDirection>,
pub caller_id_number: Option<String>,
pub caller_id_name: Option<String>,
pub destination_number: Option<String>,
pub hangup_cause: Option<String>,
pub answered_at: Option<String>,
pub other_leg_uuid: Option<String>,
pub conference: Option<ConferenceMembership>,
pub media: SessionMedia,
}
impl SessionState {
pub fn variable<V: VariableName>(&self, var: V) -> Option<&str> {
self.variables.get(var.as_str()).map(String::as_str)
}
fn snapshot(&self) -> SessionSnapshot {
SessionSnapshot {
channel_name: self.channel_name.clone(),
channel_state: self.channel_state.clone(),
initial_context: self.initial_context.clone(),
initial_destination: self.initial_destination.clone(),
dialplan_context: self.dialplan_context.clone(),
dialplan_from: self.dialplan_from.clone(),
dialplan_to: self.dialplan_to.clone(),
call_direction: self.call_direction,
caller_id_number: self.caller_id_number.clone(),
caller_id_name: self.caller_id_name.clone(),
destination_number: self.destination_number.clone(),
hangup_cause: self.hangup_cause.clone(),
answered_at: self.answered_at.clone(),
other_leg_uuid: self.other_leg_uuid.clone(),
conference: self.conference.clone(),
media: self.media.clone(),
}
}
fn update_from_entry(&mut self, entry: &LogEntry) {
let block_has_channel_data = matches!(entry.block, Some(Block::ChannelData { .. }));
if let Some(Block::ChannelData { fields, variables }) = &entry.block {
for (name, value) in fields {
match name.as_str() {
"Channel-Name" => self.channel_name = Some(value.clone()),
"Channel-State" => self.channel_state = Some(value.clone()),
"Call-Direction" => {
self.call_direction = CallDirection::from_str(value).ok();
}
"Caller-Caller-ID-Number" => {
self.caller_id_number = Some(value.clone());
}
"Caller-Caller-ID-Name" => {
self.caller_id_name = Some(value.clone());
}
"Caller-Destination-Number" => {
self.destination_number = Some(value.clone());
}
"Other-Leg-Unique-ID" => {
self.other_leg_uuid = Some(value.clone());
}
_ => {}
}
}
for (name, value) in variables {
let var_name = name.strip_prefix("variable_").unwrap_or(name);
self.variables.insert(var_name.to_string(), value.clone());
}
}
match &entry.message_kind {
MessageKind::Execute {
application,
arguments,
..
} => match application.as_str() {
"set" | "export" => {
if let Some((name, value)) = arguments.split_once('=') {
self.variables.insert(name.to_string(), value.to_string());
}
}
"bridge" => {
if let Some(info) = parse_bridge_args(arguments) {
if let Some(uuid) = &info.origination_uuid {
self.other_leg_uuid = Some(uuid.clone());
}
self.pending_bridge_target = Some(info.target_channel);
}
}
_ => {}
},
MessageKind::ChannelLifecycle { detail } => {
if let Some(name) = parse_new_channel(detail) {
if self.channel_name.is_none() {
self.channel_name = Some(name);
}
}
if let Some(cause) = parse_hangup(detail) {
self.hangup_cause = Some(cause);
}
if is_answered(detail) && self.answered_at.is_none() {
self.answered_at = Some(entry.timestamp.clone());
}
}
kind => self.apply_kind(kind),
}
self.apply_processing(&entry.message);
self.media.update_from_entry(entry);
for attached in &entry.attached {
let parsed = parse_line(attached);
self.update_from_message(parsed.message, block_has_channel_data);
}
}
fn apply_kind(&mut self, kind: &MessageKind) {
match kind {
MessageKind::Dialplan { detail, .. } => {
if let Some(dp) = parse_dialplan_context(detail) {
self.initial_context.get_or_insert(dp.context.clone());
self.dialplan_context = Some(dp.context);
self.dialplan_from = Some(dp.from);
self.dialplan_to = Some(dp.to);
}
}
MessageKind::Variable { name, value } => {
let var_name = name.strip_prefix("variable_").unwrap_or(name);
self.variables.insert(var_name.to_string(), value.clone());
}
MessageKind::ChannelField { name, value } => match name.as_str() {
"Channel-Name" => self.channel_name = Some(value.clone()),
"Channel-State" => self.channel_state = Some(value.clone()),
_ => {}
},
MessageKind::StateChange { detail } => {
if let Some(new_state) = parse_state_change(detail) {
self.channel_state = Some(new_state);
}
}
_ => {}
}
}
fn apply_processing(&mut self, msg: &str) {
if msg.contains("Processing ") && msg.contains(" in context ") {
if let Some(dp) = parse_processing_line(msg) {
self.initial_context.get_or_insert(dp.context.clone());
self.initial_destination.get_or_insert(dp.to.clone());
self.dialplan_context = Some(dp.context);
self.dialplan_from = Some(dp.from);
self.dialplan_to = Some(dp.to);
}
}
}
fn update_from_message(&mut self, msg: &str, block_provides_channel_data: bool) {
let kind = classify_message(msg);
match &kind {
MessageKind::Variable { .. } | MessageKind::ChannelField { .. }
if block_provides_channel_data => {}
kind => self.apply_kind(kind),
}
self.apply_processing(msg);
}
}
struct DialplanContext {
from: String,
to: String,
context: String,
}
fn parse_dialplan_context(detail: &str) -> Option<DialplanContext> {
if !detail.starts_with("parsing [") {
return None;
}
let rest = &detail["parsing [".len()..];
let bracket_end = rest.find(']')?;
let inner = &rest[..bracket_end];
let arrow = inner.find("->")?;
let from_part = &inner[..arrow];
let to_part = &inner[arrow + 2..];
Some(DialplanContext {
from: from_part.to_string(),
to: to_part.to_string(),
context: from_part.to_string(),
})
}
fn parse_processing_line(msg: &str) -> Option<DialplanContext> {
let proc_idx = msg.find("Processing ")?;
let after_proc = &msg[proc_idx + "Processing ".len()..];
let ctx_idx = after_proc.rfind(" in context ")?;
let head = &after_proc[..ctx_idx];
let context = after_proc[ctx_idx + " in context ".len()..]
.split_whitespace()
.next()?;
let (from, to) = match head.rfind(">->") {
Some(i) => (&head[..i + 1], &head[i + ">->".len()..]),
None => {
let i = head.rfind("->")?;
(&head[..i], &head[i + "->".len()..])
}
};
Some(DialplanContext {
from: from.to_string(),
to: to.to_string(),
context: context.to_string(),
})
}
fn parse_new_channel(detail: &str) -> Option<String> {
let rest = detail.strip_prefix("New Channel ")?;
let bracket = rest.rfind(" [")?;
Some(rest[..bracket].to_string())
}
fn parse_state_change(detail: &str) -> Option<String> {
let arrow = detail.find(" -> ")?;
Some(detail[arrow + 4..].trim().to_string())
}
fn parse_hangup(detail: &str) -> Option<String> {
if !detail.contains("Hangup ") {
return None;
}
let start = detail.rfind('[')?;
let end = detail[start..].find(']')?;
Some(detail[start + 1..start + end].to_string())
}
fn is_answered(detail: &str) -> bool {
detail.contains("has been answered")
}
pub fn parse_bridge_args(arguments: &str) -> Option<BridgeInfo> {
let dial = BridgeDialString::from_str(arguments).ok()?;
let first_ep = dial.groups().first()?.first()?;
let origination_uuid = first_ep
.variables()
.and_then(|v| v.get("origination_uuid"))
.or_else(|| dial.variables().and_then(|v| v.get("origination_uuid")))
.map(|s| s.to_string());
let mut bare = first_ep.clone();
bare.set_variables(None);
let target_channel = bare.to_string();
Some(BridgeInfo {
origination_uuid,
target_channel,
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct BridgeInfo {
pub origination_uuid: Option<String>,
pub target_channel: String,
}
fn parse_originate_success(msg: &str) -> Option<String> {
let marker = "Peer UUID: ";
let idx = msg.find(marker)?;
let uuid = msg[idx + marker.len()..].trim();
if uuid.is_empty() {
None
} else {
Some(uuid.to_string())
}
}
fn parse_originate_channel(msg: &str) -> Option<&str> {
let start = msg.find(" [")? + 2;
let end = msg[start..].find(']')?;
let chan = &msg[start..start + end];
if chan.is_empty() {
None
} else {
Some(chan)
}
}
fn is_terminal_channel_state(state: Option<&str>) -> bool {
matches!(
state,
Some("CS_HANGUP" | "CS_REPORTING" | "CS_DESTROY" | "CS_NONE" | "HANGUP")
)
}
#[derive(Debug)]
pub struct EnrichedEntry {
pub entry: LogEntry,
pub session: Option<SessionSnapshot>,
}
pub struct SessionTracker<I> {
inner: LogStream<I>,
sessions: HashMap<String, SessionState>,
by_channel_name: HashMap<String, HashSet<String>>,
by_pending_target: HashMap<String, String>,
by_other_leg: HashMap<String, String>,
conferences: ConferenceRegistry,
pre_hook: Option<SessionHook>,
post_hook: Option<SessionHook>,
}
impl<I: Iterator<Item = String>> SessionTracker<I> {
pub fn new(inner: LogStream<I>) -> Self {
SessionTracker {
inner,
sessions: HashMap::new(),
by_channel_name: HashMap::new(),
by_pending_target: HashMap::new(),
by_other_leg: HashMap::new(),
conferences: ConferenceRegistry::default(),
pre_hook: None,
post_hook: None,
}
}
pub fn with_pre_hook<F>(mut self, hook: F) -> Self
where
F: Fn(&LogEntry, &mut SessionState) + Send + 'static,
{
self.pre_hook = Some(Box::new(hook));
self
}
pub fn with_post_hook<F>(mut self, hook: F) -> Self
where
F: Fn(&LogEntry, &mut SessionState) + Send + 'static,
{
self.post_hook = Some(Box::new(hook));
self
}
pub fn sessions(&self) -> &HashMap<String, SessionState> {
&self.sessions
}
pub fn conference_members<'a>(&'a self, instance: &'a str) -> impl Iterator<Item = &'a str> {
self.conferences.members(instance)
}
pub fn remove_session(&mut self, uuid: &str) -> Option<SessionState> {
let state = self.sessions.remove(uuid)?;
if let Some(chan) = &state.channel_name {
if let Some(set) = self.by_channel_name.get_mut(chan) {
set.remove(uuid);
if set.is_empty() {
self.by_channel_name.remove(chan);
}
}
}
if let Some(target) = &state.pending_bridge_target {
self.by_pending_target.remove(target);
}
if let Some(other) = &state.other_leg_uuid {
self.by_other_leg.remove(other);
}
if let Some(conf) = &state.conference {
self.conferences.leave(&conf.name, uuid);
}
Some(state)
}
pub fn stats(&self) -> &ParseStats {
self.inner.stats()
}
pub fn drain_unclassified(&mut self) -> Vec<UnclassifiedLine> {
self.inner.drain_unclassified()
}
fn apply_index_changes(&mut self, uuid: &str, changes: &IndexedFieldChanges) {
if let Some((old, new)) = &changes.channel_name {
if let Some(old_name) = old {
if let Some(set) = self.by_channel_name.get_mut(old_name) {
set.remove(uuid);
if set.is_empty() {
self.by_channel_name.remove(old_name);
}
}
}
if let Some(new_name) = new {
self.by_channel_name
.entry(new_name.clone())
.or_default()
.insert(uuid.to_string());
}
}
if let Some((old, new)) = &changes.pending_bridge_target {
if let Some(old_target) = old {
self.by_pending_target.remove(old_target);
}
if let Some(new_target) = new {
self.by_pending_target
.insert(new_target.clone(), uuid.to_string());
}
}
if let Some((old, new)) = &changes.other_leg_uuid {
match new {
Some(new_leg) => self.index_other_leg(uuid, old.clone(), new_leg),
None => {
if let Some(old_leg) = old {
self.by_other_leg.remove(old_leg);
}
}
}
}
if let Some((old, new)) = &changes.conference {
let same_instance =
matches!((old, new), (Some(o), Some(n)) if o.instance == n.instance);
if !same_instance {
if let Some(old_conf) = old {
self.conferences.leave(&old_conf.name, uuid);
}
if let Some(new_conf) = new {
self.conferences
.join(&new_conf.name, &new_conf.instance, uuid);
}
}
}
}
fn index_other_leg(&mut self, uuid: &str, old_leg: Option<String>, new_leg: &str) {
if let Some(old) = old_leg {
if old != new_leg {
self.by_other_leg.remove(&old);
}
}
self.by_other_leg
.insert(new_leg.to_string(), uuid.to_string());
}
fn update_conference(&mut self, uuid: &str, entry: &LogEntry) {
let target = match conference::detect(entry) {
Some(ConferenceEvent::Leave) => {
if let Some(state) = self.sessions.get_mut(uuid) {
state.conference = None;
}
return;
}
Some(ConferenceEvent::Join(target)) => Some(target),
None => None,
};
let Some(state) = self.sessions.get(uuid) else {
return;
};
let Some(target) = target.or_else(|| conference::target_from_variables(&state.variables))
else {
if let Some(state) = self.sessions.get_mut(uuid) {
let SessionState {
conference,
variables,
..
} = state;
if let Some(membership) = conference {
conference::refresh(membership, variables);
}
}
return;
};
let instance = match state.conference.as_ref() {
Some(current) if current.name == target.name => current.instance.clone(),
_ => self
.conferences
.instance_for(&target.name)
.map(str::to_string)
.unwrap_or_else(|| uuid.to_string()),
};
let Some(state) = self.sessions.get_mut(uuid) else {
return;
};
let SessionState {
conference,
variables,
..
} = state;
let joining_elsewhere = conference.as_ref().is_none_or(|c| c.name != target.name);
if joining_elsewhere {
*conference = Some(ConferenceMembership {
name: target.name,
profile: target.profile.clone(),
instance,
member_id: None,
conference_uuid: None,
});
}
let Some(membership) = conference.as_mut() else {
return;
};
if target.profile.is_some() {
membership.profile = target.profile;
}
conference::refresh(membership, variables);
}
fn loopback_a_leg(&self, b_channel: &str, b_uuid: &str) -> Option<String> {
let a_channel = loopback::a_leg_name(b_channel)?;
let candidates: Vec<&String> = self
.by_channel_name
.get(&a_channel)?
.iter()
.filter(|u| *u != b_uuid)
.filter(|u| {
self.sessions
.get(*u)
.map(|s| !is_terminal_channel_state(s.channel_state.as_deref()))
.unwrap_or(false)
})
.collect();
match candidates.as_slice() {
[a_uuid] => Some((*a_uuid).clone()),
_ => None,
}
}
fn link_legs(&mut self, uuid: &str, entry: &LogEntry) {
if entry.message.contains("Originate Resulted in Success") {
let a_uuid = uuid.to_string();
if let Some(peer_uuid) = parse_originate_success(&entry.message) {
let a_old_pending = self
.sessions
.get(&a_uuid)
.and_then(|s| s.pending_bridge_target.clone());
let mut a_old_leg = None;
if let Some(a_state) = self.sessions.get_mut(&a_uuid) {
a_old_leg = a_state.other_leg_uuid.replace(peer_uuid.clone());
a_state.pending_bridge_target = None;
}
self.index_other_leg(&a_uuid, a_old_leg, &peer_uuid);
if let Some(old_target) = a_old_pending {
self.by_pending_target.remove(&old_target);
}
let b_state = self.sessions.entry(peer_uuid.clone()).or_default();
let b_old_leg = b_state.other_leg_uuid.replace(a_uuid.clone());
self.index_other_leg(&peer_uuid, b_old_leg, &a_uuid);
} else if let Some(chan) = parse_originate_channel(&entry.message) {
let candidates: Vec<String> = self
.by_channel_name
.get(chan)
.map(|set| {
set.iter()
.filter(|u| *u != &a_uuid)
.filter(|u| {
self.sessions
.get(*u)
.map(|s| !is_terminal_channel_state(s.channel_state.as_deref()))
.unwrap_or(false)
})
.cloned()
.collect()
})
.unwrap_or_default();
if let [b_uuid] = candidates.as_slice() {
let b_uuid = b_uuid.clone();
let a_old_pending = self
.sessions
.get(&a_uuid)
.and_then(|s| s.pending_bridge_target.clone());
let mut a_old_leg = None;
if let Some(a_state) = self.sessions.get_mut(&a_uuid) {
a_old_leg = a_state.other_leg_uuid.replace(b_uuid.clone());
a_state.pending_bridge_target = None;
}
let mut b_old_leg = None;
if let Some(b_state) = self.sessions.get_mut(&b_uuid) {
b_old_leg = b_state.other_leg_uuid.replace(a_uuid.clone());
}
self.index_other_leg(&a_uuid, a_old_leg, &b_uuid);
self.index_other_leg(&b_uuid, b_old_leg, &a_uuid);
if let Some(old_target) = a_old_pending {
self.by_pending_target.remove(&old_target);
}
}
}
return;
}
if let MessageKind::ChannelLifecycle { detail } = &entry.message_kind {
if let Some(channel_name) = parse_new_channel(detail) {
let b_uuid = uuid.to_string();
let a_uuid_found = self
.by_other_leg
.get(&b_uuid)
.cloned()
.or_else(|| self.by_pending_target.get(&channel_name).cloned())
.or_else(|| self.loopback_a_leg(&channel_name, &b_uuid))
.filter(|a| a != &b_uuid);
if let Some(a_uuid) = a_uuid_found {
let a_old_pending = self
.sessions
.get(&a_uuid)
.and_then(|s| s.pending_bridge_target.clone());
let mut a_old_leg = None;
if let Some(a_state) = self.sessions.get_mut(&a_uuid) {
a_old_leg = a_state.other_leg_uuid.replace(b_uuid.clone());
a_state.pending_bridge_target = None;
}
let mut b_old_leg = None;
if let Some(b_state) = self.sessions.get_mut(&b_uuid) {
b_old_leg = b_state.other_leg_uuid.replace(a_uuid.clone());
}
self.index_other_leg(&a_uuid, a_old_leg, &b_uuid);
self.index_other_leg(&b_uuid, b_old_leg, &a_uuid);
if let Some(old_target) = a_old_pending {
self.by_pending_target.remove(&old_target);
}
}
}
}
}
}
impl<I: Iterator<Item = String>> Iterator for SessionTracker<I> {
type Item = EnrichedEntry;
fn next(&mut self) -> Option<EnrichedEntry> {
let entry = self.inner.next()?;
if entry.uuid.is_empty() {
return Some(EnrichedEntry {
entry,
session: None,
});
}
let uuid = entry.uuid.clone();
let state = self.sessions.entry(uuid.clone()).or_default();
let old = IndexedFields::of(state);
if let Some(hook) = &self.pre_hook {
hook(&entry, state);
}
state.update_from_entry(&entry);
self.update_conference(&uuid, &entry);
self.link_legs(&uuid, &entry);
if let Some(hook) = &self.post_hook {
let state = self.sessions.entry(uuid.clone()).or_default();
hook(&entry, state);
}
let state = self.sessions.entry(uuid.clone()).or_default();
let changes = IndexedFieldChanges::diff(old, state);
let snapshot = state.snapshot();
self.apply_index_changes(&uuid, &changes);
Some(EnrichedEntry {
entry,
session: Some(snapshot),
})
}
}
#[cfg(test)]
mod tests {
use freeswitch_types::variables::SofiaVariable;
use freeswitch_types::ChannelVariable;
use super::*;
const UUID1: &str = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
const UUID2: &str = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
const UUID3: &str = "c3d4e5f6-a7b8-9012-cdef-234567890123";
const TS1: &str = "2025-01-15 10:30:45.123456";
const TS2: &str = "2025-01-15 10:30:46.234567";
fn full_line(uuid: &str, ts: &str, msg: &str) -> String {
format!("{uuid} {ts} 95.97% [DEBUG] sofia.c:100 {msg}")
}
fn collect_enriched(lines: Vec<String>) -> Vec<EnrichedEntry> {
let stream = LogStream::new(lines.into_iter());
SessionTracker::new(stream).collect()
}
#[test]
fn system_line_no_session() {
let lines = vec![format!(
"{TS1} 95.97% [INFO] mod_event_socket.c:1772 Event Socket command"
)];
let entries = collect_enriched(lines);
assert_eq!(entries.len(), 1);
assert!(entries[0].session.is_none());
}
#[test]
fn dialplan_context_propagation() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 answer"),
format!("{UUID1} Dialplan: sofia/internal/+15550001234@192.0.2.1 parsing [public->global] continue=true"),
full_line(UUID1, TS2, "Some later event"),
];
let entries = collect_enriched(lines);
let last = entries.last().unwrap();
let session = last.session.as_ref().unwrap();
assert_eq!(session.dialplan_context.as_deref(), Some("public"));
assert_eq!(session.dialplan_from.as_deref(), Some("public"));
assert_eq!(session.dialplan_to.as_deref(), Some("global"));
}
#[test]
fn processing_line_extracts_context() {
let lines = vec![full_line(
UUID1,
TS1,
"Processing 5551234567->5559876543 in context public",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(session.dialplan_context.as_deref(), Some("public"));
assert_eq!(session.dialplan_from.as_deref(), Some("5551234567"));
assert_eq!(session.dialplan_to.as_deref(), Some("5559876543"));
}
#[test]
fn initial_context_preserved_across_transfers() {
let lines = vec![
full_line(
UUID1,
TS1,
"Processing 5551234567->5559876543 in context public",
),
full_line(
UUID1,
TS2,
"Processing 5551234567->start_recording in context recordings",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let entries: Vec<_> = tracker.by_ref().collect();
let first = entries[0].session.as_ref().unwrap();
assert_eq!(
first.initial_context.as_deref(),
Some("public"),
"initial_context set on first Processing line"
);
assert_eq!(first.dialplan_context.as_deref(), Some("public"));
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.initial_context.as_deref(),
Some("public"),
"initial_context keeps the first context seen"
);
assert_eq!(
state.dialplan_context.as_deref(),
Some("recordings"),
"dialplan_context tracks the current context"
);
assert_eq!(state.dialplan_to.as_deref(), Some("start_recording"));
}
#[test]
fn new_channel_sets_channel_name() {
let lines = vec![full_line(
UUID1,
TS1,
"New Channel sofia/internal-v4/sos [a1b2c3d4-e5f6-7890-abcd-ef1234567890]",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(
session.channel_name.as_deref(),
Some("sofia/internal-v4/sos")
);
}
#[test]
fn originate_success_links_both_legs() {
let lines = vec![
full_line(UUID2, TS1, "New Channel sofia/esinet1-v6-tcp/sip:target.example.com [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
full_line(UUID1, TS2, "Originate Resulted in Success: [sofia/esinet1-v6-tcp/sip:target.example.com] Peer UUID: b2c3d4e5-f6a7-8901-bcde-f12345678901"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID2),
"A-leg other_leg_uuid set from Originate Resulted in Success"
);
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"B-leg other_leg_uuid points back to A-leg"
);
}
#[test]
fn originate_success_channel_fallback_links_legs() {
let lines = vec![
full_line(
UUID2,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
),
full_line(
UUID1,
TS2,
"Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID2),
"A-leg linked to B-leg via channel-name fallback when Peer UUID absent"
);
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"B-leg linked back to A-leg"
);
}
#[test]
fn originate_success_peer_uuid_wins_over_channel_fallback() {
let lines = vec![
full_line(
UUID2,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
),
full_line(
UUID3,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [c3d4e5f6-a7b8-9012-cdef-234567890123]",
),
full_line(
UUID1,
TS2,
"Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744] Peer UUID: b2c3d4e5-f6a7-8901-bcde-f12345678901",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID2),
"Peer UUID wins over channel-name match"
);
let decoy = tracker.sessions().get(UUID3).unwrap();
assert_eq!(
decoy.other_leg_uuid, None,
"Decoy session sharing channel name is not touched"
);
}
#[test]
fn originate_success_channel_fallback_skips_when_ambiguous() {
let lines = vec![
full_line(
UUID2,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
),
full_line(
UUID3,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [c3d4e5f6-a7b8-9012-cdef-234567890123]",
),
full_line(
UUID1,
TS2,
"Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid, None,
"Ambiguous channel name yields no link"
);
assert_eq!(tracker.sessions().get(UUID2).unwrap().other_leg_uuid, None);
assert_eq!(tracker.sessions().get(UUID3).unwrap().other_leg_uuid, None);
}
#[test]
fn originate_success_channel_fallback_skips_terminated_candidates() {
let lines = vec![
full_line(
UUID2,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
),
full_line(
UUID2,
TS1,
"(sofia/internal/6244@192.0.2.72:50744) State Change CS_EXECUTE -> CS_DESTROY",
),
full_line(
UUID3,
TS1,
"New Channel sofia/internal/6244@192.0.2.72:50744 [c3d4e5f6-a7b8-9012-cdef-234567890123]",
),
full_line(
UUID1,
TS2,
"Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID3),
"Live b-leg wins over CS_DESTROY straggler"
);
let live_b = tracker.sessions().get(UUID3).unwrap();
assert_eq!(
live_b.other_leg_uuid.as_deref(),
Some(UUID1),
"Live b-leg points back to a-leg"
);
let stale_b = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
stale_b.other_leg_uuid, None,
"Terminated b-leg is not touched"
);
}
#[test]
fn originate_success_channel_fallback_skips_when_no_match() {
let lines = vec![full_line(
UUID1,
TS2,
"Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744]",
)];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(a_leg.other_leg_uuid, None);
assert_eq!(a_leg.pending_bridge_target, None);
}
#[test]
fn bridge_origination_uuid_links_a_leg_immediately() {
let lines = vec![
full_line(UUID1, TS1, "EXECUTE [depth=0] sofia/internal-v6/1232@[2001:db8::10] bridge([origination_uuid=b2c3d4e5-f6a7-8901-bcde-f12345678901,leg_timeout=2]sofia/esinet1-v6-tcp/sip:target.example.com)"),
full_line(UUID2, TS1, "New Channel sofia/esinet1-v6-tcp/sip:target.example.com [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID2),
"A-leg knows B-leg UUID from origination_uuid in bridge args"
);
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"B-leg knows A-leg once New Channel correlates"
);
}
#[test]
fn bridge_target_matches_new_channel() {
let lines = vec![
full_line(UUID1, TS1, "EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 bridge(sofia/gateway/carrier/+15559876543)"),
full_line(UUID1, TS1, "Parsing session specific variables"),
full_line(UUID2, TS1, "New Channel sofia/gateway/carrier/+15559876543 [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID2),
"A-leg linked to B-leg via bridge target matching New Channel"
);
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"B-leg linked back to A-leg"
);
}
#[test]
fn originate_success_corrects_wrong_target_match() {
let lines = vec![
full_line(UUID1, TS1, "EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 bridge(sofia/gateway/carrier/+15559876543)"),
full_line(UUID2, TS1, "New Channel sofia/gateway/carrier/+15559876543 [b2c3d4e5-f6a7-8901-bcde-f12345678901]"),
full_line(UUID1, TS2, "Originate Resulted in Success: [sofia/gateway/carrier/+15559876543] Peer UUID: c3d4e5f6-a7b8-9012-cdef-234567890123"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID3),
"Originate success overrides earlier target-match guess"
);
let real_b_leg = tracker.sessions().get(UUID3).unwrap();
assert_eq!(
real_b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"Real B-leg points back to A-leg"
);
}
#[test]
fn channel_data_other_leg_uuid() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Other-Leg-Unique-ID: [{UUID2}]"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.other_leg_uuid.as_deref(),
Some(UUID2),
"other_leg_uuid set from Other-Leg-Unique-ID CHANNEL_DATA field"
);
}
#[test]
fn relink_removes_stale_by_other_leg_entry() {
let lines = vec![
full_line(UUID2, TS1, "CHANNEL_DATA:"),
format!("{UUID2} Other-Leg-Unique-ID: [{UUID3}]"),
full_line(
UUID1,
TS2,
&format!(
"Originate Resulted in Success: [sofia/internal/6244@192.0.2.72:50744] Peer UUID: {UUID2}"
),
),
full_line(
UUID3,
TS2,
&format!("New Channel sofia/external/dest@192.0.2.9 [{UUID3}]"),
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(a_leg.other_leg_uuid.as_deref(), Some(UUID2));
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"authoritative Peer UUID link must survive the unrelated New Channel"
);
let c_leg = tracker.sessions().get(UUID3).unwrap();
assert_eq!(
c_leg.other_leg_uuid, None,
"New Channel on C must not back-link via the superseded index entry"
);
}
#[test]
fn channel_data_populates_session() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
format!("{UUID1} Channel-State: [CS_EXECUTE]"),
"variable_sip_call_id: [test123@192.0.2.1]".to_string(),
"variable_direction: [inbound]".to_string(),
];
let entries = collect_enriched(lines);
assert_eq!(entries.len(), 1);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(
session.channel_name.as_deref(),
Some("sofia/internal/+15550001234@192.0.2.1")
);
assert_eq!(session.channel_state.as_deref(), Some("CS_EXECUTE"));
}
#[test]
fn variables_learned_from_channel_data() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
"variable_sip_call_id: [test123@192.0.2.1]".to_string(),
"variable_direction: [inbound]".to_string(),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.variables.get("sip_call_id").map(|s| s.as_str()),
Some("test123@192.0.2.1")
);
assert_eq!(
state.variables.get("direction").map(|s| s.as_str()),
Some("inbound")
);
}
#[test]
fn typed_variable_accessor() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
"variable_sip_call_id: [test123@192.0.2.1]".to_string(),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.variable(SofiaVariable::SipCallId),
Some("test123@192.0.2.1")
);
assert_eq!(state.variable(ChannelVariable::Direction), None);
}
#[test]
fn multi_line_variable_survives_attached_rescan() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
format!("{UUID1} variable_switch_r_sdp: [v=0"),
"o=FreeSWITCH 1737000000 1737000001 IN IP4 192.0.2.10".to_string(),
"s=FreeSWITCH".to_string(),
"c=IN IP4 192.0.2.10".to_string(),
"m=audio 30000 RTP/AVP 0 101".to_string(),
"]".to_string(),
format!("{UUID1} variable_direction: [inbound]"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
let sdp = state
.variables
.get("switch_r_sdp")
.expect("switch_r_sdp variable present");
assert!(
sdp.contains('\n'),
"expected full reassembled value, got fragment: {sdp:?}"
);
assert!(sdp.starts_with("v=0\n"));
assert!(sdp.contains("m=audio 30000 RTP/AVP 0 101"));
assert_eq!(
state.variables.get("direction").map(|s| s.as_str()),
Some("inbound")
);
assert_eq!(
state.channel_name.as_deref(),
Some("sofia/internal/+15550001234@192.0.2.1")
);
}
#[test]
fn attached_processing_line_updates_context() {
let lines = vec![
full_line(UUID1, TS1, "Ring-Ready sofia/internal-v4/sos!"),
format!(
"{UUID1} Processing Extension 1263 <1263>->start_recording in context recordings"
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(state.dialplan_context.as_deref(), Some("recordings"));
assert_eq!(
state.dialplan_from.as_deref(),
Some("Extension 1263 <1263>")
);
assert_eq!(state.dialplan_to.as_deref(), Some("start_recording"));
assert_eq!(
state.initial_destination.as_deref(),
Some("start_recording")
);
}
#[test]
fn variables_learned_from_set_execute() {
let lines = vec![
full_line(UUID1, TS1, "First"),
format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(call_direction=inbound)"),
full_line(UUID1, TS2, "After set"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let entries: Vec<_> = tracker.by_ref().collect();
assert_eq!(entries.len(), 3);
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.variables.get("call_direction").map(|s| s.as_str()),
Some("inbound")
);
}
#[test]
fn variables_learned_from_export_execute() {
let lines = vec![
full_line(UUID1, TS1, "First"),
format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 export(originate_timeout=3600)"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.variables.get("originate_timeout").map(|s| s.as_str()),
Some("3600")
);
}
#[test]
fn session_isolation_between_uuids() {
let lines = vec![
full_line(
UUID1,
TS1,
"Processing 5551111111->5552222222 in context public",
),
full_line(
UUID2,
TS2,
"Processing 5553333333->5554444444 in context private",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let s1 = tracker.sessions().get(UUID1).unwrap();
let s2 = tracker.sessions().get(UUID2).unwrap();
assert_eq!(s1.dialplan_context.as_deref(), Some("public"));
assert_eq!(s2.dialplan_context.as_deref(), Some("private"));
assert_eq!(s1.dialplan_from.as_deref(), Some("5551111111"));
assert_eq!(s2.dialplan_from.as_deref(), Some("5553333333"));
}
#[test]
fn processing_line_with_regex_type_and_angle_bracket_caller() {
let lines = vec![full_line(
UUID1,
TS1,
"Processing Emergency S R <5550001234>->start_recording in context recordings",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(session.initial_context.as_deref(), Some("recordings"));
assert_eq!(session.dialplan_context.as_deref(), Some("recordings"));
assert_eq!(
session.dialplan_from.as_deref(),
Some("Emergency S R <5550001234>")
);
assert_eq!(session.dialplan_to.as_deref(), Some("start_recording"));
}
#[test]
fn processing_line_extension_format() {
let lines = vec![full_line(
UUID1,
TS1,
"Processing Extension 1263 <1263>->start_recording in context recordings",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(session.initial_context.as_deref(), Some("recordings"));
assert_eq!(
session.dialplan_from.as_deref(),
Some("Extension 1263 <1263>")
);
assert_eq!(session.dialplan_to.as_deref(), Some("start_recording"));
}
#[test]
fn parse_processing_line_anchors_on_last_arrow() {
let dest = |msg: &str| parse_processing_line(msg).map(|dp| dp.to);
assert_eq!(
dest("Processing Anonymous <anonymous>->5550001234 in context public").as_deref(),
Some("5550001234"),
);
assert_eq!(
dest("Processing 5550009999 <5550009999>->5550001234 in context public").as_deref(),
Some("5550001234"),
);
assert_eq!(
dest("Processing Jane Doe <5550009999>->5550001234 in context internal").as_deref(),
Some("5550001234"),
);
assert_eq!(
dest("Processing Weird -> Name <5550009999>->5550001234 in context internal")
.as_deref(),
Some("5550001234"),
);
assert_eq!(
dest("Processing Jane Doe <5550009999>->start_recording in context features")
.as_deref(),
Some("start_recording"),
);
}
#[test]
fn initial_destination_first_wins() {
let lines = vec![
full_line(
UUID1,
TS1,
"Processing Jane Doe <5550009999>->5550001234 in context public",
),
full_line(
UUID1,
TS2,
"Processing Jane Doe <5550009999>->5550001234 in context transit",
),
full_line(
UUID1,
TS2,
"Processing Jane Doe <5550009999>->start_recording in context features",
),
full_line(
UUID1,
TS2,
"Processing Jane Doe <5550009999>->check_end_call in context features",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.initial_destination.as_deref(),
Some("5550001234"),
"initial_destination keeps the dialed number from the first Processing line"
);
assert_eq!(
state.dialplan_to.as_deref(),
Some("check_end_call"),
"dialplan_to is last-wins and gets clobbered by feature-context routing"
);
}
#[test]
fn state_change_updates_channel_state() {
let lines = vec![full_line(UUID1, TS1, "State Change CS_INIT -> CS_ROUTING")];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(session.channel_state.as_deref(), Some("CS_ROUTING"));
}
#[test]
fn callstate_change_updates_channel_state() {
let lines = vec![full_line(
UUID1,
TS1,
"(sofia/internal-v4/sos) Callstate Change DOWN -> RINGING",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(session.channel_state.as_deref(), Some("RINGING"));
}
#[test]
fn state_change_overrides_callstate() {
let lines = vec![
full_line(
UUID1,
TS1,
"(sofia/internal-v4/sos) Callstate Change DOWN -> RINGING",
),
full_line(
UUID1,
TS2,
"(sofia/internal-v4/sos) State Change CS_CONSUME_MEDIA -> CS_EXCHANGE_MEDIA",
),
];
let entries = collect_enriched(lines);
assert_eq!(
entries[0]
.session
.as_ref()
.unwrap()
.channel_state
.as_deref(),
Some("RINGING")
);
assert_eq!(
entries[1]
.session
.as_ref()
.unwrap()
.channel_state
.as_deref(),
Some("CS_EXCHANGE_MEDIA")
);
}
#[test]
fn bleg_lifecycle_extracts_data_from_processing() {
let lines = vec![
full_line(
UUID1,
TS1,
"New Channel sofia/internal-v4/sos [a1b2c3d4-e5f6-7890-abcd-ef1234567890]",
),
full_line(
UUID1,
TS1,
"(sofia/internal-v4/sos) State Change CS_NEW -> CS_INIT",
),
full_line(
UUID1,
TS1,
"(sofia/internal-v4/sos) State Change CS_INIT -> CS_ROUTING",
),
full_line(
UUID1,
TS1,
"(sofia/internal-v4/sos) State Change CS_ROUTING -> CS_CONSUME_MEDIA",
),
full_line(
UUID1,
TS1,
"(sofia/internal-v4/sos) Callstate Change DOWN -> RINGING",
),
full_line(
UUID1,
TS2,
"(sofia/internal-v4/sos) State Change CS_CONSUME_MEDIA -> CS_EXCHANGE_MEDIA",
),
full_line(
UUID1,
TS2,
"Processing Emergency S R <5550001234>->start_recording in context recordings",
),
full_line(
UUID1,
TS2,
"(sofia/internal-v4/sos) State Change CS_EXCHANGE_MEDIA -> CS_HANGUP",
),
];
let entries = collect_enriched(lines);
let after_ringing = entries[4].session.as_ref().unwrap();
assert_eq!(after_ringing.channel_state.as_deref(), Some("RINGING"));
assert!(after_ringing.initial_context.is_none());
let after_processing = entries[6].session.as_ref().unwrap();
assert_eq!(
after_processing.channel_state.as_deref(),
Some("CS_EXCHANGE_MEDIA")
);
assert_eq!(
after_processing.initial_context.as_deref(),
Some("recordings")
);
assert_eq!(
after_processing.dialplan_from.as_deref(),
Some("Emergency S R <5550001234>")
);
assert_eq!(
after_processing.dialplan_to.as_deref(),
Some("start_recording")
);
let after_hangup = entries[7].session.as_ref().unwrap();
assert_eq!(after_hangup.channel_state.as_deref(), Some("CS_HANGUP"));
assert_eq!(after_hangup.initial_context.as_deref(), Some("recordings"));
}
#[test]
fn channel_name_from_new_channel() {
let lines = vec![full_line(
UUID1,
TS1,
"New Channel sofia/internal-v4/sos [a1b2c3d4-e5f6-7890-abcd-ef1234567890]",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(
session.channel_name.as_deref(),
Some("sofia/internal-v4/sos")
);
}
#[test]
fn remove_session() {
let lines = vec![full_line(
UUID1,
TS1,
"Processing 5551111111->5552222222 in context public",
)];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
assert!(tracker.sessions().contains_key(UUID1));
let removed = tracker.remove_session(UUID1).unwrap();
assert_eq!(removed.dialplan_context.as_deref(), Some("public"));
assert!(!tracker.sessions().contains_key(UUID1));
}
#[test]
fn stats_delegation() {
let lines = vec![
full_line(UUID1, TS1, "First"),
full_line(UUID1, TS2, "Second"),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream);
let _: Vec<_> = tracker.by_ref().collect();
assert_eq!(tracker.stats().lines_processed, 2);
}
#[test]
fn snapshot_reflects_cumulative_state() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Channel-Name: [sofia/internal/+15550001234@192.0.2.1]"),
format!("{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(foo=bar)"),
full_line(
UUID1,
TS2,
"Processing 5551111111->5552222222 in context public",
),
];
let entries = collect_enriched(lines);
assert_eq!(entries.len(), 3);
let first = entries[0].session.as_ref().unwrap();
assert_eq!(
first.channel_name.as_deref(),
Some("sofia/internal/+15550001234@192.0.2.1"),
);
assert!(first.dialplan_context.is_none());
let last = entries[2].session.as_ref().unwrap();
assert_eq!(
last.channel_name.as_deref(),
Some("sofia/internal/+15550001234@192.0.2.1"),
);
assert_eq!(last.dialplan_context.as_deref(), Some("public"));
}
#[test]
fn post_hook_sets_other_leg_uuid() {
let lines = vec![
full_line(UUID1, TS1, "First entry"),
format!(
"{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(api_result=+OK {UUID2} Job-UUID: ...)"
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream).with_post_hook(|entry, state| {
if let MessageKind::Execute {
application,
arguments,
..
} = &entry.message_kind
{
if application == "set" {
if let Some(value) = arguments.strip_prefix("api_result=+OK ") {
let uuid = value.split_whitespace().next().unwrap_or("");
if uuid.len() == 36 && state.other_leg_uuid.is_none() {
state.other_leg_uuid = Some(uuid.to_string());
}
}
}
}
});
let entries: Vec<_> = tracker.by_ref().collect();
assert_eq!(entries.len(), 2);
let session = entries[1].session.as_ref().unwrap();
assert_eq!(
session.other_leg_uuid.as_deref(),
Some(UUID2),
"post_hook should detect uuid_bridge API result"
);
}
#[test]
fn post_hook_does_not_override_builtin() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Other-Leg-Unique-ID: [{UUID2}]"),
format!(
"{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(api_result=+OK {UUID3} Job-UUID: ...)"
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream).with_post_hook(|entry, state| {
if let MessageKind::Execute {
application,
arguments,
..
} = &entry.message_kind
{
if application == "set" {
if let Some(value) = arguments.strip_prefix("api_result=+OK ") {
let uuid = value.split_whitespace().next().unwrap_or("");
if uuid.len() == 36 && state.other_leg_uuid.is_none() {
state.other_leg_uuid = Some(uuid.to_string());
}
}
}
}
});
let entries: Vec<_> = tracker.by_ref().collect();
assert_eq!(entries.len(), 2);
let session = entries[1].session.as_ref().unwrap();
assert_eq!(
session.other_leg_uuid.as_deref(),
Some(UUID2),
"built-in Other-Leg-Unique-ID takes precedence over hook"
);
}
#[test]
fn pre_hook_runs_before_builtin() {
let lines = vec![full_line(UUID1, TS1, "State Change CS_INIT -> CS_ROUTING")];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream).with_pre_hook(|_entry, state| {
state.channel_state = Some("PRE_SET".to_string());
});
let entries: Vec<_> = tracker.by_ref().collect();
assert_eq!(
entries[0]
.session
.as_ref()
.unwrap()
.channel_state
.as_deref(),
Some("CS_ROUTING"),
"built-in overwrites pre_hook value when no guard"
);
}
#[test]
fn post_hook_runs_after_builtin() {
let lines = vec![full_line(UUID1, TS1, "State Change CS_INIT -> CS_ROUTING")];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream).with_post_hook(|_entry, state| {
if state.channel_state.as_deref() == Some("CS_ROUTING") {
state
.variables
.insert("routing_seen".to_string(), "true".to_string());
}
});
let _: Vec<_> = tracker.by_ref().collect();
let state = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
state.variables.get("routing_seen").map(|s| s.as_str()),
Some("true"),
"post_hook can read fields set by built-in"
);
}
#[test]
fn post_hook_other_leg_uuid_maintains_index_for_backlink() {
let lines = vec![
full_line(UUID1, TS1, "First entry"),
format!(
"{UUID1} EXECUTE [depth=0] sofia/internal/+15550001234@192.0.2.1 set(api_result=+OK {UUID2})"
),
full_line(
UUID2,
TS2,
"New Channel sofia/internal/target@192.0.2.9 [b2c3d4e5-f6a7-8901-bcde-f12345678901]",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream).with_post_hook(|entry, state| {
if let MessageKind::Execute {
application,
arguments,
..
} = &entry.message_kind
{
if application == "set" {
if let Some(value) = arguments.strip_prefix("api_result=+OK ") {
let uuid = value.split_whitespace().next().unwrap_or("");
if uuid.len() == 36 && state.other_leg_uuid.is_none() {
state.other_leg_uuid = Some(uuid.to_string());
}
}
}
}
});
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(a_leg.other_leg_uuid.as_deref(), Some(UUID2));
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(
b_leg.other_leg_uuid.as_deref(),
Some(UUID1),
"B-leg back-links via by_other_leg index populated by the hook"
);
}
#[test]
fn pre_hook_channel_name_maintains_index_for_originate_fallback() {
let lines = vec![
full_line(UUID2, TS1, "custom-channel-announce sofia/custom/6244"),
full_line(
UUID1,
TS2,
"Originate Resulted in Success: [sofia/custom/6244]",
),
];
let stream = LogStream::new(lines.into_iter());
let mut tracker = SessionTracker::new(stream).with_pre_hook(|entry, state| {
if let Some(chan) = entry.message.strip_prefix("custom-channel-announce ") {
state.channel_name = Some(chan.to_string());
}
});
let _: Vec<_> = tracker.by_ref().collect();
let a_leg = tracker.sessions().get(UUID1).unwrap();
assert_eq!(
a_leg.other_leg_uuid.as_deref(),
Some(UUID2),
"fallback finds hook-named B-leg via by_channel_name index"
);
let b_leg = tracker.sessions().get(UUID2).unwrap();
assert_eq!(b_leg.other_leg_uuid.as_deref(), Some(UUID1));
}
#[test]
fn parse_hangup_extracts_cause() {
assert_eq!(
parse_hangup("Hangup sofia/internal/1234 [NORMAL_CLEARING]"),
Some("NORMAL_CLEARING".to_string())
);
assert_eq!(
parse_hangup("Hangup sofia/internal/1234 [USER_BUSY]"),
Some("USER_BUSY".to_string())
);
assert_eq!(parse_hangup("Some other message"), None);
assert_eq!(parse_hangup("New Channel sofia/internal/1234 [uuid]"), None);
}
#[test]
fn is_answered_detects_answer_event() {
assert!(is_answered("sofia/internal/1234 has been answered"));
assert!(!is_answered("sofia/internal/1234 is ringing"));
assert!(!is_answered("New Channel sofia/internal/1234"));
}
#[test]
fn hangup_cause_from_lifecycle() {
let lines = vec![full_line(
UUID1,
TS1,
"Hangup sofia/internal/+15550001234@192.0.2.1 [NORMAL_CLEARING]",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(
session.hangup_cause.as_deref(),
Some("NORMAL_CLEARING"),
"hangup_cause extracted from ChannelLifecycle Hangup"
);
}
#[test]
fn answered_at_from_lifecycle() {
let lines = vec![full_line(
UUID1,
TS1,
"sofia/internal/+15550001234@192.0.2.1 has been answered",
)];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(
session.answered_at.as_deref(),
Some(TS1),
"answered_at captures timestamp when 'has been answered' seen"
);
}
#[test]
fn answered_at_not_overwritten() {
let lines = vec![
full_line(
UUID1,
TS1,
"sofia/internal/+15550001234@192.0.2.1 has been answered",
),
full_line(
UUID1,
TS2,
"sofia/internal/+15550001234@192.0.2.1 has been answered",
),
];
let entries = collect_enriched(lines);
let session = entries[1].session.as_ref().unwrap();
assert_eq!(
session.answered_at.as_deref(),
Some(TS1),
"answered_at preserves first answer timestamp"
);
}
#[test]
fn caller_id_name_from_channel_data() {
let lines = vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} Caller-Caller-ID-Name: [Test Caller Name]"),
];
let entries = collect_enriched(lines);
let session = entries[0].session.as_ref().unwrap();
assert_eq!(
session.caller_id_name.as_deref(),
Some("Test Caller Name"),
"caller_id_name extracted from CHANNEL_DATA"
);
}
fn track(lines: Vec<String>) -> SessionTracker<std::vec::IntoIter<String>> {
let mut tracker = SessionTracker::new(LogStream::new(lines.into_iter()));
for _ in tracker.by_ref() {}
tracker
}
#[test]
fn conference_execute_shares_one_instance() {
let tracker = track(vec![
format!("{UUID1} EXECUTE [depth=0] loopback/tty-a conference(835)"),
format!("{UUID2} EXECUTE [depth=0] sofia/internal/1000 conference(835)"),
format!("{UUID3} EXECUTE [depth=0] sofia/internal/1001 conference(844)"),
]);
let first = tracker.sessions()[UUID1].conference.clone().unwrap();
let second = tracker.sessions()[UUID2].conference.clone().unwrap();
let other = tracker.sessions()[UUID3].conference.clone().unwrap();
assert_eq!(first.name, "835");
assert_eq!(first.instance, UUID1, "first joiner names the instance");
assert_eq!(second.instance, UUID1);
assert_eq!(other.name, "844");
assert_ne!(other.instance, first.instance);
let mut members: Vec<&str> = tracker.conference_members(&first.instance).collect();
members.sort_unstable();
assert_eq!(members, [UUID1, UUID2]);
}
#[test]
fn conference_transfer_line_joins() {
let tracker = track(vec![full_line(
UUID1,
TS1,
"Transfer loopback/tty-a to inline[conference:835@default]",
)]);
let membership = tracker.sessions()[UUID1].conference.clone().unwrap();
assert_eq!(membership.name, "835");
assert_eq!(
membership.profile, None,
"the transfer context is not a conference profile"
);
}
#[test]
fn conference_variables_fill_member_id() {
let tracker = track(vec![
full_line(UUID1, TS1, "CHANNEL_DATA:"),
format!("{UUID1} variable_conference_name: [835]"),
format!("{UUID1} variable_conference_member_id: [3]"),
format!("{UUID1} variable_conference_uuid: [{UUID2}]"),
full_line(UUID1, TS2, "later"),
]);
let membership = tracker.sessions()[UUID1].conference.clone().unwrap();
assert_eq!(membership.name, "835");
assert_eq!(membership.member_id, Some(3));
assert_eq!(membership.conference_uuid.as_deref(), Some(UUID2));
}
#[test]
fn reused_name_after_the_last_leave_is_a_new_instance() {
let tracker = track(vec![
format!("{UUID1} EXECUTE [depth=0] loopback/tty-a conference(835)"),
full_line(
UUID1,
TS1,
"Channel leaving conference, cause: NORMAL_CLEARING",
),
format!("{UUID2} EXECUTE [depth=0] sofia/internal/1000 conference(835)"),
]);
assert!(
tracker.sessions()[UUID1].conference.is_none(),
"leaving clears the membership"
);
let rejoined = tracker.sessions()[UUID2].conference.clone().unwrap();
assert_eq!(rejoined.instance, UUID2);
}
#[test]
fn media_keeps_the_outcome_and_the_deduped_offer_set() {
let tracker = track(vec![
full_line(
UUID1,
TS1,
"Audio Codec Compare [opus:102:16000:20:0:1]/[G722:9:16000:20:64000:1]",
),
full_line(
UUID1,
TS1,
"Audio Codec Compare [opus:102:16000:20:0:1]/[opus:116:16000:20:0:1]",
),
full_line(
UUID1,
TS1,
"Audio Codec Compare [opus:116:16000:20:0:1] ++++ is saved as a match",
),
full_line(UUID1, TS2, "Video Codec Compare [H264:109]/[H263:34]"),
full_line(
UUID1,
TS2,
"Video Codec Compare [H264:109] +++ is saved as a match",
),
full_line(
UUID1,
TS2,
"Set Codec sofia/internal/1000 opus/16000 20 ms 320 samples 0 bits 1 channels",
),
full_line(
UUID1,
TS2,
"sofia/internal/1000 Original read codec set to opus:116",
),
]);
let media = &tracker.sessions()[UUID1].media;
assert_eq!(
media.audio.offered.len(),
1,
"the same remote offer compared twice is one entry: {:?}",
media.audio.offered
);
assert_eq!(media.audio.negotiated.as_ref().unwrap().name, "opus");
assert_eq!(media.audio.negotiated.as_ref().unwrap().payload_type, 116);
assert_eq!(media.video.negotiated.as_ref().unwrap().name, "H264");
assert!(
media.audio.negotiated != media.video.negotiated,
"audio and video outcomes are tracked apart"
);
assert_eq!(media.read_codec.as_ref().unwrap().payload_type, 116);
assert_eq!(media.active_audio.as_ref().unwrap().clock_rate, Some(16000));
}
#[test]
fn loopback_b_leg_links_to_its_a_leg() {
let tracker = track(vec![
full_line(UUID1, TS1, "New Channel loopback/tty-a [ignored]"),
full_line(UUID2, TS1, "New Channel loopback/tty-b [ignored]"),
]);
assert_eq!(
tracker.sessions()[UUID1].other_leg_uuid.as_deref(),
Some(UUID2)
);
assert_eq!(
tracker.sessions()[UUID2].other_leg_uuid.as_deref(),
Some(UUID1)
);
}
#[test]
fn concurrent_loopbacks_to_one_destination_do_not_link() {
let tracker = track(vec![
full_line(UUID1, TS1, "New Channel loopback/tty-a [ignored]"),
full_line(UUID2, TS1, "New Channel loopback/tty-a [ignored]"),
full_line(UUID3, TS1, "New Channel loopback/tty-b [ignored]"),
]);
assert!(
tracker.sessions()[UUID3].other_leg_uuid.is_none(),
"two live A legs share the name; picking one would be a guess"
);
}
#[test]
fn a_member_still_present_holds_the_instance_open() {
let tracker = track(vec![
format!("{UUID1} EXECUTE [depth=0] loopback/tty-a conference(835)"),
format!("{UUID2} EXECUTE [depth=0] sofia/internal/1000 conference(835)"),
full_line(
UUID1,
TS1,
"Channel leaving conference, cause: NORMAL_CLEARING",
),
format!("{UUID3} EXECUTE [depth=0] sofia/internal/1001 conference(835)"),
]);
assert_eq!(
tracker.sessions()[UUID3]
.conference
.as_ref()
.unwrap()
.instance,
UUID1
);
}
}