Generalized storage

.github/workflows/ci.yml

@@ -68,7 +68,7 @@ jobs:
         target: [
           aarch64-unknown-linux-gnu,
           powerpc64le-unknown-linux-gnu,
-          s390x-unknown-linux-gnu,
+          # s390x-unknown-linux-gnu,
         ]
     steps:
       - uses: actions/checkout@v2
@@ -118,7 +118,7 @@ jobs:
           test -e ~/.cargo/bin/cargo-tarpaulin || cargo install cargo-tarpaulin --version ${{ steps.tarpaulin-version.outputs.version }}
 
       - name: Run cargo-tarpaulin
-        run: cargo tarpaulin --all-features --timeout 600 --out Xml -- --test-threads 1
+        run: cargo tarpaulin --all-features --timeout 600 --out xml -- --test-threads 1
 
       - name: Upload coverage to codecov
         uses: codecov/codecov-action@v3
@@ -193,8 +193,20 @@ jobs:
           command: publish
           args: --dry-run -p directip
 
-      - name: Make sure we can publish direciip-client
+      - name: Make sure we can publish directip-client
         uses: actions-rs/cargo@v1
         with:
           command: publish
           args: --dry-run -p directip-client
+
+      - name: Make sure we can publish directip-dump
+        uses: actions-rs/cargo@v1
+        with:
+          command: publish
+          args: --dry-run -p directip-dump
+
+      - name: Make sure we can publish directip-storage
+        uses: actions-rs/cargo@v1
+        with:
+          command: publish
+          args: --dry-run -p directip-storage

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "directip"
-version = "0.2.5"
+version = "0.2.6"
 authors = ["Guilherme Castelão <[email protected]>", "Luiz Irber <[email protected]>"]
 edition = "2021"
 description = "Iridium SBD Direct-IP communication protocol"
@@ -12,21 +12,34 @@ rust-version = "1.60.0"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [workspace]
-members = ["directip-client", "directip-dump"]
+members = ["directip-client", "directip-dump", "crates/storage"]
 default-members = [".", "directip-client"]
 
+[workspace.package]
+authors = ["Guilherme Castelão <[email protected]>", "Luiz Irber <[email protected]>"]
+edition = "2021"
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/castelao/DirectIP"
+
 [features]
 serde = ["dep:serde", "dep:serde_bytes", "chrono/serde"]
 
 [dependencies]
 byteorder = "1"
-chrono = "0.4.23"
+chrono = "0.4.31"
 log = "0.4"
 thiserror = "1.0"
 derive_builder = "0.11.2"
 serde = { version = "1.0.147", optional = true, default-features = false, features=["serde_derive"]}
 serde_bytes = { version = "0.11.8", optional = true }
 
+[workspace.dependencies]
+chrono = "0.4.31"
+directip = { version = "0.2.6", path = "." }
+sqlx = { version = "=0.7.0", features = ["sqlite", "runtime-tokio-native-tls"] }
+tempfile = "3.8.0"
+tokio = { version = "1.32.0", features = ["macros"] }
+
 [lib]
 name = "directip"
 

README.md

@@ -16,6 +16,11 @@ the core support for applications such as the one listed below.
   components, such as the IMEI or message-id, from a Direct-IP binary
   transmission. This is an equivalent to ncdump but for Direct-IP messages.
 
