diff options
| -rw-r--r-- | tvix/castore/src/blobservice/grpc.rs | 5 | ||||
| -rw-r--r-- | tvix/castore/src/proto/grpc_blobservice_wrapper.rs | 18 | ||||
| -rw-r--r-- | tvix/castore/src/utils.rs | 4 | ||||
| -rw-r--r-- | tvix/store/src/bin/tvix-store.rs | 2 |
4 files changed, 14 insertions, 15 deletions
diff --git a/tvix/castore/src/blobservice/grpc.rs b/tvix/castore/src/blobservice/grpc.rs index f90240d884..faf50eff88 100644 --- a/tvix/castore/src/blobservice/grpc.rs +++ b/tvix/castore/src/blobservice/grpc.rs @@ -221,7 +221,6 @@ impl<W: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for GRPCBlobWriter< #[cfg(test)] mod tests { - use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; @@ -255,8 +254,8 @@ mod tests { let mut server = tonic::transport::Server::builder(); let router = server.add_service(crate::proto::blob_service_server::BlobServiceServer::new( - GRPCBlobServiceWrapper::from( - Arc::new(MemoryBlobService::default()) as Arc<dyn BlobService> + GRPCBlobServiceWrapper::new( + Box::<MemoryBlobService>::default() as Box<dyn BlobService> ), )); router.serve_with_incoming(uds_stream).await diff --git a/tvix/castore/src/proto/grpc_blobservice_wrapper.rs b/tvix/castore/src/proto/grpc_blobservice_wrapper.rs index a37cc299b8..063f0421dd 100644 --- a/tvix/castore/src/proto/grpc_blobservice_wrapper.rs +++ b/tvix/castore/src/proto/grpc_blobservice_wrapper.rs @@ -6,22 +6,19 @@ use std::{ io, ops::{Deref, DerefMut}, pin::Pin, - sync::Arc, }; use tokio_stream::StreamExt; use tokio_util::io::ReaderStream; use tonic::{async_trait, Request, Response, Status, Streaming}; use tracing::{instrument, warn}; -pub struct GRPCBlobServiceWrapper { - blob_service: Arc<dyn BlobService>, +pub struct GRPCBlobServiceWrapper<T> { + blob_service: T, } -impl From<Arc<dyn BlobService>> for GRPCBlobServiceWrapper { - fn from(value: Arc<dyn BlobService>) -> Self { - Self { - blob_service: value, - } +impl<T> GRPCBlobServiceWrapper<T> { + pub fn new(blob_service: T) -> Self { + Self { blob_service } } } @@ -84,7 +81,10 @@ unsafe impl<const N: usize> bytes::BufMut for BytesMutWithDefaultCapacity<N> { } #[async_trait] -impl super::blob_service_server::BlobService for GRPCBlobServiceWrapper { +impl<T> super::blob_service_server::BlobService for GRPCBlobServiceWrapper<T> +where + T: Deref<Target = dyn BlobService> + Send + Sync + 'static, +{ // https://github.com/tokio-rs/tokio/issues/2723#issuecomment-1534723933 type ReadStream = Pin<Box<dyn futures::Stream<Item = Result<super::BlobChunk, Status>> + Send + 'static>>; diff --git a/tvix/castore/src/utils.rs b/tvix/castore/src/utils.rs index b24627ed9b..d9a7254a56 100644 --- a/tvix/castore/src/utils.rs +++ b/tvix/castore/src/utils.rs @@ -1,8 +1,8 @@ //! A crate containing constructors to provide instances of a BlobService and //! DirectoryService. Only used for testing purposes, but across crates. //! Should be removed once we have a better concept of a "Service registry". - use std::sync::Arc; + use tonic::transport::{Channel, Endpoint, Server, Uri}; use crate::{ @@ -67,7 +67,7 @@ pub(crate) async fn gen_blobsvc_grpc_client() -> BlobServiceClient<Channel> { tokio::spawn(async { // spin up a new DirectoryService let mut server = Server::builder(); - let router = server.add_service(BlobServiceServer::new(GRPCBlobServiceWrapper::from( + let router = server.add_service(BlobServiceServer::new(GRPCBlobServiceWrapper::new( gen_blob_service(), ))); diff --git a/tvix/store/src/bin/tvix-store.rs b/tvix/store/src/bin/tvix-store.rs index 59c39f7b4e..a245af78d0 100644 --- a/tvix/store/src/bin/tvix-store.rs +++ b/tvix/store/src/bin/tvix-store.rs @@ -197,7 +197,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> { #[allow(unused_mut)] let mut router = server - .add_service(BlobServiceServer::new(GRPCBlobServiceWrapper::from( + .add_service(BlobServiceServer::new(GRPCBlobServiceWrapper::new( blob_service, ))) .add_service(DirectoryServiceServer::new( |