diff options
Diffstat (limited to 'tvix')
-rw-r--r-- | tvix/store/src/errors.rs | 7 | ||||
-rw-r--r-- | tvix/store/src/pathinfoservice/grpc.rs | 71 | ||||
-rw-r--r-- | tvix/store/src/pathinfoservice/mod.rs | 2 |
3 files changed, 80 insertions, 0 deletions
diff --git a/tvix/store/src/errors.rs b/tvix/store/src/errors.rs index 36a4320cb3e6..3b23f972b045 100644 --- a/tvix/store/src/errors.rs +++ b/tvix/store/src/errors.rs @@ -1,5 +1,6 @@ use std::sync::PoisonError; use thiserror::Error; +use tokio::task::JoinError; use tonic::Status; /// Errors related to communication with the store. @@ -18,6 +19,12 @@ impl<T> From<PoisonError<T>> for Error { } } +impl From<JoinError> for Error { + fn from(value: JoinError) -> Self { + Error::StorageError(value.to_string()) + } +} + impl From<Error> for Status { fn from(value: Error) -> Self { match value { diff --git a/tvix/store/src/pathinfoservice/grpc.rs b/tvix/store/src/pathinfoservice/grpc.rs new file mode 100644 index 000000000000..821331fbdef5 --- /dev/null +++ b/tvix/store/src/pathinfoservice/grpc.rs @@ -0,0 +1,71 @@ +use super::PathInfoService; +use crate::proto; +use tonic::{transport::Channel, Code, Status}; + +/// Connects to a (remote) tvix-store PathInfoService over gRPC. +#[derive(Clone)] +pub struct GRPCPathInfoService { + /// A handle into the active tokio runtime. Necessary to spawn tasks. + tokio_handle: tokio::runtime::Handle, + + /// The internal reference to a gRPC client. + /// Cloning it is cheap, and it internally handles concurrent requests. + grpc_client: proto::path_info_service_client::PathInfoServiceClient<Channel>, +} + +impl GRPCPathInfoService { + /// Construct a new [GRPCPathInfoService], by passing a handle to the tokio + /// runtime, and a gRPC client. + pub fn new( + tokio_handle: tokio::runtime::Handle, + grpc_client: proto::path_info_service_client::PathInfoServiceClient<Channel>, + ) -> Self { + Self { + tokio_handle, + grpc_client, + } + } +} + +impl PathInfoService for GRPCPathInfoService { + fn get( + &self, + by_what: proto::get_path_info_request::ByWhat, + ) -> Result<Option<proto::PathInfo>, crate::Error> { + // Get a new handle to the gRPC client. + let mut grpc_client = self.grpc_client.clone(); + + let task: tokio::task::JoinHandle<Result<proto::PathInfo, Status>> = + self.tokio_handle.spawn(async move { + let path_info = grpc_client + .get(proto::GetPathInfoRequest { + by_what: Some(by_what), + }) + .await? + .into_inner(); + + Ok(path_info) + }); + + match self.tokio_handle.block_on(task)? { + Ok(path_info) => Ok(Some(path_info)), + Err(e) if e.code() == Code::NotFound => Ok(None), + Err(e) => Err(crate::Error::StorageError(e.to_string())), + } + } + + fn put(&self, path_info: proto::PathInfo) -> Result<proto::PathInfo, crate::Error> { + // Get a new handle to the gRPC client. + let mut grpc_client = self.grpc_client.clone(); + + let task: tokio::task::JoinHandle<Result<proto::PathInfo, Status>> = + self.tokio_handle.spawn(async move { + let path_info = grpc_client.put(path_info).await?.into_inner(); + Ok(path_info) + }); + + self.tokio_handle + .block_on(task)? + .map_err(|e| crate::Error::StorageError(e.to_string())) + } +} diff --git a/tvix/store/src/pathinfoservice/mod.rs b/tvix/store/src/pathinfoservice/mod.rs index b20985dd80ac..410bd205be1d 100644 --- a/tvix/store/src/pathinfoservice/mod.rs +++ b/tvix/store/src/pathinfoservice/mod.rs @@ -1,8 +1,10 @@ +mod grpc; mod memory; mod sled; use crate::{proto, Error}; +pub use self::grpc::GRPCPathInfoService; pub use self::memory::MemoryPathInfoService; pub use self::sled::SledPathInfoService; |