+- [directip-storage](https://crates.io/crates/directip-storage): A library
+  that provides multiple backends to store Direct-IP messages. It can be
+  used, for instance, to archive MT messages transmitted or MO messages
+  received.
+
 ## Minimum supported Rust version
 
 Currently the minimum supported Rust version is 1.60.0

crates/storage/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "directip-storage"
+version = "0.0.1"
+authors = ["Guilherme Castelão <[email protected]>", "Luiz Irber <[email protected]>"]
+license = "MIT OR Apache-2.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+chrono.workspace = true
+directip.workspace = true
+sqlx = { workspace = true, optional = true }
+tempfile = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+
+[features]
+sqlite = ["dep:sqlx"]

crates/storage/src/filesystem.rs

@@ -0,0 +1,95 @@
+//   filesystem:///var/iridium-storage/
+
+use std::fs::File;
+use std::io::{BufWriter, Read, Write};
+use std::path::PathBuf;
+
+use chrono::Utc;
+
+use directip::Message;
+
+pub struct FileSystemStorage {
+    root: PathBuf,
+    // current_id: usize,
+}
+
+impl super::Storage for FileSystemStorage {}
+
+impl FileSystemStorage {
+    pub(super) fn connect() -> Result<Self, Box<dyn std::error::Error>> {
+        let tmp_dir = tempfile::TempDir::new().unwrap();
+        Ok(FileSystemStorage {
+            root: tmp_dir.path().into(),
+        })
+    }
+
+    pub(super) async fn save(&self, msg: Message) {
+        let mut path = self.root.clone();
+        path.push("data");
+        let mut filename = String::new();
+        // Add IMEI?
+        filename.push_str(&Utc::now().format("%Y%m%d%H%M%S%s").to_string());
+        //filename.push_str(&format!("_{}", &self.current_id));
+        filename.push_str(".isbd");
+        path.push(filename);
+
+        let mut file = BufWriter::new(File::create(path).unwrap());
+        file.write(&msg.to_vec()).unwrap();
+    }
+    /*
+    pub fn current_id(&self) -> usize {
+        self.current_id
+    }
+    */
+
+    // Probably distinguish between creating a new from scratch versus
+    // connecting to an existing one. Connecting requires less checks,
+    // faster, and a different level of errors.
+    /*
+    pub fn initiatedb(root: PathBuf) -> FileSystemStorage {
+        let mut next_session_id = root.clone();
+        next_session_id.push("next_session.txt");
+
+        let current_id: usize = if next_session_id.exists() {
+            let mut file = File::open(next_session_id).unwrap();
+            let mut buffer = String::new();
+            file.read_to_string(&mut buffer).unwrap();
+            if buffer.ends_with('\n') {
+                buffer.pop();
+            }
+            buffer.parse().unwrap()
+        } else {
+            let mut file = std::fs::File::create(next_session_id).unwrap();
+            file.write(b"1").unwrap();
+            1
+            // let mut file = BufWriter::new(File::create(next_session_id).unwrap());
+            //file.write("0").unwrap();
+        };
+        FileSystemStorage { root, current_id }
+    }
+    */
+}
+
+#[cfg(test)]
+mod test_filesystem {
+    use super::*;
+    use directip;
+
+    fn sample() -> directip::Message {
+        let msg = directip::mt::MTMessage::from_reader(
+            [
+                0x01, 0x00, 0x1c, 0x44, 0x00, 0x19, 0x00, 0x00, 0x27, 0x0f, 0x00, 0x01, 0x02, 0x03,
+                0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0xff, 0xff, 0xff,
+                0xff, 0xff, 0xf5,
+            ]
+            .as_slice(),
+        );
+        directip::Message::MT(msg.unwrap())
+    }
+
+    //#[tokio::test]
+    async fn filesystem() {
+        let storage = FileSystemStorage::connect().unwrap();
+        storage.save(sample()).await;
+    }
+}

crates/storage/src/inmemory.rs

@@ -0,0 +1,51 @@
+use std::sync::RwLock;
+
+use crate::Message;
+
+pub struct VolatileStorage {
+    // current_id: usize,
+    data: RwLock<Vec<Message>>,
+}
+
+impl super::Storage for VolatileStorage {}
+
+impl VolatileStorage {
+    pub(super) fn connect() -> Result<VolatileStorage, Box<dyn std::error::Error>> {
+        Ok(VolatileStorage {
+            data: RwLock::new(vec![]),
+        })
+    }
+
+    pub(super) async fn save(&self, msg: Message) {
+        self.data
+            .write()
+            .expect("Failed to acquire write lock.")
+            .push(msg);
+    }
+}
+
+#[cfg(test)]
+mod test_volatile {
+    use super::VolatileStorage;
+    use directip;
+
+    fn sample() -> directip::Message {
+        let msg = directip::mt::MTMessage::from_reader(
+            [
+                0x01, 0x00, 0x1c, 0x44, 0x00, 0x19, 0x00, 0x00, 0x27, 0x0f, 0x00, 0x01, 0x02, 0x03,
+                0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0xff, 0xff, 0xff,
+                0xff, 0xff, 0xf5,
+            ]
+            .as_slice(),
+        );
+        directip::Message::MT(msg.unwrap())
+    }
+
+    #[tokio::test]
+    async fn volatile() {
+        let storage = VolatileStorage::connect().unwrap();
+        storage.save(sample()).await;
+
+        assert_eq!(storage.data.read().unwrap().len(), 1);
+    }
+}

crates/storage/src/lib.rs

@@ -0,0 +1,93 @@
+mod filesystem;
+mod inmemory;
+
+use directip::Message;
+use filesystem::FileSystemStorage;
+use inmemory::VolatileStorage;
+
+// Feature sqlite
+#[cfg(feature = "sqlite")]
+mod sqlite;
+
+trait Storage {
+    //fn connect(cfg: String) -> Self;
+    // fn save(&self, msg: Message);
+}
+
+enum Database {
+    M(VolatileStorage),
+    F(FileSystemStorage),
+    #[cfg(feature = "sqlite")]
+    L(sqlite::SQLiteStorage),
+}
+
+impl Database {
+    pub async fn open(cfg: &str) -> Result<Self, Box<dyn std::error::Error>> {
+        if (&cfg.len() >= &(11 as usize)) && (cfg[..11] == "volatile://".to_string()) {
+            Ok(Database::M(VolatileStorage::connect()?))
+        } else if (&cfg.len() >= &(13 as usize)) && (cfg[..13] == "filesystem://".to_string()) {
+            Ok(Database::F(FileSystemStorage::connect()?))
+        } else if cfg[..9] == "sqlite://".to_string() {
+            #[cfg(feature = "sqlite")]
+            {
+                let db = crate::sqlite::SQLiteStorage::connect().await;
+                Ok(Database::L(db))
+            }
+            #[cfg(not(feature = "sqlite"))]
+            unimplemented!("Missing sqlite feature")
+        } else {
+            unimplemented!("Unknown storage")
+        }
+        /*
+        Database::F(FileSystemStorage::initiatedb(std::path::PathBuf::from(
+            "./",
+        )))
+        */
+    }
+
+    pub async fn save(&self, msg: Message) {
+        match self {
+            Database::M(s) => s.save(msg).await,
+            Database::F(s) => s.save(msg).await,
+            #[cfg(feature = "sqlite")]
+            Database::L(s) => s.save(msg).await,
+            _ => todo!(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Database;
+
+    fn sample() -> directip::Message {
+        let msg = directip::mt::MTMessage::from_reader(
+            [
+                0x01, 0x00, 0x1c, 0x44, 0x00, 0x19, 0x00, 0x00, 0x27, 0x0f, 0x00, 0x01, 0x02, 0x03,
+                0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0xff, 0xff, 0xff,
+                0xff, 0xff, 0xf5,
+            ]
+            .as_slice(),
+        );
+        directip::Message::MT(msg.unwrap())
+    }
+
+    #[tokio::test]
+    async fn volatile() {
+        let db = Database::open("volatile://").await.unwrap();
+        db.save(sample());
+    }
+
+    //#[tokio::test]
+    async fn filesystem() {
+        let db = Database::open("filesystem://").await.unwrap();
+        db.save(sample());
+    }
+
+    #[cfg(feature = "sqlite")]
+    #[tokio::test]
+    async fn open_sqlite() {
+        let db = Database::open("sqlite://").await.unwrap();
+        db.save(sample());
+    }
+}