opendal/services/rocksdb/
backend.rs1use std::sync::Arc;
19
20use rocksdb::DB;
21
22use super::config::RocksdbConfig;
23use super::core::*;
24use super::deleter::RocksdbDeleter;
25use super::lister::RocksdbLister;
26use super::writer::RocksdbWriter;
27use crate::raw::*;
28use crate::*;
29
30#[doc = include_str!("docs.md")]
32#[derive(Clone, Default)]
33pub struct RocksdbBuilder {
34 pub(super) config: RocksdbConfig,
35}
36
37impl RocksdbBuilder {
38 pub fn datadir(mut self, path: &str) -> Self {
40 self.config.datadir = Some(path.into());
41 self
42 }
43
44 pub fn root(mut self, root: &str) -> Self {
48 self.config.root = if root.is_empty() {
49 None
50 } else {
51 Some(root.to_string())
52 };
53
54 self
55 }
56}
57
58impl Builder for RocksdbBuilder {
59 type Config = RocksdbConfig;
60
61 fn build(self) -> Result<impl Access> {
62 let path = self.config.datadir.ok_or_else(|| {
63 Error::new(ErrorKind::ConfigInvalid, "datadir is required but not set")
64 .with_context("service", Scheme::Rocksdb)
65 })?;
66 let db = DB::open_default(&path).map_err(|e| {
67 Error::new(ErrorKind::ConfigInvalid, "open default transaction db")
68 .with_context("service", Scheme::Rocksdb)
69 .with_context("datadir", path)
70 .set_source(e)
71 })?;
72
73 let root = normalize_root(&self.config.root.unwrap_or_default());
74
75 Ok(RocksdbBackend::new(RocksdbCore { db: Arc::new(db) }).with_normalized_root(root))
76 }
77}
78
79#[derive(Clone, Debug)]
81pub struct RocksdbBackend {
82 core: Arc<RocksdbCore>,
83 root: String,
84 info: Arc<AccessorInfo>,
85}
86
87impl RocksdbBackend {
88 pub fn new(core: RocksdbCore) -> Self {
89 let info = AccessorInfo::default();
90 info.set_scheme(Scheme::Rocksdb.into_static())
91 .set_name(&core.db.path().to_string_lossy())
92 .set_root("/")
93 .set_native_capability(Capability {
94 read: true,
95 stat: true,
96 write: true,
97 write_can_empty: true,
98 delete: true,
99 list: true,
100 list_with_recursive: true,
101 shared: false,
102 ..Default::default()
103 });
104
105 Self {
106 core: Arc::new(core),
107 root: "/".to_string(),
108 info: Arc::new(info),
109 }
110 }
111
112 fn with_normalized_root(mut self, root: String) -> Self {
113 self.info.set_root(&root);
114 self.root = root;
115 self
116 }
117}
118
119impl Access for RocksdbBackend {
120 type Reader = Buffer;
121 type Writer = RocksdbWriter;
122 type Lister = oio::HierarchyLister<RocksdbLister>;
123 type Deleter = oio::OneShotDeleter<RocksdbDeleter>;
124
125 fn info(&self) -> Arc<AccessorInfo> {
126 self.info.clone()
127 }
128
129 async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
130 let p = build_abs_path(&self.root, path);
131
132 if p == build_abs_path(&self.root, "") {
133 Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
134 } else {
135 let bs = self.core.get(&p)?;
136 match bs {
137 Some(bs) => Ok(RpStat::new(
138 Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
139 )),
140 None => Err(Error::new(ErrorKind::NotFound, "kv not found in rocksdb")),
141 }
142 }
143 }
144
145 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
146 let p = build_abs_path(&self.root, path);
147 let bs = match self.core.get(&p)? {
148 Some(bs) => bs,
149 None => {
150 return Err(Error::new(ErrorKind::NotFound, "kv not found in rocksdb"));
151 }
152 };
153 Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
154 }
155
156 async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
157 let p = build_abs_path(&self.root, path);
158 let writer = RocksdbWriter::new(self.core.clone(), p);
159 Ok((RpWrite::new(), writer))
160 }
161
162 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
163 let deleter = RocksdbDeleter::new(self.core.clone(), self.root.clone());
164 Ok((RpDelete::default(), oio::OneShotDeleter::new(deleter)))
165 }
166
167 async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
168 let p = build_abs_path(&self.root, path);
169 let lister = RocksdbLister::new(self.core.clone(), self.root.clone(), p)?;
170 Ok((
171 RpList::default(),
172 oio::HierarchyLister::new(lister, path, args.recursive()),
173 ))
174 }
175}