pub const SQL_STATEMENT_CREATE_TABLE: &'static str = "
CREATE TABLE schema (
schema_version INTEGER PRIMARY KEY,
ros_distro TEXT NOT NULL
);
";
pub const SQL_STATEMENT_CREATE_METADATA: &'static str = "
CREATE TABLE metadata (
id INTEGER PRIMARY KEY,
metadata_version INTEGER NOT NULL,
metadata TEXT NOT NULL
);
";
pub const SQL_STATEMENT_CREATE_TOPICS: &'static str = "
CREATE TABLE topics (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
type TEXT NOT NULL,
serialization_format TEXT NOT NULL,
offered_qos_profiles TEXT NOT NULL,
type_description_hash TEXT NOT NULL
);
";
pub const SQL_STATEMENT_CREATE_MSG_DEFINITIONS: &'static str = "
CREATE TABLE message_definitions (
id INTEGER PRIMARY KEY,
topic_type TEXT NOT NULL,
encoding TEXT NOT NULL,
encoded_message_definition TEXT NOT NULL,
type_description_hash TEXT NOT NULL
);
";
pub const SQL_STATEMENT_CREATE_MSGS: &'static str = "
CREATE TABLE messages (
id INTEGER PRIMARY KEY,
topic_id INTEGER NOT NULL,
timestamp INTEGER NOT NULL,
data BLOB NOT NULL,
FOREIGN KEY (topic_id) REFERENCES topics(id)
);
";
pub const SQL_STATEMENT_CREATE_INDEX: &'static str = "
CREATE INDEX timestamp_idx ON messages (timestamp ASC);
";
pub const SQL_STATEMENT_INSERT_SCHEMA: &'static str = "
INSERT INTO schema (schema_version, ros_distro) VALUES (4, 'rolling');
";
pub const SQL_STATEMENT_INSERT_MSG: &'static str =
"INSERT INTO messages (topic_id, timestamp, data) VALUES (?1, ?2, ?3)";
pub const SQL_STATEMENT_INSERT_TOPIC: &'static str = concat!(
"INSERT INTO topics (name, type, serialization_format, offered_qos_profiles, ",
"type_description_hash) VALUES (?1, ?2, 'cdr', '",
"- history: 3\n",
" depth: 0\n",
" reliability: 1\n",
" durability: 2\n",
" deadline:\n",
" sec: 9223372036\n",
" nsec: 854775807\n",
" lifespan:\n",
" sec: 9223372036\n",
" nsec: 854775807\n",
" liveliness: 1\n",
" liveliness_lease_duration:\n",
" sec: 9223372036\n",
" nsec: 854775807\n",
" avoid_ros_namespace_conventions: false",
"', ?3)"
);
pub const SQL_STATEMENT_REMOVE_TOPIC_MESSAGES: &'static str =
"DELETE FROM messages WHERE topic_id IN (SELECT id FROM topics WHERE name = ?1)";
pub const SQL_STATEMENT_REMOVE_TOPIC: &'static str = "DELETE FROM topics WHERE name = ?1";
pub const SQL_STATEMENT_SELECT_TOPICS: &'static str =
"SELECT id, name, type, serialization_format, offered_qos_profiles FROM topics";
pub const SQL_STATEMENT_SELECT_TOPIC_IDS: &'static str = "SELECT id, name FROM topics";
pub const SQL_STATEMENT_SELECT_TIME_RANGE: &'static str =
"SELECT MIN(timestamp), MAX(timestamp) FROM messages";
pub fn sql_statement_message_filter(topics: Option<usize>, begin: bool, end: bool) -> String {
const BASE: &str = "SELECT m.id, m.timestamp, m.topic_id, m.data \
FROM messages m";
fn repeat_marks(n: usize) -> String {
std::iter::repeat("?")
.take(n)
.collect::<Vec<_>>()
.join(", ")
}
let sql: String = match (topics, begin, end) {
(None, false, false) => format!("{BASE} ORDER BY m.timestamp ASC, m.id ASC"),
(None, false, true) => {
format!("{BASE} WHERE m.timestamp < ? ORDER BY m.timestamp ASC, m.id ASC")
}
(None, true, false) => {
format!("{BASE} WHERE m.timestamp >= ? ORDER BY m.timestamp ASC, m.id ASC")
}
(None, true, true) => format!(
"{BASE} WHERE m.timestamp >= ? AND m.timestamp < ? \
ORDER BY m.timestamp ASC, m.id ASC"
),
(Some(topics), false, false) => {
let marks = repeat_marks(topics);
format!("{BASE} WHERE m.topic_id IN ({marks}) ORDER BY m.timestamp ASC, m.id ASC")
}
(Some(topics), false, true) => {
let marks = repeat_marks(topics);
format!(
"{BASE} WHERE m.topic_id IN ({marks}) AND m.timestamp < ? \
ORDER BY m.timestamp ASC, m.id ASC"
)
}
(Some(topics), true, false) => {
let marks = repeat_marks(topics);
format!(
"{BASE} WHERE m.topic_id IN ({marks}) AND m.timestamp >= ? \
ORDER BY m.timestamp ASC, m.id ASC"
)
}
(Some(topics), true, true) => {
let marks = repeat_marks(topics);
format!(
"{BASE} WHERE m.topic_id IN ({marks}) AND m.timestamp >= ? AND m.timestamp < ? \
ORDER BY m.timestamp ASC, m.id ASC",
)
}
};
sql
}
pub fn sql_params_message_filter(
filter: &crate::message::filter::MessageFilter<'_>,
topic_mapping: &std::collections::HashMap<String, i64>,
) -> Result<Vec<turso::Value>, crate::Error> {
let mut params = Vec::new();
if let Some(topics) = &filter.topics {
for name in topics {
let id = topic_mapping
.get::<str>(name)
.ok_or(crate::Error::TopicNotFound(name.to_string()))?;
params.push(turso::Value::Integer(*id));
}
}
if let Some(begin) = filter.begin {
params.push(turso::Value::Integer(begin));
}
if let Some(end) = filter.end {
params.push(turso::Value::Integer(end));
}
Ok(params)
}