nfs3_client 0.9.0

//! RPC client implementation

use std::fmt::Debug;

use nfs3_types::rpc::{
    RPC_VERSION_2, accept_stat_data, call_body, fragment_header, msg_body, opaque_auth, reply_body,
    rpc_msg,
};
use nfs3_types::xdr_codec::{Pack, Unpack};

use crate::RpcError;
use crate::io::{AsyncRead, AsyncWrite};

/// RPC client
pub struct RpcClient<IO> {
    io: IO,
    xid: u32,
    credential: opaque_auth<'static>,
    verifier: opaque_auth<'static>,
}

impl<IO> Debug for RpcClient<IO> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
        f.debug_struct("RpcClient").finish()
    }
}

impl<IO> RpcClient<IO>
where
    IO: AsyncRead + AsyncWrite + Send,
{
    /// Create a new RPC client. XID is initialized to a random value.
    pub fn new(io: IO) -> Self {
        Self::new_with_auth(io, opaque_auth::default(), opaque_auth::default())
    }

    /// Create a new RPC client with custom credential and verifier.
    pub fn new_with_auth(
        io: IO,
        credential: opaque_auth<'static>,
        verifier: opaque_auth<'static>,
    ) -> Self {
        Self {
            io,
            xid: fastrand::u32(..),
            credential,
            verifier,
        }
    }

    /// Call an RPC procedure
    ///
    /// # Errors and connection state
    ///
    /// On a successful return the connection is in a clean state and may be
    /// reused for the next call. When an error is returned, use
    /// [`RpcError::is_connection_reusable`] to decide whether the
    /// connection can be kept.
    #[expect(
        clippy::similar_names,
        reason = "prog and proc are fields of call_body"
    )]
    pub async fn call<C, R>(
        &mut self,
        prog: u32,
        vers: u32,
        proc: u32,
        args: &C,
    ) -> Result<R, RpcError>
    where
        R: Unpack,
        C: Pack + Send + Sync,
    {
        let call = call_body {
            rpcvers: RPC_VERSION_2,
            prog,
            vers,
            proc,
            cred: self.credential.borrow(),
            verf: self.verifier.borrow(),
        };
        let msg = rpc_msg {
            xid: self.xid,
            body: msg_body::CALL(call),
        };
        self.xid = self.xid.wrapping_add(1);

        Self::send_call(&mut self.io, &msg, args).await?;
        Self::recv_reply::<R>(&mut self.io, msg.xid).await
    }

    async fn send_call<T>(io: &mut IO, msg: &rpc_msg<'_, '_>, args: &T) -> Result<(), RpcError>
    where
        T: Pack + Send + Sync,
    {
        let total_len = msg.packed_size() + args.packed_size();
        if !total_len.is_multiple_of(4) {
            return Err(RpcError::WrongLength);
        }

        let fragment_header = nfs3_types::rpc::fragment_header::new(
            u32::try_from(total_len).expect("message is too large"),
            true,
        );
        let mut buf = Vec::with_capacity(total_len + 4);
        fragment_header.pack(&mut buf)?;
        msg.pack(&mut buf)?;
        args.pack(&mut buf)?;
        if buf.len() - 4 != total_len {
            return Err(RpcError::WrongLength);
        }
        io.async_write_all(&buf).await?;
        Ok(())
    }

    async fn recv_reply<T>(io: &mut IO, xid: u32) -> Result<T, RpcError>
    where
        T: Unpack,
    {
        let mut buf = [0u8; 4];
        io.async_read_exact(&mut buf).await?;
        let fragment_header: fragment_header = buf.into();
        if !fragment_header.eof() {
            return Err(RpcError::FragmentedReply);
        }

        let total_len = fragment_header.fragment_length();
        let mut buf = vec![0u8; total_len as usize];
        io.async_read_exact(&mut buf).await?;

        let mut cursor = std::io::Cursor::new(buf);
        let (resp_msg, _) = rpc_msg::unpack(&mut cursor)?;

        if resp_msg.xid != xid {
            return Err(RpcError::UnexpectedXid);
        }

        let reply = match resp_msg.body {
            msg_body::REPLY(reply_body::MSG_ACCEPTED(reply)) => reply,
            msg_body::REPLY(reply_body::MSG_DENIED(r)) => return Err(RpcError::from(r)),
            msg_body::CALL(_) => return Err(RpcError::UnexpectedCall),
        };

        if !matches!(reply.reply_data, accept_stat_data::SUCCESS) {
            return Err(RpcError::try_from(reply.reply_data)
                .expect("accept_stat_data::SUCCESS is not a valid error"));
        }

        let (final_value, _) = T::unpack(&mut cursor)?;
        if cursor.position() != u64::from(total_len) {
            let pos = cursor.position();
            return Err(RpcError::NotFullyParsed {
                buf: cursor.into_inner(),
                pos,
            });
        }
        Ok(final_value)
    }
}