serial_codec/sms/
concat.rs

1//! Reassembly of concatenated SMS (23.040 §9.2.3.24.1/.8).
2//!
3//! Parts are keyed by (TPDU type, address, reference and its width, total) so a
4//! DELIVER and a SUBMIT sharing a reference never merge, nor do an 8-bit and a 16-bit
5//! reference with the same value. Payloads are joined before decoding: GSM 7-bit
6//! septets, UCS2 octets (a surrogate pair split across parts decodes correctly).
7//!
8//! Memory is bounded: at most [`Reassembler::max_pending`] incomplete messages are held
9//! (default [`DEFAULT_MAX_PENDING`]); a new one beyond that evicts the oldest (by the
10//! newest part's timestamp, then arrival). Part slots are allocated as parts arrive. No
11//! timers: pass a timestamp of your choice to [`Reassembler::push`] and call
12//! [`Reassembler::expire_before`] periodically so stale parts do not wait for eviction.
13
14use super::gsm7::{self, Tables};
15use super::pdu::{Alphabet, Tpdu};
16use super::ucs2;
17use std::collections::{BTreeMap, HashMap};
18
19/// Incomplete messages a [`Reassembler::new`] holds before evicting the oldest.
20pub const DEFAULT_MAX_PENDING: usize = 64;
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
23pub enum Kind {
24    Deliver,
25    Submit,
26    StatusReport,
27}
28
29#[derive(Debug, Clone, PartialEq, Eq, Hash)]
30pub struct Key {
31    pub kind: Kind,
32    /// Originator / destination / recipient as displayed (`+` for international).
33    pub address: String,
34    pub reference: u16,
35    /// IEI 0x08 (16-bit reference) rather than 0x00: a different reference space.
36    pub ref_16bit: bool,
37    pub total: u8,
38}
39
40/// One part's user data, UDH removed.
41#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct Fragment {
43    pub alphabet: Alphabet,
44    pub tables: Tables,
45    /// Septets (GSM 7-bit) or octets.
46    pub payload: Vec<u8>,
47}
48
49/// A complete message.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct Message {
52    /// `None` for a message that was not concatenated.
53    pub key: Option<Key>,
54    pub alphabet: Alphabet,
55    /// Joined payloads (septets or octets).
56    pub payload: Vec<u8>,
57    /// `None` for 8-bit data.
58    pub text: Option<String>,
59    pub parts: usize,
60}
61
62struct Pending {
63    /// Parts received so far, by `seq`.
64    parts: BTreeMap<u8, Fragment>,
65    last_seen: u64,
66    /// Arrival order of the newest part (tie-break for eviction).
67    tick: u64,
68}
69
70pub struct Reassembler {
71    pending: HashMap<Key, Pending>,
72    max_pending: usize,
73    tick: u64,
74}
75
76impl Default for Reassembler {
77    fn default() -> Self {
78        Self::with_max_pending(DEFAULT_MAX_PENDING)
79    }
80}
81
82impl Reassembler {
83    /// Holds up to [`DEFAULT_MAX_PENDING`] incomplete messages.
84    pub fn new() -> Self {
85        Self::default()
86    }
87
88    /// Holds up to `max` incomplete messages (at least 1).
89    pub fn with_max_pending(max: usize) -> Self {
90        Reassembler {
91            pending: HashMap::new(),
92            max_pending: max.max(1),
93            tick: 0,
94        }
95    }
96
97    /// The cap on incomplete messages.
98    pub fn max_pending(&self) -> usize {
99        self.max_pending
100    }
101
102    /// Add part `seq` (1-based) of `key`. Returns the message once every part is in.
103    /// A repeated `seq` replaces the earlier copy; `seq` outside `1..=total` is ignored.
104    /// A part of a new message when [`Self::max_pending`] are held evicts the oldest.
105    pub fn push(&mut self, key: Key, seq: u8, fragment: Fragment, now: u64) -> Option<Message> {
106        if seq == 0 || seq > key.total {
107            return None;
108        }
109        self.tick += 1;
110        if !self.pending.contains_key(&key) && self.pending.len() >= self.max_pending {
111            // ponytail: O(max_pending) scan per new message; an ordered index if the cap
112            // ever gets large.
113            let oldest = self
114                .pending
115                .iter()
116                .min_by_key(|(_, p)| (p.last_seen, p.tick))
117                .map(|(k, _)| k.clone());
118            if let Some(k) = oldest {
119                self.pending.remove(&k);
120            }
121        }
122        let entry = self.pending.entry(key.clone()).or_insert_with(|| Pending {
123            parts: BTreeMap::new(),
124            last_seen: now,
125            tick: 0,
126        });
127        entry.parts.insert(seq, fragment);
128        entry.last_seen = entry.last_seen.max(now);
129        entry.tick = self.tick;
130        if entry.parts.len() < usize::from(key.total) {
131            return None;
132        }
133        let parts: Vec<Fragment> = self.pending.remove(&key)?.parts.into_values().collect();
134        Some(join(Some(key), parts))
135    }
136
137    /// Feed a decoded TPDU. Messages without a valid concatenation IE complete at once.
138    pub fn push_tpdu(&mut self, tpdu: &Tpdu, now: u64) -> Option<Message> {
139        let ud = tpdu.user_data()?;
140        let dcs = tpdu.dcs().copied().unwrap_or(super::pdu::Dcs::decode(0));
141        let alphabet = if dcs.compressed {
142            Alphabet::Bit8
143        } else {
144            dcs.alphabet
145        };
146        let fragment = Fragment {
147            alphabet,
148            tables: ud.udh.as_ref().map_or(Tables::DEFAULT, |u| u.tables()),
149            payload: ud.payload.clone(),
150        };
151        let Some(c) = ud.udh.as_ref().and_then(|u| u.concat()) else {
152            return Some(join(None, vec![fragment]));
153        };
154        let kind = match tpdu {
155            Tpdu::Deliver(_) => Kind::Deliver,
156            Tpdu::Submit(_) => Kind::Submit,
157            Tpdu::StatusReport(_) => Kind::StatusReport,
158        };
159        let key = Key {
160            kind,
161            address: tpdu.address().to_string(),
162            reference: c.reference,
163            ref_16bit: c.ref_16bit,
164            total: c.total,
165        };
166        self.push(key, c.seq, fragment, now)
167    }
168
169    /// Drop incomplete messages whose newest part arrived before `cutoff`; returns
170    /// their keys.
171    pub fn expire_before(&mut self, cutoff: u64) -> Vec<Key> {
172        let old: Vec<Key> = self
173            .pending
174            .iter()
175            .filter(|(_, p)| p.last_seen < cutoff)
176            .map(|(k, _)| k.clone())
177            .collect();
178        for k in &old {
179            self.pending.remove(k);
180        }
181        old
182    }
183
184    /// Number of incomplete messages held.
185    pub fn pending(&self) -> usize {
186        self.pending.len()
187    }
188}
189
190fn join(key: Option<Key>, parts: Vec<Fragment>) -> Message {
191    let first = parts[0].clone();
192    let uniform = parts
193        .iter()
194        .all(|p| p.alphabet == first.alphabet && p.tables == first.tables);
195    let payload: Vec<u8> = parts
196        .iter()
197        .flat_map(|p| p.payload.iter().copied())
198        .collect();
199    let decode = |alphabet, tables, bytes: &[u8]| match alphabet {
200        Alphabet::Gsm7 => Some(gsm7::decode(bytes, tables)),
201        Alphabet::Ucs2 => Some(ucs2::decode(bytes)),
202        Alphabet::Bit8 => None,
203    };
204    let text = if uniform {
205        decode(first.alphabet, first.tables, &payload)
206    } else {
207        parts
208            .iter()
209            .map(|p| decode(p.alphabet, p.tables, &p.payload))
210            .collect()
211    };
212    Message {
213        key,
214        alphabet: first.alphabet,
215        payload,
216        text,
217        parts: parts.len(),
218    }
219}
serial_codec/sms/concat.rs

serial_codec/sms/
concat.rs
