freeswitch_log_parser/session/
state.rs

1//! Per-UUID accumulated state and the point-in-time snapshot attached to each
2//! enriched entry.
3
4use std::collections::HashMap;
5use std::str::FromStr;
6
7use freeswitch_types::variables::{SofiaVariable, VariableName};
8use freeswitch_types::{CallDirection, CallState, ChannelState, EventHeader, HangupCause};
9
10use crate::line::parse_line;
11use crate::message::{classify_message, LifecycleEvent, MessageKind};
12use crate::stream::{Block, LogEntry, ParseWarning, SessionReading};
13use crate::uuid::is_uuid;
14
15use super::conference::ConferenceMembership;
16use super::media::SessionMedia;
17use super::parse::{
18    parse_bridge_args, parse_dialplan_context, parse_hangup, parse_new_channel,
19    parse_processing_line, parse_state_change, StateChange,
20};
21
22/// Resolve a typed value into `slot`, or record that it could not be read.
23///
24/// The failed reading leaves whatever `slot` already held — a repeat of a field
25/// the vocabulary cannot read must not erase what an earlier one established.
26fn read<T: FromStr>(
27    slot: &mut Option<T>,
28    value: &str,
29    reading: SessionReading,
30    warnings: &mut Vec<ParseWarning>,
31) {
32    match T::from_str(value) {
33        Ok(parsed) => *slot = Some(parsed),
34        Err(_) => warnings.push(ParseWarning::UnreadableValue {
35            reading,
36            value: ParseWarning::excerpt(value),
37        }),
38    }
39}
40
41/// Where an entry's dump values come from.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43enum DumpSource {
44    /// A reassembled block holds them, so the raw attached lines are skipped:
45    /// re-applying one would clobber a multi-line value with its first fragment.
46    Block,
47    /// The attached lines are all there is.
48    AttachedLines,
49}
50
51/// Mutable per-UUID state accumulator, updated as entries are processed.
52///
53/// Fields are `None` until the corresponding data is first seen in the stream.
54/// Variables accumulate from CHANNEL_DATA dumps, `set()`/`export()` executions,
55/// `SET`/`EXPORT` log lines, and inline `variable_*` lines.
56#[derive(Debug, Clone, Default)]
57#[non_exhaustive]
58pub struct SessionState {
59    /// `None` until a `Channel-Name` field is encountered.
60    pub channel_name: Option<String>,
61    /// `None` until a `State Change` line or a `Channel-State` field is seen.
62    pub channel_state: Option<ChannelState>,
63    /// `None` until a `Callstate Change` line is seen. A distinct vocabulary from
64    /// [`channel_state`](Self::channel_state); neither displaces the other.
65    pub call_state: Option<CallState>,
66    /// First dialplan context seen; set once and never overwritten.
67    pub initial_context: Option<String>,
68    /// Destination of the first `Processing` line = the dialed number at ingress;
69    /// set once and never overwritten (unlike last-wins `dialplan_to`).
70    pub initial_destination: Option<String>,
71    /// Number half of the first `Processing` line's caller side, the caller
72    /// profile's `caller_id_number` at ingress. Set once, and first *seen*: a
73    /// window starting mid-call pins whatever line arrived first in it.
74    pub initial_caller_number: Option<String>,
75    /// Name half of the first `Processing` line's caller side. Set once on the
76    /// same terms as [`initial_caller_number`](Self::initial_caller_number), and
77    /// a rendering rather than the bytes — the line is session-bound, so the
78    /// prefix stage paired off its apostrophes and consumed its backslashes.
79    pub initial_caller_name: Option<String>,
80    /// Current dialplan context; updated on each transfer/continue.
81    pub dialplan_context: Option<String>,
82    /// Caller side of the last `Processing` line — a number, or a display name
83    /// and number. `None` until one is seen.
84    pub dialplan_from: Option<String>,
85    /// Dialed destination of the last `Processing` line. `None` until one is seen.
86    pub dialplan_to: Option<String>,
87    /// Call direction from `Call-Direction` CHANNEL_DATA field; `None` until seen.
88    pub call_direction: Option<CallDirection>,
89    /// Caller ID number from `Caller-Caller-ID-Number` CHANNEL_DATA field; `None` until seen.
90    pub caller_id_number: Option<String>,
91    /// Caller ID name from `Caller-Caller-ID-Name` CHANNEL_DATA field; `None` until seen.
92    pub caller_id_name: Option<String>,
93    /// Destination number from `Caller-Destination-Number` CHANNEL_DATA field; `None` until seen.
94    pub destination_number: Option<String>,
95    /// Hangup cause extracted from ChannelLifecycle Hangup detail; `None` until hangup seen.
96    pub hangup_cause: Option<HangupCause>,
97    /// Timestamp when "has been answered" lifecycle event was seen; `None` until answered.
98    pub answered_at: Option<String>,
99    /// Other leg's UUID; `None` until bridged. Set from `Originate Resulted in Success` on A-leg,
100    /// and from `New Channel` on B-leg (back-pointing to A-leg via originate context).
101    pub other_leg_uuid: Option<String>,
102    /// Conference this session is currently a member of; `None` once it leaves.
103    pub conference: Option<ConferenceMembership>,
104    /// Codecs negotiated on this leg, by media type and direction.
105    pub media: SessionMedia,
106    /// Pending bridge target channel from `EXECUTE bridge()`, consumed when B-leg `New Channel` matches.
107    pub(crate) pending_bridge_target: Option<String>,
108    /// All variables learned so far, with the `variable_` prefix stripped from names.
109    pub variables: HashMap<String, String>,
110}
111/// Immutable point-in-time copy of a session's state, attached to each [`EnrichedEntry`](crate::EnrichedEntry).
112///
113/// Does not include `variables` to keep snapshots lightweight — access the full
114/// variable map via [`SessionTracker::sessions()`](crate::SessionTracker::sessions).
115#[derive(Debug, Clone)]
116#[non_exhaustive]
117pub struct SessionSnapshot {
118    pub channel_name: Option<String>,
119    pub channel_state: Option<ChannelState>,
120    pub call_state: Option<CallState>,
121    pub initial_context: Option<String>,
122    pub initial_destination: Option<String>,
123    pub initial_caller_number: Option<String>,
124    pub initial_caller_name: Option<String>,
125    pub dialplan_context: Option<String>,
126    pub dialplan_from: Option<String>,
127    pub dialplan_to: Option<String>,
128    pub call_direction: Option<CallDirection>,
129    pub caller_id_number: Option<String>,
130    pub caller_id_name: Option<String>,
131    pub destination_number: Option<String>,
132    pub hangup_cause: Option<HangupCause>,
133    pub answered_at: Option<String>,
134    pub other_leg_uuid: Option<String>,
135    pub conference: Option<ConferenceMembership>,
136    pub media: SessionMedia,
137}
138
139impl SessionState {
140    /// Value of a typed channel variable, or `None` if this session never saw it.
141    ///
142    /// Accepts any of `freeswitch-types`' variable-name enums, and spares the
143    /// caller from knowing that [`variables`](Self::variables) keys are stored
144    /// with the `variable_` prefix stripped.
145    pub fn variable<V: VariableName>(&self, var: V) -> Option<&str> {
146        self.variables.get(var.as_str()).map(String::as_str)
147    }
148
149    /// The caller's number, from the most trustworthy source this session saw:
150    /// the dump's `Caller-Caller-ID-Number`, then `sip_from_user`, then the
151    /// first `Processing` line's number half.
152    ///
153    /// Whether a leg dumps depends on the box's dialplan and log verbosity, so
154    /// the lower rungs are what answer for the legs that never do.
155    ///
156    /// Not on [`SessionSnapshot`], which carries no variables and so cannot walk
157    /// the middle rung.
158    pub fn caller_number(&self) -> Option<&str> {
159        self.caller_id_number
160            .as_deref()
161            .or_else(|| self.variable(SofiaVariable::SipFromUser))
162            .or(self.initial_caller_number.as_deref())
163    }
164
165    /// The dialed number, on the same terms as
166    /// [`caller_number`](Self::caller_number): the dump's
167    /// `Caller-Destination-Number`, then `sip_to_user`, then the first
168    /// `Processing` line's destination.
169    pub fn callee_number(&self) -> Option<&str> {
170        self.destination_number
171            .as_deref()
172            .or_else(|| self.variable(SofiaVariable::SipToUser))
173            .or(self.initial_destination.as_deref())
174    }
175
176    /// Bound exhaustively so a field added here and forgotten in the snapshot
177    /// fails to compile; the two `_` bindings are the deliberate omissions.
178    pub(super) fn snapshot(&self) -> SessionSnapshot {
179        let SessionState {
180            channel_name,
181            channel_state,
182            call_state,
183            initial_context,
184            initial_destination,
185            initial_caller_number,
186            initial_caller_name,
187            dialplan_context,
188            dialplan_from,
189            dialplan_to,
190            call_direction,
191            caller_id_number,
192            caller_id_name,
193            destination_number,
194            hangup_cause,
195            answered_at,
196            other_leg_uuid,
197            conference,
198            media,
199            pending_bridge_target: _,
200            variables: _,
201        } = self;
202
203        SessionSnapshot {
204            channel_name: channel_name.clone(),
205            channel_state: *channel_state,
206            call_state: *call_state,
207            initial_context: initial_context.clone(),
208            initial_destination: initial_destination.clone(),
209            initial_caller_number: initial_caller_number.clone(),
210            initial_caller_name: initial_caller_name.clone(),
211            dialplan_context: dialplan_context.clone(),
212            dialplan_from: dialplan_from.clone(),
213            dialplan_to: dialplan_to.clone(),
214            call_direction: *call_direction,
215            caller_id_number: caller_id_number.clone(),
216            caller_id_name: caller_id_name.clone(),
217            destination_number: destination_number.clone(),
218            hangup_cause: *hangup_cause,
219            answered_at: answered_at.clone(),
220            other_leg_uuid: other_leg_uuid.clone(),
221            conference: conference.clone(),
222            media: media.clone(),
223        }
224    }
225
226    /// Absorb one CHANNEL_DATA field. The collision splitter can hand a dump
227    /// field over as its own entry rather than inside a block, so both arrival
228    /// shapes decode here — a lighter reading on one of them would drop whatever
229    /// it left out, `Other-Leg-Unique-ID` included, only for split dumps.
230    fn apply_channel_field(&mut self, name: &str, value: &str, warnings: &mut Vec<ParseWarning>) {
231        match EventHeader::from_str(name) {
232            Ok(EventHeader::ChannelName) => self.channel_name = Some(value.to_string()),
233            Ok(EventHeader::ChannelState) => read(
234                &mut self.channel_state,
235                value,
236                SessionReading::ChannelState,
237                warnings,
238            ),
239            Ok(EventHeader::CallDirection) => read(
240                &mut self.call_direction,
241                value,
242                SessionReading::CallDirection,
243                warnings,
244            ),
245            Ok(EventHeader::CallerCallerIdNumber) => {
246                self.caller_id_number = Some(value.to_string())
247            }
248            Ok(EventHeader::CallerCallerIdName) => self.caller_id_name = Some(value.to_string()),
249            Ok(EventHeader::CallerDestinationNumber) => {
250                self.destination_number = Some(value.to_string())
251            }
252            Ok(EventHeader::OtherLegUniqueId) => self.other_leg_uuid = Some(value.to_string()),
253            _ => {}
254        }
255    }
256
257    /// Whether this session has reached a state it cannot leave. Stragglers in a
258    /// terminal state are never candidates for leg linking.
259    ///
260    /// `DOWN` is not terminal: it doubles as the initial call state before any
261    /// change is observed.
262    pub(super) fn is_terminal(&self) -> bool {
263        matches!(
264            self.channel_state,
265            Some(
266                ChannelState::CsHangup
267                    | ChannelState::CsReporting
268                    | ChannelState::CsDestroy
269                    | ChannelState::CsNone
270            )
271        ) || matches!(self.call_state, Some(CallState::Hangup))
272    }
273
274    /// Absorb an entry, returning whatever readings its values defeated.
275    pub(super) fn update_from_entry(&mut self, entry: &LogEntry) -> Vec<ParseWarning> {
276        let mut warnings = Vec::new();
277        let dump = match entry.block {
278            Some(Block::ChannelData { .. }) => DumpSource::Block,
279            _ => DumpSource::AttachedLines,
280        };
281        if let Some(Block::ChannelData { fields, variables }) = &entry.block {
282            for (name, value) in fields {
283                self.apply_channel_field(name, value, &mut warnings);
284            }
285            for (name, value) in variables {
286                self.variables
287                    .insert(name.bare().to_string(), value.clone());
288            }
289        }
290
291        match &entry.message_kind {
292            MessageKind::Execute {
293                application,
294                arguments,
295                ..
296            } => match application.as_str() {
297                "set" | "export" => {
298                    if let Some((name, value)) = arguments.split_once('=') {
299                        self.variables.insert(name.to_string(), value.to_string());
300                    }
301                }
302                "bridge" => {
303                    if let Some(info) = parse_bridge_args(arguments) {
304                        if let Some(uuid) = &info.origination_uuid {
305                            self.other_leg_uuid = Some(uuid.clone());
306                        }
307                        self.pending_bridge_target = Some(info.target_channel);
308                    }
309                }
310                _ => {}
311            },
312            MessageKind::ChannelLifecycle { event, detail } => match event {
313                LifecycleEvent::NewChannel => {
314                    if self.channel_name.is_none() {
315                        self.channel_name = parse_new_channel(detail).map(str::to_string);
316                    }
317                }
318                LifecycleEvent::Hangup => {
319                    if let Some(cause) = parse_hangup(detail) {
320                        read(
321                            &mut self.hangup_cause,
322                            cause,
323                            SessionReading::HangupCause,
324                            &mut warnings,
325                        );
326                    }
327                }
328                LifecycleEvent::Answered => {
329                    self.answered_at
330                        .get_or_insert_with(|| entry.timestamp.clone());
331                }
332                _ => {}
333            },
334            kind => self.apply_kind(kind, &mut warnings),
335        }
336
337        self.apply_processing(&entry.message);
338        self.media.update_from_entry(entry);
339
340        for attached in &entry.attached {
341            let parsed = parse_line(attached);
342            self.update_from_message(parsed.message, dump, &mut warnings);
343        }
344        warnings
345    }
346
347    /// Canonical extraction for message kinds that appear on both primary and
348    /// attached lines. Entry-only kinds (Execute, ChannelLifecycle — the
349    /// latter needs the entry timestamp) stay in `update_from_entry`.
350    fn apply_kind(&mut self, kind: &MessageKind, warnings: &mut Vec<ParseWarning>) {
351        match kind {
352            MessageKind::Dialplan { detail, .. } => {
353                if let Some(context) = parse_dialplan_context(detail) {
354                    self.initial_context.get_or_insert(context.to_string());
355                    self.dialplan_context = Some(context.to_string());
356                }
357            }
358            MessageKind::Variable { name, value } => {
359                self.variables
360                    .insert(name.bare().to_string(), value.clone());
361            }
362            MessageKind::ChannelField { name, value } => {
363                self.apply_channel_field(name, value, warnings)
364            }
365            // A suffix the log wrote is evidence of a peer, so a value that is
366            // not a UUID is named rather than left to read as an absent suffix.
367            MessageKind::OriginateSuccess {
368                peer_uuid: Some(peer),
369                ..
370            } if !is_uuid(peer) => warnings.push(ParseWarning::UnreadableValue {
371                reading: SessionReading::PeerUuid,
372                value: ParseWarning::excerpt(peer),
373            }),
374            MessageKind::StateChange { detail } => match parse_state_change(detail) {
375                Some(StateChange::Channel(to)) => read(
376                    &mut self.channel_state,
377                    to,
378                    SessionReading::ChannelState,
379                    warnings,
380                ),
381                Some(StateChange::Call(to)) => read(
382                    &mut self.call_state,
383                    to,
384                    SessionReading::CallState,
385                    warnings,
386                ),
387                None => {}
388            },
389            _ => {}
390        }
391    }
392
393    /// `Processing <caller>-><dest> in context <ctx>` — emitted by the dialplan
394    /// hunt on both primary and attached lines; anchored on raw text because
395    /// `classify_message` folds it into `Dialplan` with the prefix stripped.
396    fn apply_processing(&mut self, msg: &str) {
397        if let Some(dp) = parse_processing_line(msg) {
398            self.initial_context.get_or_insert(dp.context.clone());
399            self.initial_destination.get_or_insert(dp.to.clone());
400            if let Some(number) = dp.from_number {
401                self.initial_caller_number.get_or_insert(number);
402            }
403            if let Some(name) = dp.from_name {
404                self.initial_caller_name.get_or_insert(name);
405            }
406            self.dialplan_context = Some(dp.context);
407            self.dialplan_from = Some(dp.from);
408            self.dialplan_to = Some(dp.to);
409        }
410    }
411
412    fn update_from_message(
413        &mut self,
414        msg: &str,
415        dump: DumpSource,
416        warnings: &mut Vec<ParseWarning>,
417    ) {
418        let kind = classify_message(msg);
419        match &kind {
420            MessageKind::Variable { .. } | MessageKind::ChannelField { .. }
421                if dump == DumpSource::Block => {}
422            kind => self.apply_kind(kind, warnings),
423        }
424        self.apply_processing(msg);
425    }
426}
freeswitch_log_parser/session/state.rs

freeswitch_log_parser/session/
state.rs
