lightstreamer_rs/client/subscription.rs
1//! What to subscribe to, and how.
2//!
3//! A **subscription** binds a set of *items* to a set of *fields* and a
4//! *mode*. The server's Metadata Adapter resolves the item group and the field
5//! schema into the actual items and fields, so what a group name means is
6//! decided by the server, not by this crate
7//! [`docs/spec/03-requests.md` §6.1].
8
9use std::num::NonZeroU32;
10
11use crate::config::ConfigError;
12use crate::session::{
13 DecimalNumber, RequestedBufferSize, RequestedMaxFrequency, Snapshot as WireSnapshot,
14 SubscriptionMode as WireMode, SubscriptionSpec,
15};
16
17/// How the server treats the items of a subscription
18/// [`docs/spec/03-requests.md` §6.1].
19///
20/// The mode is not a preference — it changes what the data *means*, and an
21/// item may only be subscribed in the modes its Data Adapter supports.
22///
23/// | Mode | The server keeps | A snapshot is | Typical use |
24/// |---|---|---|---|
25/// | [`Merge`](SubscriptionMode::Merge) | the current value of every field | the current state | a quote: price, bid, ask |
26/// | [`Distinct`](SubscriptionMode::Distinct) | recent events, unmerged | the most recent events | a news feed, a chat |
27/// | [`Command`](SubscriptionMode::Command) | a keyed table | the rows of the table | an order book, a portfolio |
28/// | [`Raw`](SubscriptionMode::Raw) | nothing | not available | a firehose you filter yourself |
29///
30/// In every mode but `Raw` the server may filter, buffer and limit the flow;
31/// `Raw` forwards each update as it is and is the one mode with no server-side
32/// bookkeeping at all.
33#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
34#[repr(u8)]
35pub enum SubscriptionMode {
36 /// Every update is forwarded as-is. No item state is kept, no snapshot is
37 /// available, and no filtering is applied.
38 Raw,
39 /// The server keeps the current value of each field and merges each update
40 /// into it, so an update carries only what changed and a snapshot is the
41 /// current state.
42 Merge,
43 /// Each update is a distinct event that does not replace the previous one.
44 /// A snapshot is the most recent events, up to a length the server or the
45 /// subscription chooses.
46 Distinct,
47 /// Updates carry `ADD`, `UPDATE` and `DELETE` commands against a key
48 /// field, forming a table. The field schema **must** contain a `key` field
49 /// and a `command` field, or the server rejects the subscription with
50 /// Appendix B code 15 or 16 [`docs/spec/05-error-codes.md` §2].
51 Command,
52}
53
54impl SubscriptionMode {
55 /// The mode as the protocol spells it: `RAW`, `MERGE`, `DISTINCT` or
56 /// `COMMAND`.
57 #[must_use]
58 #[inline]
59 pub const fn as_str(self) -> &'static str {
60 match self {
61 Self::Raw => "RAW",
62 Self::Merge => "MERGE",
63 Self::Distinct => "DISTINCT",
64 Self::Command => "COMMAND",
65 }
66 }
67
68 /// The wire enum this crate encodes with.
69 pub(crate) const fn to_wire(self) -> WireMode {
70 match self {
71 Self::Raw => WireMode::Raw,
72 Self::Merge => WireMode::Merge,
73 Self::Distinct => WireMode::Distinct,
74 Self::Command => WireMode::Command,
75 }
76 }
77}
78
79/// Which items a subscription covers.
80///
81/// The protocol sends a single `LS_group` string and lets the server's
82/// Metadata Adapter decide what it names
83/// [`docs/spec/03-requests.md` §6.1]. In practice adapters use one of two
84/// conventions, and this type supports both:
85///
86/// - a **group name** the adapter knows, resolved server-side into a list;
87/// - a **literal list** of item names separated by spaces, which is what the
88/// stock adapters and the public demo server accept.
89///
90/// # Examples
91///
92/// ```
93/// use lightstreamer_rs::ItemGroup;
94///
95/// let listed = ItemGroup::from_items(["item1", "item2", "item3"])?;
96/// assert_eq!(listed.as_str(), "item1 item2 item3");
97///
98/// let named = ItemGroup::try_new("portfolio1")?;
99/// assert_eq!(named.as_str(), "portfolio1");
100/// # Ok::<(), lightstreamer_rs::ConfigError>(())
101/// ```
102#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
103pub struct ItemGroup {
104 text: String,
105 /// Whether the caller spelled the items out. When it did, this crate knows
106 /// each position's name and can report it on every update; when the group
107 /// is a server-side name, nothing in TLCP ever reveals the item names
108 /// [`docs/spec/04-notifications.md` §3.1] and positions are reported by
109 /// index instead.
110 listed: bool,
111}
112
113impl ItemGroup {
114 /// Wraps a group name the server's Metadata Adapter will resolve.
115 ///
116 /// # Errors
117 ///
118 /// - [`ConfigError::EmptyItemGroup`] if the name is empty or only
119 /// whitespace.
120 /// - [`ConfigError::ItemGroupControlCharacter`] if it contains an ASCII
121 /// control character. A guard of this crate: the specification gives no
122 /// grammar for the name, but a control character in one is always a
123 /// mistake.
124 #[must_use = "a checked item group does nothing unless it is used"]
125 pub fn try_new(group: impl Into<String>) -> Result<Self, ConfigError> {
126 let group = group.into();
127 if group.trim().is_empty() {
128 return Err(ConfigError::EmptyItemGroup);
129 }
130 if group.chars().any(char::is_control) {
131 return Err(ConfigError::ItemGroupControlCharacter);
132 }
133 Ok(Self {
134 text: group,
135 listed: false,
136 })
137 }
138
139 /// Builds the space-separated list convention from individual item names.
140 ///
141 /// # Errors
142 ///
143 /// - [`ConfigError::EmptyItemGroup`] if the iterator is empty.
144 /// - [`ConfigError::ItemNameHasWhitespace`] if any name contains
145 /// whitespace, which would make it two items once joined.
146 /// - [`ConfigError::ItemGroupControlCharacter`] if any name contains an
147 /// ASCII control character.
148 ///
149 /// # Examples
150 ///
151 /// ```
152 /// use lightstreamer_rs::{ConfigError, ItemGroup};
153 ///
154 /// assert!(matches!(
155 /// ItemGroup::from_items(["two words"]),
156 /// Err(ConfigError::ItemNameHasWhitespace { .. })
157 /// ));
158 /// ```
159 #[must_use = "a checked item group does nothing unless it is used"]
160 pub fn from_items<I, S>(items: I) -> Result<Self, ConfigError>
161 where
162 I: IntoIterator<Item = S>,
163 S: Into<String>,
164 {
165 let names: Vec<String> = items.into_iter().map(Into::into).collect();
166 if names.is_empty() {
167 return Err(ConfigError::EmptyItemGroup);
168 }
169 for name in &names {
170 if name.is_empty() || name.chars().any(char::is_whitespace) {
171 return Err(ConfigError::ItemNameHasWhitespace { name: name.clone() });
172 }
173 if name.chars().any(char::is_control) {
174 return Err(ConfigError::ItemGroupControlCharacter);
175 }
176 }
177 Ok(Self {
178 text: names.join(" "),
179 listed: true,
180 })
181 }
182
183 /// The group as it goes on the wire, in `LS_group`.
184 #[must_use]
185 #[inline]
186 pub fn as_str(&self) -> &str {
187 &self.text
188 }
189
190 /// The item names, when the caller spelled them out.
191 ///
192 /// Empty for a group *name* the server resolves: TLCP never transmits item
193 /// names, so this crate cannot know them and reports each item by its
194 /// 1-based position instead [`docs/spec/04-notifications.md` §3.1].
195 ///
196 /// # Examples
197 ///
198 /// ```
199 /// use lightstreamer_rs::ItemGroup;
200 ///
201 /// assert_eq!(ItemGroup::from_items(["a", "b"])?.names(), vec!["a", "b"]);
202 /// assert!(ItemGroup::try_new("portfolio1")?.names().is_empty());
203 /// # Ok::<(), lightstreamer_rs::ConfigError>(())
204 /// ```
205 #[must_use]
206 pub fn names(&self) -> Vec<&str> {
207 if self.listed {
208 self.text.split_whitespace().collect()
209 } else {
210 Vec::new()
211 }
212 }
213}
214
215/// Which fields a subscription asks for.
216///
217/// The mirror of [`ItemGroup`] on the field side: `LS_schema` is one string
218/// the Metadata Adapter resolves, either a schema name it knows or a
219/// space-separated list of field names [`docs/spec/03-requests.md` §6.1].
220///
221/// The **order** matters: it is the order in which values arrive in every
222/// update, and the order the field indices of an update refer to.
223///
224/// # Examples
225///
226/// ```
227/// use lightstreamer_rs::FieldSchema;
228///
229/// let schema = FieldSchema::from_fields(["last_price", "time", "pct_change"])?;
230/// assert_eq!(schema.as_str(), "last_price time pct_change");
231/// # Ok::<(), lightstreamer_rs::ConfigError>(())
232/// ```
233#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
234pub struct FieldSchema {
235 text: String,
236 /// Whether the caller spelled the fields out. See [`ItemGroup::listed`].
237 listed: bool,
238}
239
240impl FieldSchema {
241 /// Wraps a schema name the server's Metadata Adapter will resolve.
242 ///
243 /// # Errors
244 ///
245 /// - [`ConfigError::EmptyFieldSchema`] if the name is empty or only
246 /// whitespace.
247 /// - [`ConfigError::FieldSchemaControlCharacter`] if it contains an ASCII
248 /// control character.
249 #[must_use = "a checked field schema does nothing unless it is used"]
250 pub fn try_new(schema: impl Into<String>) -> Result<Self, ConfigError> {
251 let schema = schema.into();
252 if schema.trim().is_empty() {
253 return Err(ConfigError::EmptyFieldSchema);
254 }
255 if schema.chars().any(char::is_control) {
256 return Err(ConfigError::FieldSchemaControlCharacter);
257 }
258 Ok(Self {
259 text: schema,
260 listed: false,
261 })
262 }
263
264 /// Builds the space-separated list convention from individual field names.
265 ///
266 /// # Errors
267 ///
268 /// - [`ConfigError::EmptyFieldSchema`] if the iterator is empty.
269 /// - [`ConfigError::FieldNameHasWhitespace`] if any name contains
270 /// whitespace.
271 /// - [`ConfigError::FieldSchemaControlCharacter`] if any name contains an
272 /// ASCII control character.
273 #[must_use = "a checked field schema does nothing unless it is used"]
274 pub fn from_fields<I, S>(fields: I) -> Result<Self, ConfigError>
275 where
276 I: IntoIterator<Item = S>,
277 S: Into<String>,
278 {
279 let names: Vec<String> = fields.into_iter().map(Into::into).collect();
280 if names.is_empty() {
281 return Err(ConfigError::EmptyFieldSchema);
282 }
283 for name in &names {
284 if name.is_empty() || name.chars().any(char::is_whitespace) {
285 return Err(ConfigError::FieldNameHasWhitespace { name: name.clone() });
286 }
287 if name.chars().any(char::is_control) {
288 return Err(ConfigError::FieldSchemaControlCharacter);
289 }
290 }
291 Ok(Self {
292 text: names.join(" "),
293 listed: true,
294 })
295 }
296
297 /// The schema as it goes on the wire, in `LS_schema`.
298 #[must_use]
299 #[inline]
300 pub fn as_str(&self) -> &str {
301 &self.text
302 }
303
304 /// The field names, when the caller spelled them out.
305 ///
306 /// Empty for a schema *name* the server resolves: TLCP announces only how
307 /// many fields a subscription has, never what they are called
308 /// [`docs/spec/04-notifications.md` §3.1], so in that case every field is
309 /// reported by its 1-based position instead.
310 #[must_use]
311 pub fn names(&self) -> Vec<&str> {
312 if self.listed {
313 self.text.split_whitespace().collect()
314 } else {
315 Vec::new()
316 }
317 }
318}
319
320/// Whether the server should send the current state before the live flow
321/// [`docs/spec/03-requests.md` §6.1].
322///
323/// A snapshot lets an application start from a complete picture instead of
324/// waiting for every field to tick. Its meaning depends on the mode: in
325/// [`SubscriptionMode::Merge`] it is the current value of every field, in
326/// [`SubscriptionMode::Distinct`] the most recent events, in
327/// [`SubscriptionMode::Command`] the rows of the table. It is not available in
328/// [`SubscriptionMode::Raw`] at all.
329#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
330#[repr(u8)]
331pub enum Snapshot {
332 /// Do not send a snapshot; start from the live flow. The server's default.
333 #[default]
334 Off,
335 /// Send the snapshot, which may legitimately be empty.
336 On,
337 /// Send at most this many snapshot events.
338 ///
339 /// Accepted **only** with [`SubscriptionMode::Distinct`]; the server
340 /// rejects it for the other modes. Fewer events may arrive, if the server
341 /// has fewer or imposes a stricter limit.
342 Length(NonZeroU32),
343}
344
345impl Snapshot {
346 /// Whether this asks the server for a snapshot at all.
347 #[must_use]
348 #[inline]
349 pub const fn is_requested(self) -> bool {
350 !matches!(self, Self::Off)
351 }
352
353 /// The wire enum this crate encodes with.
354 const fn to_wire(self) -> WireSnapshot {
355 match self {
356 Self::Off => WireSnapshot::Off,
357 Self::On => WireSnapshot::On,
358 Self::Length(n) => WireSnapshot::Length(n),
359 }
360 }
361}
362
363/// A ceiling on how often the server may send updates for an item
364/// [`docs/spec/03-requests.md` §6.1].
365///
366/// Frequency limiting happens **on the server**, before the data is sent, so
367/// it is the effective way to protect a slow consumer: this crate never drops
368/// an update it has received (see [`Updates`](crate::Updates)).
369#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
370pub enum MaxFrequency {
371 /// No limit: each update is sent as soon as possible. The server's
372 /// default.
373 #[default]
374 Unlimited,
375 /// No limit **and no losses**: the server never merges or discards an
376 /// update to keep up. If it cannot, it says so with an overflow
377 /// notification rather than dropping silently
378 /// [`docs/spec/04-notifications.md` §3]. Not compatible with a buffer size
379 /// request, which the server ignores in this case.
380 Unfiltered,
381 /// At most this many updates per second, per item.
382 Limited(FrequencyLimit),
383}
384
385impl MaxFrequency {
386 /// A ceiling in updates per second, written as a decimal number.
387 ///
388 /// # Errors
389 ///
390 /// [`ConfigError::InvalidFrequency`] if the text is not a plain decimal
391 /// number with a dot separator — `2`, `0.5` and `12.75` are accepted; a
392 /// sign, an exponent, a comma separator and a bare `.5` are not.
393 ///
394 /// # Examples
395 ///
396 /// ```
397 /// use lightstreamer_rs::{ConfigError, MaxFrequency};
398 ///
399 /// let twice_a_second = MaxFrequency::limited("2.0")?;
400 /// assert!(matches!(MaxFrequency::limited("2,0"), Err(ConfigError::InvalidFrequency { .. })));
401 /// # Ok::<(), ConfigError>(())
402 /// ```
403 pub fn limited(updates_per_second: impl Into<String>) -> Result<Self, ConfigError> {
404 let text = updates_per_second.into();
405 DecimalNumber::try_new(text.clone())
406 .map(|number| Self::Limited(FrequencyLimit(number)))
407 .map_err(|_| ConfigError::InvalidFrequency { value: text })
408 }
409
410 /// The wire enum this crate encodes with.
411 fn to_wire(&self) -> RequestedMaxFrequency {
412 match self {
413 Self::Unlimited => RequestedMaxFrequency::Unlimited,
414 Self::Unfiltered => RequestedMaxFrequency::Unfiltered,
415 Self::Limited(limit) => RequestedMaxFrequency::Limited(limit.0.clone()),
416 }
417 }
418}
419
420/// A checked decimal frequency, in updates per second.
421///
422/// Held as text rather than as a number because the protocol carries it as
423/// text: `2.0` and `2` are the same rate but not the same bytes, and this
424/// crate sends back exactly what the caller wrote. It never does arithmetic on
425/// the value.
426#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
427pub struct FrequencyLimit(DecimalNumber);
428
429impl FrequencyLimit {
430 /// The limit as it goes on the wire.
431 #[must_use]
432 #[inline]
433 pub fn as_str(&self) -> &str {
434 self.0.as_str()
435 }
436}
437
438/// How much the server may buffer per item before it has to drop or merge
439/// [`docs/spec/03-requests.md` §6.1].
440#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
441pub enum BufferSize {
442 /// Let the server choose from its own configuration. The default.
443 #[default]
444 ServerDecides,
445 /// Ask for no limit. The server may still impose one.
446 Unlimited,
447 /// Ask for this many update events.
448 Events(NonZeroU32),
449}
450
451impl BufferSize {
452 /// The wire enum this crate encodes with, or `None` to omit the parameter.
453 const fn to_wire(self) -> Option<RequestedBufferSize> {
454 match self {
455 Self::ServerDecides => None,
456 Self::Unlimited => Some(RequestedBufferSize::Unlimited),
457 Self::Events(n) => Some(RequestedBufferSize::Events(n)),
458 }
459 }
460}
461
462/// A request to subscribe to a set of items.
463///
464/// Build one, hand it to [`Client::subscribe`](crate::Client::subscribe), and
465/// consume the [`Updates`](crate::Updates) stream you get back. The three
466/// values with no default — the mode, the items and the fields — are taken by
467/// [`Subscription::new`]; everything else is optional and defaults to letting
468/// the server decide.
469///
470/// # Not every combination is a subscription
471///
472/// The optional parameters are not independent of the mode: a snapshot length
473/// is admitted only in [`SubscriptionMode::Distinct`], and a snapshot, a
474/// frequency cap and a buffer size are each meaningless in
475/// [`SubscriptionMode::Raw`] [`docs/spec/03-requests.md` §6.1]. The setters
476/// stay infallible so that they chain, and the whole combination is checked at
477/// once by [`Subscription::validate`] — which
478/// [`Client::subscribe`](crate::Client::subscribe) calls before anything is
479/// sent, so an impossible subscription is a typed error at the call rather
480/// than a stream that never activates.
481///
482/// # Examples
483///
484/// ```
485/// use lightstreamer_rs::{FieldSchema, ItemGroup, Snapshot, Subscription, SubscriptionMode};
486///
487/// let subscription = Subscription::new(
488/// SubscriptionMode::Merge,
489/// ItemGroup::from_items(["item1", "item2"])?,
490/// FieldSchema::from_fields(["last_price", "time"])?,
491/// )
492/// .with_snapshot(Snapshot::On);
493///
494/// assert_eq!(subscription.mode(), SubscriptionMode::Merge);
495/// # Ok::<(), lightstreamer_rs::ConfigError>(())
496/// ```
497#[derive(Debug, Clone, PartialEq, Eq)]
498pub struct Subscription {
499 mode: SubscriptionMode,
500 items: ItemGroup,
501 fields: FieldSchema,
502 data_adapter: Option<String>,
503 selector: Option<String>,
504 snapshot: Snapshot,
505 max_frequency: MaxFrequency,
506 buffer_size: BufferSize,
507}
508
509impl Subscription {
510 /// A subscription to a set of items, for a set of fields, in a mode.
511 #[must_use = "builders do nothing unless the subscription is used"]
512 pub fn new(mode: SubscriptionMode, items: ItemGroup, fields: FieldSchema) -> Self {
513 Self {
514 mode,
515 items,
516 fields,
517 data_adapter: None,
518 selector: None,
519 snapshot: Snapshot::default(),
520 max_frequency: MaxFrequency::default(),
521 buffer_size: BufferSize::default(),
522 }
523 }
524
525 /// Names the Data Adapter within the session's Adapter Set that provides
526 /// these items.
527 ///
528 /// Left unset, the server uses the Adapter Set's default Data Adapter
529 /// [`docs/spec/03-requests.md` §6.1].
530 #[must_use = "builders do nothing unless the subscription is used"]
531 pub fn with_data_adapter(mut self, name: impl Into<String>) -> Self {
532 self.data_adapter = Some(name.into());
533 self
534 }
535
536 /// Passes a selector name to the Metadata Adapter, which uses it to filter
537 /// the updates of every item in the subscription
538 /// [`docs/spec/03-requests.md` §6.1].
539 ///
540 /// What a selector name means is entirely up to the adapter.
541 #[must_use = "builders do nothing unless the subscription is used"]
542 pub fn with_selector(mut self, name: impl Into<String>) -> Self {
543 self.selector = Some(name.into());
544 self
545 }
546
547 /// Asks for a snapshot before the live flow.
548 #[must_use = "builders do nothing unless the subscription is used"]
549 pub const fn with_snapshot(mut self, snapshot: Snapshot) -> Self {
550 self.snapshot = snapshot;
551 self
552 }
553
554 /// Caps how often the server sends updates for each item.
555 #[must_use = "builders do nothing unless the subscription is used"]
556 pub fn with_max_frequency(mut self, frequency: MaxFrequency) -> Self {
557 self.max_frequency = frequency;
558 self
559 }
560
561 /// Asks the server for a particular per-item buffer size.
562 #[must_use = "builders do nothing unless the subscription is used"]
563 pub const fn with_buffer_size(mut self, size: BufferSize) -> Self {
564 self.buffer_size = size;
565 self
566 }
567
568 /// The mode this subscription was built with.
569 #[must_use]
570 #[inline]
571 pub const fn mode(&self) -> SubscriptionMode {
572 self.mode
573 }
574
575 /// The items this subscription covers.
576 #[must_use]
577 #[inline]
578 pub const fn items(&self) -> &ItemGroup {
579 &self.items
580 }
581
582 /// The fields this subscription asks for.
583 #[must_use]
584 #[inline]
585 pub const fn fields(&self) -> &FieldSchema {
586 &self.fields
587 }
588
589 /// Whether a snapshot was asked for.
590 #[must_use]
591 #[inline]
592 pub const fn snapshot(&self) -> Snapshot {
593 self.snapshot
594 }
595
596 /// Checks that the mode admits every option this subscription carries.
597 ///
598 /// [`Client::subscribe`](crate::Client::subscribe) calls this first, so a
599 /// caller never has to; it is public because checking a configuration
600 /// before a client exists is sometimes what you want, and because the
601 /// error is more useful at the line that built the subscription than at
602 /// the line that sent it.
603 ///
604 /// # Errors
605 ///
606 /// - [`ConfigError::SnapshotLengthNeedsDistinct`] — a
607 /// [`Snapshot::Length`] outside [`SubscriptionMode::Distinct`]. This one
608 /// is the specification's own rule: a length is "admitted only if
609 /// `LS_mode` is `DISTINCT`" [`docs/spec/03-requests.md` §6.1].
610 /// - [`ConfigError::SnapshotNeedsAStatefulMode`] — a snapshot in
611 /// [`SubscriptionMode::Raw`], which keeps no state to snapshot.
612 /// - [`ConfigError::FrequencyNeedsAFilteredMode`] — a frequency cap in
613 /// [`SubscriptionMode::Raw`], which the server does not filter.
614 /// - [`ConfigError::BufferSizeNeedsMergeOrDistinct`] — a buffer size
615 /// outside [`SubscriptionMode::Merge`] and [`SubscriptionMode::Distinct`].
616 /// - [`ConfigError::BufferSizeWithUnfilteredFrequency`] — a buffer size
617 /// alongside [`MaxFrequency::Unfiltered`], which the server ignores.
618 ///
619 /// The last four are refused rather than dropped. The server would accept
620 /// the request and ignore the parameter [`docs/spec/03-requests.md` §6.1],
621 /// which means a caller who asked for a cap would run without one and
622 /// never be told — a silent difference between what was written and what
623 /// is happening.
624 ///
625 /// # Examples
626 ///
627 /// ```
628 /// use lightstreamer_rs::{
629 /// ConfigError, FieldSchema, ItemGroup, Snapshot, Subscription, SubscriptionMode,
630 /// };
631 ///
632 /// let raw = Subscription::new(
633 /// SubscriptionMode::Raw,
634 /// ItemGroup::from_items(["item1"])?,
635 /// FieldSchema::from_fields(["last_price"])?,
636 /// )
637 /// .with_snapshot(Snapshot::On);
638 ///
639 /// assert!(matches!(
640 /// raw.validate(),
641 /// Err(ConfigError::SnapshotNeedsAStatefulMode)
642 /// ));
643 /// # Ok::<(), ConfigError>(())
644 /// ```
645 pub fn validate(&self) -> Result<(), ConfigError> {
646 let filtered = !matches!(self.mode, SubscriptionMode::Raw);
647
648 if matches!(self.snapshot, Snapshot::Length(_)) && self.mode != SubscriptionMode::Distinct {
649 return Err(ConfigError::SnapshotLengthNeedsDistinct {
650 mode: self.mode.as_str(),
651 });
652 }
653 if self.snapshot.is_requested() && !filtered {
654 return Err(ConfigError::SnapshotNeedsAStatefulMode);
655 }
656 if !matches!(self.max_frequency, MaxFrequency::Unlimited) && !filtered {
657 return Err(ConfigError::FrequencyNeedsAFilteredMode);
658 }
659 if !matches!(self.buffer_size, BufferSize::ServerDecides) {
660 if !matches!(
661 self.mode,
662 SubscriptionMode::Merge | SubscriptionMode::Distinct
663 ) {
664 return Err(ConfigError::BufferSizeNeedsMergeOrDistinct {
665 mode: self.mode.as_str(),
666 });
667 }
668 if matches!(self.max_frequency, MaxFrequency::Unfiltered) {
669 return Err(ConfigError::BufferSizeWithUnfilteredFrequency);
670 }
671 }
672 Ok(())
673 }
674
675 /// The item names the caller declared, or an empty vector for a
676 /// server-resolved group.
677 pub(crate) fn declared_items(&self) -> Vec<String> {
678 self.items.names().into_iter().map(str::to_owned).collect()
679 }
680
681 /// The field names the caller declared, or an empty vector for a
682 /// server-resolved schema.
683 pub(crate) fn declared_fields(&self) -> Vec<String> {
684 self.fields.names().into_iter().map(str::to_owned).collect()
685 }
686
687 /// Translates into what the session state machine subscribes with.
688 pub(crate) fn into_spec(self) -> SubscriptionSpec {
689 let declared_items = self.declared_items();
690 let declared_fields = self.declared_fields();
691 SubscriptionSpec {
692 declared_items,
693 declared_fields,
694 group: self.items.text,
695 schema: self.fields.text,
696 mode: self.mode.to_wire(),
697 data_adapter: self.data_adapter,
698 selector: self.selector,
699 requested_buffer_size: self.buffer_size.to_wire(),
700 requested_max_frequency: match self.max_frequency {
701 // Omitting the parameter and asking for `unlimited` mean the
702 // same thing to the server, and omitting it is the smaller
703 // request [`docs/spec/03-requests.md` §6.1].
704 MaxFrequency::Unlimited => None,
705 ref other => Some(other.to_wire()),
706 },
707 snapshot: match self.snapshot {
708 // Likewise: `false` is the documented default.
709 Snapshot::Off => None,
710 other => Some(other.to_wire()),
711 },
712 }
713 }
714}
715
716#[cfg(test)]
717mod tests {
718 use super::*;
719
720 fn group() -> ItemGroup {
721 match ItemGroup::from_items(["item1", "item2"]) {
722 Ok(group) => group,
723 Err(error) => unreachable!("the fixture group is valid: {error}"),
724 }
725 }
726
727 fn schema() -> FieldSchema {
728 match FieldSchema::from_fields(["last_price", "time"]) {
729 Ok(schema) => schema,
730 Err(error) => unreachable!("the fixture schema is valid: {error}"),
731 }
732 }
733
734 #[test]
735 fn test_item_group_joins_a_list_with_spaces() {
736 assert!(matches!(
737 ItemGroup::from_items(["a", "b", "c"]),
738 Ok(group) if group.as_str() == "a b c"
739 ));
740 }
741
742 #[test]
743 fn test_item_group_rejects_an_empty_list() {
744 let empty: [&str; 0] = [];
745 assert!(matches!(
746 ItemGroup::from_items(empty),
747 Err(ConfigError::EmptyItemGroup)
748 ));
749 }
750
751 #[test]
752 fn test_item_group_rejects_a_name_with_whitespace() {
753 assert!(matches!(
754 ItemGroup::from_items(["ok", "not ok"]),
755 Err(ConfigError::ItemNameHasWhitespace { name }) if name == "not ok"
756 ));
757 }
758
759 #[test]
760 fn test_item_group_rejects_an_empty_name_in_a_list() {
761 assert!(matches!(
762 ItemGroup::from_items(["ok", ""]),
763 Err(ConfigError::ItemNameHasWhitespace { .. })
764 ));
765 }
766
767 #[test]
768 fn test_item_group_rejects_an_empty_group_name() {
769 assert!(matches!(
770 ItemGroup::try_new(" "),
771 Err(ConfigError::EmptyItemGroup)
772 ));
773 }
774
775 #[test]
776 fn test_item_group_rejects_a_control_character() {
777 assert!(matches!(
778 ItemGroup::try_new("a\rb"),
779 Err(ConfigError::ItemGroupControlCharacter)
780 ));
781 }
782
783 #[test]
784 fn test_field_schema_rejects_every_bad_shape() {
785 let empty: [&str; 0] = [];
786 assert!(matches!(
787 FieldSchema::from_fields(empty),
788 Err(ConfigError::EmptyFieldSchema)
789 ));
790 assert!(matches!(
791 FieldSchema::from_fields(["a b"]),
792 Err(ConfigError::FieldNameHasWhitespace { .. })
793 ));
794 assert!(matches!(
795 FieldSchema::try_new(""),
796 Err(ConfigError::EmptyFieldSchema)
797 ));
798 assert!(matches!(
799 FieldSchema::try_new("a\nb"),
800 Err(ConfigError::FieldSchemaControlCharacter)
801 ));
802 }
803
804 #[test]
805 fn test_field_schema_lists_its_names() {
806 assert_eq!(schema().names(), vec!["last_price", "time"]);
807 }
808
809 #[test]
810 fn test_max_frequency_accepts_a_decimal_number() {
811 assert!(MaxFrequency::limited("2").is_ok());
812 assert!(MaxFrequency::limited("0.5").is_ok());
813 }
814
815 #[test]
816 fn test_max_frequency_rejects_anything_that_is_not_one() {
817 for bad in ["2,0", ".5", "5.", "1e3", "-1", ""] {
818 assert!(
819 matches!(
820 MaxFrequency::limited(bad),
821 Err(ConfigError::InvalidFrequency { .. })
822 ),
823 "{bad} was accepted"
824 );
825 }
826 }
827
828 #[test]
829 fn test_snapshot_knows_when_it_asks_for_one() {
830 assert!(!Snapshot::Off.is_requested());
831 assert!(Snapshot::On.is_requested());
832 assert!(Snapshot::Length(NonZeroU32::MIN).is_requested());
833 }
834
835 #[test]
836 fn test_subscription_defaults_omit_every_optional_parameter() {
837 let spec = Subscription::new(SubscriptionMode::Merge, group(), schema()).into_spec();
838 assert_eq!(spec.group, "item1 item2");
839 assert_eq!(spec.schema, "last_price time");
840 assert_eq!(spec.mode, WireMode::Merge);
841 assert_eq!(spec.data_adapter, None);
842 assert_eq!(spec.selector, None);
843 assert_eq!(spec.snapshot, None);
844 assert_eq!(spec.requested_max_frequency, None);
845 assert_eq!(spec.requested_buffer_size, None);
846 }
847
848 #[test]
849 fn test_subscription_carries_every_option_it_was_given() {
850 let frequency = match MaxFrequency::limited("2.5") {
851 Ok(frequency) => frequency,
852 Err(error) => unreachable!("the fixture frequency is valid: {error}"),
853 };
854 let spec = Subscription::new(SubscriptionMode::Distinct, group(), schema())
855 .with_data_adapter("QUOTE_ADAPTER")
856 .with_selector("only_big")
857 .with_snapshot(Snapshot::Length(NonZeroU32::MIN))
858 .with_max_frequency(frequency)
859 .with_buffer_size(BufferSize::Unlimited)
860 .into_spec();
861
862 assert_eq!(spec.mode, WireMode::Distinct);
863 assert_eq!(spec.data_adapter.as_deref(), Some("QUOTE_ADAPTER"));
864 assert_eq!(spec.selector.as_deref(), Some("only_big"));
865 assert_eq!(spec.snapshot, Some(WireSnapshot::Length(NonZeroU32::MIN)));
866 assert_eq!(
867 spec.requested_buffer_size,
868 Some(RequestedBufferSize::Unlimited)
869 );
870 assert!(matches!(
871 spec.requested_max_frequency,
872 Some(RequestedMaxFrequency::Limited(_))
873 ));
874 }
875
876 // -----------------------------------------------------------------------
877 // A-02: every mode-specific constraint fails at the public boundary
878 // -----------------------------------------------------------------------
879
880 fn subscription(mode: SubscriptionMode) -> Subscription {
881 Subscription::new(mode, group(), schema())
882 }
883
884 #[test]
885 fn test_a_plain_subscription_validates_in_every_mode() {
886 for mode in [
887 SubscriptionMode::Raw,
888 SubscriptionMode::Merge,
889 SubscriptionMode::Distinct,
890 SubscriptionMode::Command,
891 ] {
892 assert!(
893 subscription(mode).validate().is_ok(),
894 "{} was rejected",
895 mode.as_str()
896 );
897 }
898 }
899
900 #[test]
901 fn test_a_snapshot_length_is_refused_outside_distinct_mode() {
902 // The spec's own admissibility rule [`docs/spec/03-requests.md` §6.1].
903 for mode in [
904 SubscriptionMode::Raw,
905 SubscriptionMode::Merge,
906 SubscriptionMode::Command,
907 ] {
908 let built = subscription(mode).with_snapshot(Snapshot::Length(NonZeroU32::MIN));
909 assert!(
910 matches!(
911 built.validate(),
912 Err(ConfigError::SnapshotLengthNeedsDistinct { mode: named }) if named == mode.as_str()
913 ),
914 "{} accepted a snapshot length",
915 mode.as_str()
916 );
917 }
918 assert!(
919 subscription(SubscriptionMode::Distinct)
920 .with_snapshot(Snapshot::Length(NonZeroU32::MIN))
921 .validate()
922 .is_ok()
923 );
924 }
925
926 #[test]
927 fn test_a_snapshot_is_refused_in_raw_mode() {
928 assert!(matches!(
929 subscription(SubscriptionMode::Raw)
930 .with_snapshot(Snapshot::On)
931 .validate(),
932 Err(ConfigError::SnapshotNeedsAStatefulMode)
933 ));
934 // `Snapshot::Off` is the default and asks for nothing, so it stays legal.
935 assert!(
936 subscription(SubscriptionMode::Raw)
937 .with_snapshot(Snapshot::Off)
938 .validate()
939 .is_ok()
940 );
941 }
942
943 #[test]
944 fn test_a_frequency_is_refused_in_raw_mode() {
945 let Ok(limited) = MaxFrequency::limited("2.0") else {
946 unreachable!("the fixture frequency is valid");
947 };
948 for frequency in [MaxFrequency::Unfiltered, limited] {
949 assert!(
950 matches!(
951 subscription(SubscriptionMode::Raw)
952 .with_max_frequency(frequency.clone())
953 .validate(),
954 Err(ConfigError::FrequencyNeedsAFilteredMode)
955 ),
956 "{frequency:?} was accepted with RAW"
957 );
958 }
959 assert!(
960 subscription(SubscriptionMode::Raw)
961 .with_max_frequency(MaxFrequency::Unlimited)
962 .validate()
963 .is_ok()
964 );
965 }
966
967 #[test]
968 fn test_a_buffer_size_is_refused_outside_merge_and_distinct() {
969 for mode in [SubscriptionMode::Raw, SubscriptionMode::Command] {
970 assert!(
971 matches!(
972 subscription(mode)
973 .with_buffer_size(BufferSize::Unlimited)
974 .validate(),
975 Err(ConfigError::BufferSizeNeedsMergeOrDistinct { mode: named }) if named == mode.as_str()
976 ),
977 "{} accepted a buffer size",
978 mode.as_str()
979 );
980 }
981 for mode in [SubscriptionMode::Merge, SubscriptionMode::Distinct] {
982 assert!(
983 subscription(mode)
984 .with_buffer_size(BufferSize::Events(NonZeroU32::MIN))
985 .validate()
986 .is_ok(),
987 "{} rejected a buffer size",
988 mode.as_str()
989 );
990 }
991 }
992
993 #[test]
994 fn test_a_buffer_size_is_refused_alongside_an_unfiltered_frequency() {
995 assert!(matches!(
996 subscription(SubscriptionMode::Merge)
997 .with_max_frequency(MaxFrequency::Unfiltered)
998 .with_buffer_size(BufferSize::Unlimited)
999 .validate(),
1000 Err(ConfigError::BufferSizeWithUnfilteredFrequency)
1001 ));
1002 // Either alone is fine; it is the pair the server ignores.
1003 assert!(
1004 subscription(SubscriptionMode::Merge)
1005 .with_max_frequency(MaxFrequency::Unfiltered)
1006 .validate()
1007 .is_ok()
1008 );
1009 }
1010
1011 #[test]
1012 fn test_subscription_mode_spells_itself_the_way_the_protocol_does() {
1013 assert_eq!(SubscriptionMode::Raw.as_str(), "RAW");
1014 assert_eq!(SubscriptionMode::Merge.as_str(), "MERGE");
1015 assert_eq!(SubscriptionMode::Distinct.as_str(), "DISTINCT");
1016 assert_eq!(SubscriptionMode::Command.as_str(), "COMMAND");
1017 }
1018}