datastore.rs 2.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. use std::marker::PhantomData;
  2. use log::debug;
  3. use sled::Batch;
  4. use crate::{
  5. util::serial::{deserialize, serialize, Decodable, Encodable},
  6. Result,
  7. };
  8. use super::{Log, NodeId};
  9. const SLED_LOGS_TREE: &[u8] = b"_logs";
  10. const SLED_COMMITS_TREE: &[u8] = b"_commits";
  11. const SLED_COMMITS_LENGTH_TREE: &[u8] = b"_commit_length";
  12. const SLED_VOTED_FOR_TREE: &[u8] = b"_voted_for";
  13. const SLED_CURRENT_TERM_TREE: &[u8] = b"_current_term";
  14. pub struct DataStore<T> {
  15. _db: sled::Db,
  16. pub logs: DataTree<Log>,
  17. pub commits: DataTree<T>,
  18. pub commits_length: DataTree<u64>,
  19. pub voted_for: DataTree<Option<NodeId>>,
  20. pub current_term: DataTree<u64>,
  21. }
  22. impl<T: Encodable + Decodable> DataStore<T> {
  23. pub fn new(db_path: &str) -> Result<Self> {
  24. let _db = sled::open(db_path)?;
  25. let logs = DataTree::new(&_db, SLED_LOGS_TREE)?;
  26. let commits = DataTree::new(&_db, SLED_COMMITS_TREE)?;
  27. let commits_length = DataTree::new(&_db, SLED_COMMITS_LENGTH_TREE)?;
  28. let voted_for = DataTree::new(&_db, SLED_VOTED_FOR_TREE)?;
  29. let current_term = DataTree::new(&_db, SLED_CURRENT_TERM_TREE)?;
  30. Ok(Self { _db, logs, commits, commits_length, voted_for, current_term })
  31. }
  32. pub async fn cancel(&self) -> Result<()> {
  33. debug!(target: "raft", "DataStore flush");
  34. self._db.flush_async().await?;
  35. Ok(())
  36. }
  37. }
  38. pub struct DataTree<T> {
  39. tree: sled::Tree,
  40. phantom: PhantomData<T>,
  41. }
  42. impl<T: Decodable + Encodable> DataTree<T> {
  43. pub fn new(db: &sled::Db, tree_name: &[u8]) -> Result<Self> {
  44. let tree = db.open_tree(tree_name)?;
  45. Ok(Self { tree, phantom: PhantomData })
  46. }
  47. pub fn insert(&self, data: &T) -> Result<()> {
  48. let serialized = serialize(data);
  49. let datahash = blake3::hash(&serialized);
  50. self.tree.insert(datahash.as_bytes(), serialized)?;
  51. Ok(())
  52. }
  53. pub fn wipe_insert_all(&self, data: &Vec<T>) -> Result<()> {
  54. self.tree.clear()?;
  55. let mut batch = Batch::default();
  56. for i in data {
  57. let serialized = serialize(i);
  58. let hash = blake3::hash(&serialized);
  59. batch.insert(hash.as_bytes(), serialized);
  60. }
  61. self.tree.apply_batch(batch)?;
  62. Ok(())
  63. }
  64. pub fn get_all(&self) -> Result<Vec<T>> {
  65. let mut ret: Vec<T> = Vec::new();
  66. for i in self.tree.iter() {
  67. let da = deserialize(&i?.1)?;
  68. ret.push(da)
  69. }
  70. Ok(ret)
  71. }
  72. pub fn get_last(&self) -> Result<Option<T>> {
  73. if let Some(found) = self.tree.last()? {
  74. let da = deserialize(&found.1)?;
  75. return Ok(Some(da))
  76. }
  77. Ok(None)
  78. }
  79. }