1use ::kvdb::{DBTransaction, KeyValueDB};
20
21use crate::{error, Change, ColumnId, Database, Transaction};
22
23struct DbAdapter<D: KeyValueDB + 'static>(D);
24
25fn handle_err<T>(result: std::io::Result<T>) -> T {
26 match result {
27 Ok(r) => r,
28 Err(e) => {
29 panic!("Critical database error: {:?}", e);
30 },
31 }
32}
33
34fn read_counter(
36 db: &dyn KeyValueDB,
37 col: ColumnId,
38 key: &[u8],
39) -> error::Result<(Vec<u8>, Option<u32>)> {
40 let mut counter_key = key.to_vec();
41 counter_key.push(0);
42 Ok(match db.get(col, &counter_key).map_err(|e| error::DatabaseError(Box::new(e)))? {
43 Some(data) => {
44 let mut counter_data = [0; 4];
45 if data.len() != 4 {
46 return Err(error::DatabaseError(Box::new(std::io::Error::new(
47 std::io::ErrorKind::Other,
48 format!("Unexpected counter len {}", data.len()),
49 ))))
50 }
51 counter_data.copy_from_slice(&data);
52 let counter = u32::from_le_bytes(counter_data);
53 (counter_key, Some(counter))
54 },
55 None => (counter_key, None),
56 })
57}
58
59fn commit_impl<H: Clone + AsRef<[u8]>>(
61 db: &dyn KeyValueDB,
62 transaction: Transaction<H>,
63) -> error::Result<()> {
64 let mut tx = DBTransaction::new();
65 for change in transaction.0.into_iter() {
66 match change {
67 Change::Set(col, key, value) => tx.put_vec(col, &key, value),
68 Change::Remove(col, key) => tx.delete(col, &key),
69 Change::Store(col, key, value) => match read_counter(db, col, key.as_ref())? {
70 (counter_key, Some(mut counter)) => {
71 counter += 1;
72 tx.put(col, &counter_key, &counter.to_le_bytes());
73 },
74 (counter_key, None) => {
75 let d = 1u32.to_le_bytes();
76 tx.put(col, &counter_key, &d);
77 tx.put_vec(col, key.as_ref(), value);
78 },
79 },
80 Change::Reference(col, key) => {
81 if let (counter_key, Some(mut counter)) = read_counter(db, col, key.as_ref())? {
82 counter += 1;
83 tx.put(col, &counter_key, &counter.to_le_bytes());
84 }
85 },
86 Change::Release(col, key) => {
87 if let (counter_key, Some(mut counter)) = read_counter(db, col, key.as_ref())? {
88 counter -= 1;
89 if counter == 0 {
90 tx.delete(col, &counter_key);
91 tx.delete(col, key.as_ref());
92 } else {
93 tx.put(col, &counter_key, &counter.to_le_bytes());
94 }
95 }
96 },
97 }
98 }
99 db.write(tx).map_err(|e| error::DatabaseError(Box::new(e)))
100}
101
102pub fn as_database<D, H>(db: D) -> std::sync::Arc<dyn Database<H>>
104where
105 D: KeyValueDB + 'static,
106 H: Clone + AsRef<[u8]>,
107{
108 std::sync::Arc::new(DbAdapter(db))
109}
110
111impl<D: KeyValueDB, H: Clone + AsRef<[u8]>> Database<H> for DbAdapter<D> {
112 fn commit(&self, transaction: Transaction<H>) -> error::Result<()> {
113 commit_impl(&self.0, transaction)
114 }
115
116 fn get(&self, col: ColumnId, key: &[u8]) -> Option<Vec<u8>> {
117 handle_err(self.0.get(col, key))
118 }
119
120 fn contains(&self, col: ColumnId, key: &[u8]) -> bool {
121 handle_err(self.0.has_key(col, key))
122 }
123}
124
125#[cfg(feature = "rocksdb")]
127pub struct RocksDbAdapter(kvdb_rocksdb::Database);
128
129#[cfg(feature = "rocksdb")]
130impl<H: Clone + AsRef<[u8]>> Database<H> for RocksDbAdapter {
131 fn commit(&self, transaction: Transaction<H>) -> error::Result<()> {
132 commit_impl(&self.0, transaction)
133 }
134
135 fn get(&self, col: ColumnId, key: &[u8]) -> Option<Vec<u8>> {
136 handle_err(self.0.get(col, key))
137 }
138
139 fn contains(&self, col: ColumnId, key: &[u8]) -> bool {
140 handle_err(self.0.has_key(col, key))
141 }
142
143 fn optimize_db_col(&self, col: ColumnId) -> error::Result<()> {
144 self.0.force_compact(col).map_err(|e| error::DatabaseError(Box::new(e)))
145 }
146}
147
148#[cfg(feature = "rocksdb")]
150pub fn as_rocksdb_database<H>(db: kvdb_rocksdb::Database) -> std::sync::Arc<dyn Database<H>>
151where
152 H: Clone + AsRef<[u8]>,
153{
154 std::sync::Arc::new(RocksDbAdapter(db))
155}