snix_build/buildservice/
grpc.rs

1use futures::stream::BoxStream;
2use futures::{StreamExt, TryStreamExt};
3use tracing::instrument;
4
5use crate::buildservice::{BuildFailure, BuildRequest, BuildUpdate};
6use crate::proto::build_service_client::BuildServiceClient;
7
8use super::BuildService;
9
10pub struct GRPCBuildService<T> {
11    client: BuildServiceClient<T>,
12}
13
14impl<T> GRPCBuildService<T> {
15    #[allow(dead_code)]
16    pub fn from_client(client: BuildServiceClient<T>) -> Self {
17        Self { client }
18    }
19}
20
21impl<T> BuildService for GRPCBuildService<T>
22where
23    T: tonic::client::GrpcService<tonic::body::Body> + Send + Sync + Clone + 'static,
24    T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
25    <T::ResponseBody as tonic::codegen::Body>::Error: Into<tonic::codegen::StdError> + Send,
26    T::Future: Send,
27{
28    #[instrument(skip(self))]
29    fn do_build(&self, request: BuildRequest) -> BoxStream<'_, BuildUpdate> {
30        let mut client = self.client.clone();
31
32        async_stream::try_stream! {
33            let mut stream = client
34                .do_build(tonic::Request::new(request.into()))
35                .await?
36                .into_inner();
37
38            while let Some(proto_build_update) = stream.try_next().await? {
39                yield match BuildUpdate::try_from(proto_build_update) {
40                    Ok(build_update) => build_update,
41                    Err(err) => Err::<_, BuildFailure>(err.into())?,
42                }
43            }
44        }
45        .map(|elem: Result<BuildUpdate, BuildFailure>| match elem {
46            Ok(build_update) => build_update,
47            Err(err) => err.into(),
48        })
49        .boxed()
50    }
51}
snix_build/buildservice/grpc.rs

snix_build/buildservice/
grpc.rs
