use std::{collections::{HashMap, VecDeque}, fmt::{Debug, Display}, time::Duration};
use serde::{Deserialize, Serialize};
use crate::{ClientHandle, data::{DataType, NetworkTableData}, error::ConnectionClosedError, net::{Announce, Unannounce}, publish::{GenericPublisher, NewPublisherError, Publisher}, subscribe::{Subscriber, SubscriptionOptions}};
pub mod collection;
#[macro_export]
macro_rules! path {
() => {
$crate::topic::TopicPath::default();
};
($($segment: literal),+ $(,)?) => {{
let mut segments = std::collections::VecDeque::new();
$(
segments.push_back($segment.to_string());
)*
$crate::topic::TopicPath::new(segments)
}};
}
#[derive(Debug, Clone)]
pub struct Topic {
name: String,
handle: ClientHandle,
}
impl PartialEq for Topic {
fn eq(&self, other: &Self) -> bool {
self.name == other.name
}
}
impl Eq for Topic { }
impl Topic {
pub(super) fn new(
name: String,
handle: ClientHandle
) -> Self {
Self { name, handle }
}
pub fn name(&self) -> &str {
&self.name
}
pub fn name_mut(&mut self) -> &mut String {
&mut self.name
}
pub fn child(&self, name: impl AsRef<str>) -> Self {
Self::new(self.name.clone() + name.as_ref(), self.handle.clone())
}
pub async fn publish<T: NetworkTableData>(&self, properties: Properties) -> Result<Publisher<T>, NewPublisherError> {
Publisher::new(self.name.clone(), properties, self.handle.time(), self.handle.server_send.clone(), self.handle.client_send.subscribe()).await
}
#[cfg(feature = "publish_bypass")]
pub async fn publish_bypass<T: NetworkTableData>(&self, properties: Properties) -> Result<Publisher<T>, ConnectionClosedError> {
Publisher::new_bypass(self.name.clone(), properties, self.handle.time(), self.handle.server_send.clone(), self.handle.client_send.subscribe()).await
}
pub async fn generic_publish(&self, r#type: DataType, properties: Properties) -> Result<GenericPublisher, NewPublisherError> {
GenericPublisher::new(self.name.clone(), properties, r#type, self.handle.time(), self.handle.server_send.clone(), self.handle.client_send.subscribe()).await
}
#[cfg(feature = "publish_bypass")]
pub async fn generic_publish_bypass(&self, r#type: DataType, properties: Properties) -> Result<GenericPublisher, ConnectionClosedError> {
GenericPublisher::new_bypass(self.name.clone(), properties, r#type, self.handle.time(), self.handle.server_send.clone(), self.handle.client_send.subscribe()).await
}
pub async fn subscribe(&self, options: SubscriptionOptions) -> Result<Subscriber, ConnectionClosedError> {
Subscriber::new(vec![self.name.clone()], options, self.handle.announced_topics.clone(), self.handle.server_send.clone(), self.handle.client_send.subscribe()).await
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct AnnouncedTopic {
name: String,
id: i32,
r#type: DataType,
pub(crate) properties: Properties,
value: Option<rmpv::Value>,
last_updated: Option<Duration>,
}
impl AnnouncedTopic {
pub fn name(&self) -> &str {
&self.name
}
pub fn id(&self) -> i32 {
self.id
}
pub fn r#type(&self) -> &DataType {
&self.r#type
}
pub fn properties(&self) -> &Properties {
&self.properties
}
pub fn value(&self) -> Option<&rmpv::Value> {
self.value.as_ref()
}
pub fn last_updated(&self) -> Option<&Duration> {
self.last_updated.as_ref()
}
pub(crate) fn update(&mut self, when: Duration) {
self.last_updated = Some(when);
}
pub(crate) fn update_value(&mut self, value: rmpv::Value) {
self.value = Some(value);
}
pub fn matches(&self, names: &[String], options: &SubscriptionOptions) -> bool {
names.iter()
.any(|name| &self.name == name || (options.prefix.is_some_and(|flag| flag) && self.name.starts_with(name)))
}
}
impl From<&Announce> for AnnouncedTopic {
fn from(value: &Announce) -> Self {
Self {
name: value.name.clone(),
id: value.id,
r#type: value.r#type.clone(),
properties: value.properties.clone(),
value: None,
last_updated: None,
}
}
}
#[derive(Default, Debug, Clone, PartialEq)]
pub struct AnnouncedTopics {
topics: HashMap<i32, AnnouncedTopic>,
name_to_id: HashMap<String, i32>,
}
impl AnnouncedTopics {
pub fn new() -> Self {
Default::default()
}
pub(crate) fn insert(&mut self, announce: &Announce) {
self.topics.insert(announce.id, announce.into());
self.name_to_id.insert(announce.name.clone(), announce.id);
}
pub(crate) fn remove(&mut self, unannounce: &Unannounce) {
self.topics.remove(&unannounce.id);
self.name_to_id.remove(&unannounce.name);
}
pub fn get_from_id(&self, id: i32) -> Option<&AnnouncedTopic> {
self.topics.get(&id)
}
pub fn get_mut_from_id(&mut self, id: i32) -> Option<&mut AnnouncedTopic> {
self.topics.get_mut(&id)
}
pub fn get_from_name(&self, name: &str) -> Option<&AnnouncedTopic> {
self.name_to_id.get(name).and_then(|id| self.topics.get(id))
}
pub fn get_mut_from_name(&mut self, name: &str) -> Option<&mut AnnouncedTopic> {
self.name_to_id.get(name).and_then(|id| self.topics.get_mut(id))
}
pub fn get_id(&self, name: &str) -> Option<i32> {
self.name_to_id.get(name).copied()
}
pub fn id_values(&self) -> std::collections::hash_map::Values<'_, i32, AnnouncedTopic> {
self.topics.values()
}
}
#[derive(Serialize, Deserialize, Default, Debug, Clone, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub struct Properties {
#[serde(skip_serializing_if = "Option::is_none")]
pub persistent: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retained: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cached: Option<bool>,
#[serde(flatten)]
pub extra: HashMap<String, serde_json::Value>,
}
#[derive(Default, Debug, Clone, PartialEq, Eq, Hash)]
pub struct TopicPath {
pub segments: VecDeque<String>,
}
impl TopicPath {
pub const DELIMITER: char = '/';
pub fn new(segments: VecDeque<String>) -> Self {
Self { segments }
}
}
impl From<VecDeque<String>> for TopicPath {
fn from(value: VecDeque<String>) -> Self {
Self { segments: value }
}
}
impl Display for TopicPath {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let full_path = self.segments.iter().fold(String::new(), |prev, curr| prev + "/" + curr);
f.write_str(&full_path)
}
}
impl From<&str> for TopicPath {
fn from(value: &str) -> Self {
value.to_string().into()
}
}
impl From<String> for TopicPath {
fn from(value: String) -> Self {
let str = value.strip_prefix(Self::DELIMITER).unwrap_or(&value);
let str = str
.strip_suffix(Self::DELIMITER)
.map(|str| str.to_owned())
.unwrap_or_else(|| str.to_owned());
str.chars().fold((VecDeque::<String>::new(), true), |(mut parts, prev_is_delimiter), char| {
if prev_is_delimiter {
parts.push_back(String::from(char));
(parts, false)
} else {
let is_delimiter = char == Self::DELIMITER;
if !is_delimiter { parts.back_mut().unwrap().push(char); };
(parts, is_delimiter)
}
}).0.into()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_single_item() {
assert_eq!(into_path("Topic"), path!["Topic"]);
assert_eq!(into_path("123thing"), path!["123thing"]);
assert_eq!(into_path("/mydata"), path!["mydata"]);
assert_eq!(into_path("value/"), path!["value"]);
assert_eq!(into_path("//thing"), path!["/thing"]);
assert_eq!(into_path("cooldata//"), path!["cooldata"]);
}
#[test]
fn test_multi_item() {
assert_eq!(into_path("some/thing"), path!["some", "thing"]);
assert_eq!(into_path("Topic/thing/value"), path!["Topic", "thing", "value"]);
assert_eq!(into_path("/hello/there"), path!["hello", "there"]);
assert_eq!(into_path("my/long/path/"), path!["my", "long", "path"]);
assert_eq!(into_path("//weird///path/and/slash//"), path!["/weird", "/", "path", "and", "slash"]);
assert_eq!(into_path("//////"), path!["/", "/"]);
}
#[test]
fn test_parse_to_string() {
let path = path!["simple"];
assert_eq!(into_path(&path.to_string()), path);
let path = path!["my", "data"];
assert_eq!(into_path(&path.to_string()), path);
let path = path!["/something", "really", "/weird"];
assert_eq!(into_path(&path.to_string()), path);
}
fn into_path(s: &str) -> TopicPath {
s.into()
}
}