use crate::message::filter::MessageFilter;
use crate::storage::ConnectionType;
use crate::{Error, Result};
use crate::{Message, storage::Db3File};
use cdr::CdrBe;
use serde::{Deserialize, Serialize};
pub struct MessageWriter<'a> {
connections: &'a ConnectionType,
}
impl<'a> MessageWriter<'a> {
pub fn create(connections: &'a crate::storage::ConnectionType) -> MessageWriter<'a> {
Self { connections }
}
pub async fn new_topic(&self, name: &str, r#type: &str) -> Result<()> {
todo!("Insert a new topic into the topics table")
}
pub async fn rm_topics(&self, topics: impl IntoIterator<Item = &str>) -> Result<()> {
todo!("Remove a topic from the topics table")
}
pub async fn insert<T: Serialize + for<'c> Deserialize<'c>>(
&self,
topic: &str,
timestamp: i64,
msg: T,
) -> Result<()> {
let data = cdr::serialize::<_, _, CdrBe>(&msg, cdr::Infinite)?;
let db3 = match &self.connections {
ConnectionType::Atomic(db3_file) => db3_file,
ConnectionType::Splitted(splitted_rosbag) => {
Self::connection_for_timestamp(&splitted_rosbag.connections, timestamp).await?
}
};
let topic_id = db3.topic_id(topic)?;
db3.insert_message(topic_id, timestamp, data).await?;
Ok(())
}
async fn connection_for_timestamp(connections: &[Db3File], timestamp: i64) -> Result<&Db3File> {
let mut newest: Option<(i64, usize)> = None;
for (idx, db3_file) in connections.iter().enumerate() {
let (min, max) = db3_file.time_range().await?;
if min.is_some_and(|min| timestamp >= min) && max.is_some_and(|max| timestamp <= max) {
return Ok(db3_file);
}
if let Some(max) = max {
if newest.is_none_or(|(newest_max, _)| max > newest_max) {
newest = Some((max, idx));
}
}
}
let (_, idx) = newest.ok_or(Error::NoDb3File(std::path::PathBuf::new()))?;
Ok(&connections[idx])
}
pub async fn pop_by_filter<M: Message>(&self, filter: &MessageFilter<'_>) -> Result<M> {
todo!("Remove and return the last message of the given topic")
}
pub async fn pop<M: Message>(&self) -> Result<M> {
todo!("Remove and return the last message of the bag")
}
pub async fn pop_front<M: Message>(&self) -> Result<M> {
todo!("Remove and return the first message of the bag")
}
}