opendal_service_compfs/
core.rs1use std::future::Future;
19use std::path::Path;
20use std::path::PathBuf;
21
22use compio::buf::{IoBuf, IoVectoredBuf};
23use compio::dispatcher::Dispatcher;
24
25use opendal_core::raw::*;
26use opendal_core::*;
27
28#[derive(Debug, Clone)]
30pub struct CompfsBuffer(pub Vec<compio::bytes::Bytes>);
31
32impl IoBuf for CompfsBuffer {
33 fn as_init(&self) -> &[u8] {
34 self.0.first().map_or(&[], |b| b.as_ref())
35 }
36}
37
38impl From<CompfsBuffer> for opendal_core::Buffer {
39 fn from(buf: CompfsBuffer) -> Self {
40 buf.0.into()
41 }
42}
43
44impl From<opendal_core::Buffer> for CompfsBuffer {
45 fn from(mut buf: opendal_core::Buffer) -> Self {
46 Self(buf.by_ref().collect())
47 }
48}
49
50impl IoVectoredBuf for CompfsBuffer {
51 fn iter_slice(&self) -> impl Iterator<Item = &[u8]> {
52 self.0.iter().map(|b| b.as_ref())
53 }
54}
55
56#[derive(Debug)]
57pub(super) struct CompfsCore {
58 pub info: ServiceInfo,
59 pub capability: Capability,
60
61 pub root: PathBuf,
62 pub dispatcher: Dispatcher,
63 pub buf_pool: oio::PooledBuf,
64}
65
66impl CompfsCore {
67 pub fn prepare_path(&self, path: &str) -> Result<PathBuf> {
72 use std::path::Component;
73 let trimmed = path.trim_end_matches('/');
74 if Path::new(trimmed).components().any(|c| {
75 matches!(
76 c,
77 Component::ParentDir | Component::RootDir | Component::Prefix(_)
78 )
79 }) {
80 return Err(
81 Error::new(ErrorKind::NotFound, "path escapes the configured root")
82 .with_context("path", path),
83 );
84 }
85 Ok(self.root.join(trimmed))
86 }
87
88 pub async fn exec<Fn, Fut, R>(&self, f: Fn) -> opendal_core::Result<R>
89 where
90 Fn: FnOnce() -> Fut + Send + 'static,
91 Fut: Future<Output = std::io::Result<R>> + 'static,
92 R: Send + 'static,
93 {
94 self.dispatcher
95 .dispatch(f)
96 .map_err(|_| Error::new(ErrorKind::Unexpected, "compio spawn io task failed"))?
97 .await
98 .map_err(|_| Error::new(ErrorKind::Unexpected, "compio task cancelled"))?
99 .map_err(new_std_io_error)
100 }
101
102 pub async fn exec_blocking<Fn, R>(&self, f: Fn) -> Result<R>
103 where
104 Fn: FnOnce() -> R + Send + 'static,
105 R: Send + 'static,
106 {
107 self.dispatcher
108 .dispatch_blocking(f)
109 .map_err(|_| Error::new(ErrorKind::Unexpected, "compio spawn blocking task failed"))?
110 .await
111 .map_err(|_| Error::new(ErrorKind::Unexpected, "compio task cancelled"))
112 }
113}
114
115#[cfg(test)]
116mod tests {
117 use bytes::Bytes;
118 use rand::{RngExt, rng};
119
120 use super::*;
121
122 fn setup_buffer() -> (CompfsBuffer, usize, Bytes) {
123 let mut rng = rng();
124
125 let bs = (0..100)
126 .map(|_| {
127 let len = rng.random_range(1..100);
128 let mut buf = vec![0; len];
129 rng.fill(&mut buf[..]);
130 Bytes::from(buf)
131 })
132 .collect::<Vec<_>>();
133
134 let total_size = bs.iter().map(|b| b.len()).sum::<usize>();
135 let total_content = bs.iter().flatten().copied().collect::<Bytes>();
136 let buf = Buffer::from(bs);
137
138 (CompfsBuffer::from(buf), total_size, total_content)
139 }
140
141 #[test]
142 fn test_io_buf() {
143 let (buf, _len, _bytes) = setup_buffer();
144 let slice = IoBuf::as_init(&buf);
145
146 assert_eq!(slice, buf.0.first().unwrap().as_ref())
147 }
148
149 #[test]
150 fn test_io_vectored_buf() {
151 let (buf, len, bytes) = setup_buffer();
152 let collected = buf.iter_slice().flatten().copied().collect::<Bytes>();
153
154 assert_eq!(buf.total_len(), len);
155 assert_eq!(collected, bytes);
156 }
157
158 fn new_test_core() -> CompfsCore {
159 CompfsCore {
160 info: ServiceInfo::new("compfs", "", ""),
161 capability: Capability::default(),
162 root: PathBuf::from("/data/root"),
163 dispatcher: Dispatcher::new().unwrap(),
164 buf_pool: oio::PooledBuf::new(16),
165 }
166 }
167
168 #[test]
169 fn test_prepare_path_rejects_parent_dir() {
170 let core = new_test_core();
171 for key in ["../etc/passwd", "../../etc/passwd", "a/../../b", "a/.."] {
172 let err = core.prepare_path(key).unwrap_err();
173 assert_eq!(
174 err.kind(),
175 ErrorKind::NotFound,
176 "key should be rejected: {key}"
177 );
178 }
179 }
180
181 #[test]
182 fn test_prepare_path_rejects_absolute_path() {
183 let core = new_test_core();
184 for key in ["/etc/passwd", "//etc/passwd"] {
186 let err = core.prepare_path(key).unwrap_err();
187 assert_eq!(
188 err.kind(),
189 ErrorKind::NotFound,
190 "key should be rejected: {key}"
191 );
192 }
193 }
194
195 #[test]
196 fn test_prepare_path_allows_normal_keys() {
197 let core = new_test_core();
198 assert_eq!(
200 core.prepare_path("a/b.txt").unwrap(),
201 PathBuf::from("/data/root/a/b.txt")
202 );
203 assert_eq!(
204 core.prepare_path("a/b/").unwrap(),
205 PathBuf::from("/data/root/a/b")
206 );
207 assert_eq!(
208 core.prepare_path("./a/b").unwrap(),
209 PathBuf::from("/data/root/a/b")
210 );
211 assert_eq!(
213 core.prepare_path("a..b").unwrap(),
214 PathBuf::from("/data/root/a..b")
215 );
216 }
217}