Failed to implement Future for PutFut

kv.rs

use crate::error::Result;
use curp::client::ClientApi;
use futures::{future::BoxFuture, FutureExt};
use pin_project::pin_project;
use std::future::Future;
use std::task::Poll;
use std::{pin::pin, sync::Arc};
use tonic::Status;
use xlineapi::{
    command::{Command, CommandResponse, KeyRange, SyncResponse},
    execute_error::ExecuteError,
    RequestWrapper,
};
pub use xlineapi::{
    CompactionResponse, CompareResult, CompareTarget, DeleteRangeResponse, PutResponse,
    RangeResponse, Response, ResponseOp, SortOrder, SortTarget, TargetUnion, TxnResponse,
};

/// Future for `Put`, make it awaitable.
///
/// Before first poll, fut will be [`Option::None`].
/// Once It's been polled, fut will be [`Option::Some`] and inner will be [`Option::None`].
#[pin_project]
pub struct PutFut<'a> {
    #[pin]
    fut: Option<
        BoxFuture<
            'a,
            std::result::Result<
                std::result::Result<(CommandResponse, Option<SyncResponse>), ExecuteError>,
                Status,
            >,
        >,
    >,
    inner: Option<xlineapi::PutRequest>,
    curp_client: &'a Arc<dyn ClientApi<Error = Status, Cmd = Command> + Send + Sync>,
    token: Option<&'a String>,
}

impl<'a> PutFut<'a> {
    #[inline]
    #[must_use]
    /// `key` is the key, in bytes, to put into the key-value store.
    /// `value` is the value, in bytes, to associate with the key in the key-value store.
    pub fn new(
        curp_client: &'a Arc<dyn ClientApi<Error = Status, Cmd = Command> + Send + Sync>,
        token: Option<&'a String>,
        key: Vec<u8>,
        value: Vec<u8>,
    ) -> Self {
        Self {
            curp_client,
            token,
            fut: None,
            inner: Some(xlineapi::PutRequest {
                key,
                value,
                ..Default::default()
            }),
        }
    }

    /// lease is the lease ID to associate with the key in the key-value store.
    /// A lease value of 0 indicates no lease.
    #[inline]
    #[must_use]
    pub fn with_lease(mut self, lease: i64) -> Self {
        self.inner = self.inner.map(|mut inner| {
            inner.lease = lease;
            inner
        });
        self
    }

    /// If `prev_kv` is set, Xline gets the previous key-value pair before changing it.
    /// The previous key-value pair will be returned in the put response.
    #[inline]
    #[must_use]
    pub fn with_prev_kv(mut self, prev_kv: bool) -> Self {
        self.inner = self.inner.map(|mut inner| {
            inner.prev_kv = prev_kv;
            inner
        });
        self
    }

    /// If `ignore_value` is set, Xline updates the key using its current value.
    /// Returns an error if the key does not exist.
    #[inline]
    #[must_use]
    pub fn with_ignore_value(mut self, ignore_value: bool) -> Self {
        self.inner = self.inner.map(|mut inner| {
            inner.ignore_value = ignore_value;
            inner
        });
        self
    }

    /// If `ignore_lease` is set, Xline updates the key using its current lease.
    /// Returns an error if the key does not exist.
    #[inline]
    #[must_use]
    pub fn with_ignore_lease(mut self, ignore_lease: bool) -> Self {
        self.inner = self.inner.map(|mut inner| {
            inner.ignore_lease = ignore_lease;
            inner
        });
        self
    }
}

impl Future for PutFut<'_> {
    type Output = Result<PutResponse>;

    /// Poll the inner future constructed by [`xlineapi::PutRequest`].
    ///
    /// # panic
    ///
    /// panic if inner is `None`.
    fn poll(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Self::Output> {
        if self.fut.is_none() {
            let cmd = Command::new(RequestWrapper::from(self.inner.take().unwrap()));
            self.fut = Some(self.curp_client.propose(&cmd, self.token, true));
        }
        match self.fut.as_mut().unwrap().poll_unpin(cx) {
            Poll::Ready(res) => Poll::Ready(Ok(res??.0.into_inner().into())),
            Poll::Pending => Poll::Pending,
        }
    }
}