use std::fmt;
use std::ops::Range;
use std::sync::Arc;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum FieldValue<'a> {
Null,
Text(&'a str),
}
impl<'a> FieldValue<'a> {
#[must_use]
#[inline]
pub const fn is_null(self) -> bool {
matches!(self, Self::Null)
}
#[must_use]
#[inline]
pub fn is_empty_text(self) -> bool {
matches!(self, Self::Text(text) if text.is_empty())
}
#[must_use]
#[inline]
pub const fn text(self) -> Option<&'a str> {
match self {
Self::Null => None,
Self::Text(text) => Some(text),
}
}
#[must_use]
#[inline]
pub fn text_or(self, default: &'a str) -> &'a str {
self.text().unwrap_or(default)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum ItemCommand {
Add,
Update,
Delete,
Unrecognized(
String,
),
}
impl ItemCommand {
#[must_use]
fn from_wire(value: &str) -> Self {
if value.eq_ignore_ascii_case("ADD") {
Self::Add
} else if value.eq_ignore_ascii_case("UPDATE") {
Self::Update
} else if value.eq_ignore_ascii_case("DELETE") {
Self::Delete
} else {
Self::Unrecognized(value.to_owned())
}
}
#[must_use]
#[inline]
pub fn as_str(&self) -> &str {
match self {
Self::Add => "ADD",
Self::Update => "UPDATE",
Self::Delete => "DELETE",
Self::Unrecognized(literal) => literal,
}
}
}
impl fmt::Display for ItemCommand {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum UpdateKind {
Snapshot,
RealTime,
}
impl UpdateKind {
#[must_use]
#[inline]
pub const fn is_snapshot(self) -> bool {
matches!(self, Self::Snapshot)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct SchemaName {
text: String,
declared: bool,
}
impl SchemaName {
#[must_use]
fn declared(text: impl Into<String>) -> Self {
Self {
text: text.into(),
declared: true,
}
}
#[must_use]
fn positional(position: usize) -> Self {
Self {
text: position.to_string(),
declared: false,
}
}
#[must_use]
#[inline]
fn as_str(&self) -> &str {
&self.text
}
#[must_use]
#[inline]
fn declared_str(&self) -> Option<&str> {
if self.declared {
Some(&self.text)
} else {
None
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct SubscriptionSchema {
items: Vec<SchemaName>,
fields: Vec<SchemaName>,
key_field: Option<usize>,
command_field: Option<usize>,
}
impl SubscriptionSchema {
#[must_use]
pub(crate) fn new(
item_count: usize,
field_count: usize,
declared_items: &[String],
declared_fields: &[String],
command_fields: Option<(usize, usize)>,
) -> Self {
let (key_field, command_field) = match command_fields {
Some((key, command)) => (Some(key), Some(command)),
None => (None, None),
};
Self {
items: Self::name_positions(item_count, declared_items),
fields: Self::name_positions(field_count, declared_fields),
key_field,
command_field,
}
}
#[cfg(feature = "test-util")]
#[must_use]
pub(crate) fn from_optional_names(
items: Vec<Option<String>>,
fields: Vec<Option<String>>,
command_fields: Option<(usize, usize)>,
) -> Self {
fn named(slots: Vec<Option<String>>) -> Vec<SchemaName> {
slots
.into_iter()
.enumerate()
.map(|(index, name)| match name {
Some(text) => SchemaName::declared(text),
None => SchemaName::positional(index + 1),
})
.collect()
}
let (key_field, command_field) = match command_fields {
Some((key, command)) => (Some(key), Some(command)),
None => (None, None),
};
Self {
items: named(items),
fields: named(fields),
key_field,
command_field,
}
}
#[must_use]
fn name_positions(count: usize, declared: &[String]) -> Vec<SchemaName> {
let mut names = Vec::with_capacity(count);
for index in 0..count {
match declared.get(index) {
Some(name) => names.push(SchemaName::declared(name.clone())),
None => names.push(SchemaName::positional(index + 1)),
}
}
names
}
#[allow(dead_code)]
#[must_use]
#[inline]
pub(crate) fn item_count(&self) -> usize {
self.items.len()
}
#[must_use]
#[inline]
pub(crate) fn field_count(&self) -> usize {
self.fields.len()
}
#[must_use]
#[inline]
pub(crate) fn is_command_mode(&self) -> bool {
self.key_field.is_some() && self.command_field.is_some()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ItemUpdate {
schema: Arc<SubscriptionSchema>,
item: usize,
kind: UpdateKind,
values: Vec<Option<String>>,
changed: Vec<usize>,
}
impl ItemUpdate {
#[must_use]
pub(crate) fn new(
schema: Arc<SubscriptionSchema>,
item: usize,
kind: UpdateKind,
values: Vec<Option<String>>,
changed: Vec<usize>,
) -> Self {
Self {
schema,
item,
kind,
values,
changed,
}
}
#[must_use]
#[inline]
pub const fn item_index(&self) -> usize {
self.item + 1
}
#[must_use]
#[inline]
pub fn item_name(&self) -> &str {
self.schema
.items
.get(self.item)
.map_or("", SchemaName::as_str)
}
#[must_use]
#[inline]
pub fn declared_item_name(&self) -> Option<&str> {
self.schema
.items
.get(self.item)
.and_then(SchemaName::declared_str)
}
#[must_use]
#[inline]
pub const fn kind(&self) -> UpdateKind {
self.kind
}
#[must_use]
#[inline]
pub const fn is_snapshot(&self) -> bool {
self.kind.is_snapshot()
}
#[must_use]
#[inline]
pub fn field_count(&self) -> usize {
self.schema.field_count()
}
#[must_use]
#[inline]
pub fn field_name(&self, position: usize) -> Option<&str> {
self.schema
.fields
.get(position.checked_sub(1)?)
.map(SchemaName::as_str)
}
#[must_use]
pub fn field(&self, position: usize) -> Option<FieldValue<'_>> {
let index = position.checked_sub(1)?;
Some(match self.values.get(index)? {
None => FieldValue::Null,
Some(text) => FieldValue::Text(text),
})
}
#[must_use]
pub fn field_by_name(&self, name: &str) -> Option<FieldValue<'_>> {
self.field(self.field_position(name)?)
}
#[must_use]
pub fn field_position(&self, name: &str) -> Option<usize> {
self.schema
.fields
.iter()
.position(|field| field.declared_str() == Some(name))
.map(|index| index + 1)
}
#[must_use]
#[inline]
pub fn fields(&self) -> Fields<'_> {
Fields {
update: self,
positions: 0..self.values.len(),
}
}
#[must_use]
#[inline]
pub fn changed_fields(&self) -> ChangedFields<'_> {
ChangedFields {
update: self,
positions: self.changed.iter(),
}
}
#[must_use]
#[inline]
pub fn changed_count(&self) -> usize {
self.changed.len()
}
#[must_use]
pub fn is_field_changed(&self, position: usize) -> bool {
position
.checked_sub(1)
.is_some_and(|index| self.changed.contains(&index))
}
#[must_use]
pub fn is_field_changed_by_name(&self, name: &str) -> bool {
self.field_position(name)
.is_some_and(|position| self.is_field_changed(position))
}
#[must_use]
#[inline]
pub fn is_command_mode(&self) -> bool {
self.schema.is_command_mode()
}
#[must_use]
pub fn key(&self) -> Option<FieldValue<'_>> {
let index = self.schema.key_field?;
self.field(index.checked_add(1)?)
}
#[must_use]
pub fn command(&self) -> Option<ItemCommand> {
let index = self.schema.command_field?;
let value = self.field(index.checked_add(1)?)?;
value.text().map(ItemCommand::from_wire)
}
#[must_use]
pub fn is_command_changed(&self) -> bool {
self.schema
.command_field
.and_then(|index| index.checked_add(1))
.is_some_and(|position| self.is_field_changed(position))
}
#[must_use]
pub fn is_key_changed(&self) -> bool {
self.schema
.key_field
.and_then(|index| index.checked_add(1))
.is_some_and(|position| self.is_field_changed(position))
}
}
#[derive(Clone, Copy)]
pub struct Field<'a> {
update: &'a ItemUpdate,
index: usize,
}
impl<'a> Field<'a> {
#[must_use]
#[inline]
pub const fn position(&self) -> usize {
self.index + 1
}
#[must_use]
#[inline]
pub fn name(&self) -> &'a str {
self.update
.schema
.fields
.get(self.index)
.map_or("", SchemaName::as_str)
}
#[must_use]
#[inline]
pub fn declared_name(&self) -> Option<&'a str> {
self.update
.schema
.fields
.get(self.index)
.and_then(SchemaName::declared_str)
}
#[must_use]
#[inline]
pub fn value(&self) -> FieldValue<'a> {
match self.update.values.get(self.index) {
Some(Some(text)) => FieldValue::Text(text),
Some(None) | None => FieldValue::Null,
}
}
#[must_use]
#[inline]
pub fn is_changed(&self) -> bool {
self.update.changed.contains(&self.index)
}
}
impl fmt::Debug for Field<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Field")
.field("position", &self.position())
.field("name", &self.name())
.field("value", &self.value())
.field("changed", &self.is_changed())
.finish()
}
}
#[derive(Clone)]
pub struct Fields<'a> {
update: &'a ItemUpdate,
positions: Range<usize>,
}
impl<'a> Iterator for Fields<'a> {
type Item = Field<'a>;
fn next(&mut self) -> Option<Self::Item> {
let index = self.positions.next()?;
Some(Field {
update: self.update,
index,
})
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.positions.size_hint()
}
}
impl ExactSizeIterator for Fields<'_> {}
impl fmt::Debug for Fields<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_map()
.entries(self.clone().map(|field| (field.name(), field.value())))
.finish()
}
}
#[derive(Clone)]
pub struct ChangedFields<'a> {
update: &'a ItemUpdate,
positions: std::slice::Iter<'a, usize>,
}
impl<'a> Iterator for ChangedFields<'a> {
type Item = Field<'a>;
fn next(&mut self) -> Option<Self::Item> {
let index = *self.positions.next()?;
Some(Field {
update: self.update,
index,
})
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.positions.size_hint()
}
}
impl ExactSizeIterator for ChangedFields<'_> {}
impl fmt::Debug for ChangedFields<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_map()
.entries(self.clone().map(|field| (field.name(), field.value())))
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn names(values: &[&str]) -> Vec<String> {
values.iter().map(|value| (*value).to_owned()).collect()
}
fn quote_schema() -> Arc<SubscriptionSchema> {
Arc::new(SubscriptionSchema::new(
2,
3,
&names(&["item1", "item2"]),
&names(&["last_price", "time", "status"]),
None,
))
}
fn update_with(values: Vec<Option<String>>, changed: Vec<usize>) -> ItemUpdate {
ItemUpdate::new(quote_schema(), 0, UpdateKind::RealTime, values, changed)
}
#[test]
fn test_field_value_null_and_empty_text_are_distinct() {
assert_ne!(FieldValue::Null, FieldValue::Text(""));
assert!(FieldValue::Null.is_null());
assert!(!FieldValue::Null.is_empty_text());
assert!(FieldValue::Text("").is_empty_text());
assert!(!FieldValue::Text("").is_null());
assert_eq!(FieldValue::Null.text(), None);
assert_eq!(FieldValue::Text("").text(), Some(""));
}
#[test]
fn test_field_value_text_or_substitutes_only_for_null() {
assert_eq!(FieldValue::Null.text_or("-"), "-");
assert_eq!(FieldValue::Text("").text_or("-"), "");
}
#[test]
fn test_null_and_empty_reach_the_caller_distinctly() {
let update = update_with(
vec![None, Some(String::new()), Some("live".to_owned())],
vec![0, 1, 2],
);
assert_eq!(update.field(1), Some(FieldValue::Null));
assert_eq!(update.field(2), Some(FieldValue::Text("")));
assert_eq!(update.field_by_name("last_price"), Some(FieldValue::Null));
assert_eq!(update.field_by_name("time"), Some(FieldValue::Text("")));
assert_ne!(update.field(1), update.field(2));
}
#[test]
fn test_every_rendering_of_a_field_value_keeps_null_distinguishable() {
assert_eq!(FieldValue::Null.text_or("(null)"), "(null)");
assert_eq!(FieldValue::Text("").text_or("(null)"), "");
assert_eq!(FieldValue::Text("x").text_or("(null)"), "x");
assert_eq!(format!("{:?}", FieldValue::Null), "Null");
assert_eq!(format!("{:?}", FieldValue::Text("")), r#"Text("")"#);
}
#[test]
fn test_positions_are_one_based() {
let update = update_with(vec![Some("a".to_owned()), None, None], vec![0]);
assert_eq!(update.item_index(), 1);
assert_eq!(update.field(1), Some(FieldValue::Text("a")));
assert_eq!(update.field(0), None);
assert_eq!(update.field(4), None);
assert_eq!(update.field_name(1), Some("last_price"));
assert_eq!(update.field_name(0), None);
}
#[test]
fn test_names_come_from_the_caller() {
let update = update_with(vec![None, None, None], vec![]);
assert_eq!(update.item_name(), "item1");
assert_eq!(update.declared_item_name(), Some("item1"));
assert_eq!(update.field_position("status"), Some(3));
}
#[test]
fn test_undeclared_names_fall_back_to_the_position() {
let schema = Arc::new(SubscriptionSchema::new(1, 2, &[], &[], None));
let update = ItemUpdate::new(
schema,
0,
UpdateKind::RealTime,
vec![Some("a".to_owned()), None],
vec![0],
);
assert_eq!(update.item_name(), "1");
assert_eq!(update.declared_item_name(), None);
assert_eq!(update.field_name(2), Some("2"));
assert_eq!(update.field_by_name("2"), None);
assert_eq!(update.field_position("1"), None);
}
#[test]
fn test_partially_declared_names_are_taken_positionally() {
let schema = Arc::new(SubscriptionSchema::new(
1,
3,
&names(&["only_item"]),
&names(&["first"]),
None,
));
let update = ItemUpdate::new(schema, 0, UpdateKind::RealTime, vec![None; 3], vec![]);
assert_eq!(update.field_name(1), Some("first"));
assert_eq!(update.field_name(2), Some("2"));
assert_eq!(update.field_position("first"), Some(1));
}
#[test]
fn test_unknown_field_name_is_none_everywhere() {
let update = update_with(vec![Some("a".to_owned()), None, None], vec![0]);
assert_eq!(update.field_by_name("no_such_field"), None);
assert_eq!(update.field_position("no_such_field"), None);
assert!(!update.is_field_changed_by_name("no_such_field"));
}
#[test]
fn test_duplicate_declared_names_resolve_to_the_first() {
let schema = Arc::new(SubscriptionSchema::new(
1,
2,
&names(&["i"]),
&names(&["dup", "dup"]),
None,
));
let update = ItemUpdate::new(
schema,
0,
UpdateKind::RealTime,
vec![Some("first".to_owned()), Some("second".to_owned())],
vec![0, 1],
);
assert_eq!(update.field_position("dup"), Some(1));
assert_eq!(update.field_by_name("dup"), Some(FieldValue::Text("first")));
}
#[test]
fn test_changed_fields_reports_only_carried_tokens() {
let update = update_with(
vec![
Some("3.04".to_owned()),
Some("20:00:33".to_owned()),
Some(String::new()),
],
vec![0, 2],
);
let changed: Vec<(usize, &str)> = update
.changed_fields()
.map(|field| (field.position(), field.name()))
.collect();
assert_eq!(changed, vec![(1, "last_price"), (3, "status")]);
assert_eq!(update.changed_count(), 2);
assert!(update.is_field_changed(1));
assert!(!update.is_field_changed(2));
assert!(!update.is_field_changed(0));
assert!(update.is_field_changed_by_name("status"));
assert!(!update.is_field_changed_by_name("time"));
}
#[test]
fn test_fields_iterates_the_whole_schema_with_changed_flags() {
let update = update_with(
vec![Some("a".to_owned()), None, Some(String::new())],
vec![0, 2],
);
let seen: Vec<(usize, bool)> = update
.fields()
.map(|field| (field.position(), field.is_changed()))
.collect();
assert_eq!(seen, vec![(1, true), (2, false), (3, true)]);
assert_eq!(update.fields().len(), 3);
assert_eq!(update.changed_fields().len(), 2);
}
#[test]
fn test_changed_fields_debug_is_a_readable_map() {
let update = update_with(
vec![Some("3.04".to_owned()), None, Some(String::new())],
vec![0, 1],
);
let rendered = format!("{:?}", update.changed_fields());
assert!(rendered.contains("last_price"), "{rendered}");
assert!(rendered.contains("Text(\"3.04\")"), "{rendered}");
assert!(rendered.contains("Null"), "{rendered}");
}
#[test]
fn test_adr_0003_usage_compiles_and_prints() {
let update = update_with(vec![Some("3.04".to_owned()), None, None], vec![0]);
let line = format!("{}: {:?}", update.item_name(), update.changed_fields());
assert!(line.starts_with("item1: "), "{line}");
}
fn command_update(key: &str, command: &str) -> ItemUpdate {
let schema = Arc::new(SubscriptionSchema::new(
1,
3,
&names(&["portfolio"]),
&names(&["key", "command", "qty"]),
Some((0, 1)),
));
ItemUpdate::new(
schema,
0,
UpdateKind::RealTime,
vec![
Some(key.to_owned()),
Some(command.to_owned()),
Some("10".to_owned()),
],
vec![0, 1, 2],
)
}
#[test]
fn test_command_mode_exposes_key_and_command() {
let update = command_update("AAPL", "ADD");
assert!(update.is_command_mode());
assert_eq!(update.key(), Some(FieldValue::Text("AAPL")));
assert_eq!(update.command(), Some(ItemCommand::Add));
}
#[test]
fn test_command_literals_are_case_insensitive() {
assert_eq!(command_update("k", "add").command(), Some(ItemCommand::Add));
assert_eq!(
command_update("k", "Update").command(),
Some(ItemCommand::Update)
);
assert_eq!(
command_update("k", "DELETE").command(),
Some(ItemCommand::Delete)
);
}
#[test]
fn test_unrecognized_command_is_preserved_not_rejected() {
let update = command_update("k", "MERGE_ROW");
assert_eq!(
update.command(),
Some(ItemCommand::Unrecognized("MERGE_ROW".to_owned()))
);
assert_eq!(
update.command().as_ref().map(ItemCommand::as_str),
Some("MERGE_ROW")
);
}
#[test]
fn test_command_names_render_uppercase() {
assert_eq!(ItemCommand::Add.to_string(), "ADD");
assert_eq!(ItemCommand::Update.as_str(), "UPDATE");
assert_eq!(ItemCommand::Delete.as_str(), "DELETE");
}
#[test]
fn test_non_command_subscription_has_no_key_or_command() {
let update = update_with(vec![Some("a".to_owned()), None, None], vec![0]);
assert!(!update.is_command_mode());
assert_eq!(update.key(), None);
assert_eq!(update.command(), None);
assert!(!update.is_key_changed());
assert!(!update.is_command_changed());
}
#[test]
fn test_a_retained_command_is_reported_as_not_changed() {
let schema = Arc::new(SubscriptionSchema::new(
1,
3,
&names(&["portfolio"]),
&names(&["key", "command", "qty"]),
Some((0, 1)),
));
let update = ItemUpdate::new(
schema,
0,
UpdateKind::RealTime,
vec![
Some("AAPL".to_owned()),
Some("DELETE".to_owned()),
Some("10".to_owned()),
],
vec![2],
);
assert_eq!(update.key(), Some(FieldValue::Text("AAPL")));
assert_eq!(update.command(), Some(ItemCommand::Delete));
assert!(!update.is_key_changed());
assert!(!update.is_command_changed());
}
#[test]
fn test_a_carried_command_is_reported_as_changed() {
let update = command_update("AAPL", "DELETE");
assert!(update.is_key_changed());
assert!(update.is_command_changed());
}
#[test]
fn test_null_key_is_reported_as_null_not_missing() {
let schema = Arc::new(SubscriptionSchema::new(
1,
2,
&[],
&names(&["key", "command"]),
Some((0, 1)),
));
let update = ItemUpdate::new(
schema,
0,
UpdateKind::RealTime,
vec![None, None],
vec![0, 1],
);
assert_eq!(update.key(), Some(FieldValue::Null));
assert_eq!(update.command(), None);
}
#[test]
fn test_update_kind_accessors() {
let schema = quote_schema();
let snapshot = ItemUpdate::new(
Arc::clone(&schema),
0,
UpdateKind::Snapshot,
vec![None; 3],
vec![],
);
let live = ItemUpdate::new(schema, 1, UpdateKind::RealTime, vec![None; 3], vec![]);
assert!(snapshot.is_snapshot());
assert_eq!(snapshot.kind(), UpdateKind::Snapshot);
assert!(!live.is_snapshot());
assert_eq!(live.item_index(), 2);
assert_eq!(live.item_name(), "item2");
}
}