librados 0.2.0

Idiomatic (async) rust bindings for librados
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

use std::{ffi::CString, pin::Pin};

use crate::{
    IoCtx, RadosCompletion, RadosError, Result,
    error::maybe_err,
    librados::{
        rados_aio_read_op_operate, rados_create_read_op, rados_read_op_operate, rados_read_op_t,
        rados_release_read_op,
    },
};

struct ReadOpPtr(rados_read_op_t);

impl ReadOpPtr {
    fn new() -> Result<Self> {
        let read_op = unsafe { rados_create_read_op() };

        (!read_op.is_null())
            .then_some(Self(read_op))
            .ok_or(RadosError::Unknown(-1))
    }

    fn get(&self) -> rados_read_op_t {
        self.0
    }
}

impl Drop for ReadOpPtr {
    fn drop(&mut self) {
        unsafe { rados_release_read_op(self.0) };
    }
}

pub(crate) struct ReadOpExecutor<'rados, 'ioctx, T>
where
    T: ReadOp,
{
    operation: T,
    inner: ReadOpPtr,
    ioctx: &'ioctx IoCtx<'rados>,
}

impl<'rados, 'ioctx, T> ReadOpExecutor<'rados, 'ioctx, T>
where
    T: ReadOp,
{
    pub fn new(ioctx: &'ioctx IoCtx<'rados>, operation: T) -> Result<Self> {
        Ok(Self {
            inner: ReadOpPtr::new()?,
            operation,
            ioctx,
        })
    }

    pub fn execute(self, object: &str) -> Result<T::Output> {
        let object = CString::new(object).expect("Object ID had interior NUL.");
        let mut output = T::OperationState::default();

        let pinned = unsafe { Pin::new_unchecked(&mut output) };

        self.operation
            .construct_in_place(self.inner.get(), pinned)?;

        let result = unsafe {
            rados_read_op_operate(self.inner.get(), self.ioctx.inner(), object.as_ptr(), 0)
        };

        maybe_err(result)?;

        T::complete(output)
    }

    pub async fn execute_async(self, object: &str) -> Result<T::Output>
    where
        T::OperationState: 'static + Unpin,
    {
        let object = CString::new(object).expect("Object ID had interior NUL");
        let state = T::OperationState::default();

        let completion = unsafe {
            RadosCompletion::new_with(false, (object, state), |completion, mut full_state| {
                let pinned = &mut full_state.1;
                let op_state = core::pin::Pin::new_unchecked(pinned);

                self.operation
                    .construct_in_place(self.inner.get(), op_state)?;

                maybe_err(rados_aio_read_op_operate(
                    self.inner.get(),
                    self.ioctx.inner(),
                    completion,
                    full_state.0.as_ptr(),
                    0,
                ))
            })?
        };

        let (_, (_, output)) = completion.wait_for().await?;

        T::complete(output)
    }
}

pub trait ReadOp
where
    Self: Sized,
{
    type OperationState: Default;
    type Output;

    fn construct_in_place(
        &self,
        read_op: rados_read_op_t,
        state: Pin<&mut Self::OperationState>,
    ) -> Result<()>;

    fn complete(state: Self::OperationState) -> Result<Self::Output>;
}