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) RETURNING id"
);
pub const SQL_STATEMENT_REMOVE_TOPIC_MESSAGES: &'static str =
"DELETE FROM messages WHERE topic_id = ?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 const SQL_STATEMENT_SELECT_TOPIC_TIMESTAMPS: &'static str = "
SELECT t.name, m.timestamp
FROM topics t
JOIN messages m ON m.topic_id = t.id
ORDER BY m.timestamp ASC, m.id ASC
";
pub const OFFERED_QOS_PROFILES: &'static str = concat!(
"- 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",
);
pub const SQL_STATEMENT_POP_MESSAGE: &'static str = "
DELETE FROM messages
WHERE id = (
SELECT id FROM messages
ORDER BY timestamp DESC, id DESC
LIMIT 1
)
RETURNING id, timestamp, topic_id, data
";
pub const SQL_STATEMENT_POP_FRONT_MESSAGE: &'static str = "
DELETE FROM messages
WHERE id = (
SELECT id FROM messages
ORDER BY timestamp ASC, id ASC
LIMIT 1
)
RETURNING id, timestamp, topic_id, data
";
pub fn sql_statement_message_filter(
topics: Option<usize>,
begin: bool,
end: bool,
delete: bool,
) -> String {
fn repeat_marks(n: usize) -> String {
std::iter::repeat("?")
.take(n)
.collect::<Vec<_>>()
.join(", ")
}
let mut conditions = Vec::new();
if let Some(topics) = topics {
conditions.push(format!("m.topic_id IN ({})", repeat_marks(topics)));
}
if begin {
conditions.push("m.timestamp >= ?".to_string());
}
if end {
conditions.push("m.timestamp < ?".to_string());
}
let where_clause = if conditions.is_empty() {
String::new()
} else {
format!(" WHERE {}", conditions.join(" AND "))
};
if delete {
format!(
"DELETE FROM messages WHERE id IN (\
SELECT m.id FROM messages m{where_clause}\
) RETURNING id, timestamp, topic_id, data"
)
} else {
format!(
"SELECT m.id, m.timestamp, m.topic_id, m.data \
FROM messages m{where_clause} \
ORDER BY m.timestamp ASC, m.id ASC"
)
}
}
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)
}