ngdb 2.2.2

A high-performance, thread-safe RocksDB wrapper
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

//! Thread-safe transaction example demonstrating concurrent access patterns

use ngdb::{DatabaseConfig, Result, Storable, ngdb};
use std::sync::Arc;
use std::thread;

#[ngdb("counters")]
struct Counter {
    id: String,
    value: i64,
}

impl Storable for Counter {
    type Key = String;

    fn key(&self) -> Self::Key {
        self.id.clone()
    }
}

fn main() -> Result<()> {
    let db = DatabaseConfig::new("./data/thread_safe")
        .create_if_missing(true)
        .add_column_family("counters")
        .open()?;

    let counters = Counter::collection(&db)?;

    // Shared transaction across threads
    {
        let txn = Arc::new(db.transaction()?);
        let mut handles = vec![];

        for i in 0..3 {
            let txn_clone = Arc::clone(&txn);
            let handle = thread::spawn(move || {
                let collection = txn_clone
                    .collection::<Counter>("counters")
                    .expect("Failed to get collection");

                collection
                    .put(&Counter {
                        id: format!("thread_{}", i),
                        value: i as i64 * 100,
                    })
                    .expect("Failed to put counter");
            });
            handles.push(handle);
        }

        for handle in handles {
            handle.join().unwrap();
        }

        match Arc::try_unwrap(txn) {
            Ok(transaction) => transaction.commit()?,
            Err(_) => panic!("Failed to unwrap Arc"),
        }
    }
    println!("Committed shared transaction across 3 threads");

    // Transaction read isolation
    {
        let txn = db.transaction()?;
        let txn_counters = txn.collection::<Counter>("counters")?;

        txn_counters.put(&Counter {
            id: "isolated".to_string(),
            value: 999,
        })?;

        let read_back = txn_counters.get(&"isolated".to_string())?;
        println!(
            "Within transaction, value: {:?}",
            read_back.as_ref().map(|c| c.value)
        );

        txn.rollback()?;

        let after_rollback = counters.get(&"isolated".to_string())?;
        println!("After rollback, value: {:?}", after_rollback);
    }

    // Concurrent independent transactions
    {
        let mut handles = vec![];

        for i in 0..3 {
            let db_clone = db.clone();
            let handle = thread::spawn(move || {
                let txn = db_clone
                    .transaction()
                    .expect("Failed to create transaction");
                let collection = txn
                    .collection::<Counter>("counters")
                    .expect("Failed to get collection");

                collection
                    .put(&Counter {
                        id: format!("concurrent_{}", i),
                        value: i as i64 * 1000,
                    })
                    .expect("Failed to put");

                txn.commit().expect("Failed to commit");
            });
            handles.push(handle);
        }

        for handle in handles {
            handle.join().unwrap();
        }
    }
    println!("Completed 3 concurrent independent transactions");

    // Iteration status tracking
    let status = counters.iter()?.for_each(|counter| counter.value < 1000)?;

    match status {
        ngdb::IterationStatus::Completed => println!("Iteration completed all records"),
        ngdb::IterationStatus::StoppedEarly => println!("Iteration stopped early"),
    }

    let count = counters.iter()?.count()?;
    println!("Total counters: {}", count);

    db.shutdown()?;
    Ok(())
}