streamling-state 0.1.0

State management and persistence for Streamling.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

use crate::common::create_logger;
use serde_derive::{Deserialize, Serialize};
use serial_test::serial;
use std::sync::Arc;
use streamling_state::sqlite::SqliteStateOperatorBackendFactory;
use streamling_state::{
    StateBackendError, StateKey, StateOperatorBackend, StateOperatorBackendFactory,
};

pub mod common;

#[derive(Serialize, Deserialize, Debug, Clone, Eq, PartialEq)]
struct TestValue {
    field1: String,
    field2: i32,
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn test_state_backend_sqlite() -> Result<(), StateBackendError> {
    let _ = create_logger().try_init();

    let db_file = "./test_state.db";

    let factory = SqliteStateOperatorBackendFactory::new(db_file.to_string(), None, None).await?;
    let backend: Arc<dyn StateOperatorBackend<TestValue>> = factory.create("test_ns");

    let key = StateKey::from("test_key");
    let value = TestValue {
        field1: "test_value".to_string(),
        field2: 42,
    };

    backend.put(key.clone(), value.clone()).await?;
    assert_eq!(backend.get(key.clone()).await?, Some(value.clone()));

    backend.remove(key.clone()).await?;
    assert_eq!(backend.get(key).await?, None);

    std::fs::remove_file(db_file).unwrap_or_else(|_| {
        eprintln!("Failed to delete the SQLite database file.");
    });

    Ok(())
}