risingwave_pb/
stream_service.rs

1// This file is @generated by prost-build.
2#[derive(prost_helpers::AnyPB)]
3#[derive(Clone, PartialEq, ::prost::Message)]
4pub struct InjectBarrierRequest {
5    #[prost(string, tag = "1")]
6    pub request_id: ::prost::alloc::string::String,
7    #[prost(message, optional, tag = "2")]
8    pub barrier: ::core::option::Option<super::stream_plan::Barrier>,
9    #[prost(uint32, tag = "3")]
10    pub database_id: u32,
11    #[prost(uint32, repeated, tag = "4")]
12    pub actor_ids_to_collect: ::prost::alloc::vec::Vec<u32>,
13    #[prost(uint32, repeated, tag = "5")]
14    pub table_ids_to_sync: ::prost::alloc::vec::Vec<u32>,
15    #[prost(uint32, tag = "6")]
16    pub partial_graph_id: u32,
17    #[prost(message, repeated, tag = "9")]
18    pub actors_to_build: ::prost::alloc::vec::Vec<
19        inject_barrier_request::FragmentBuildActorInfo,
20    >,
21    #[prost(message, repeated, tag = "10")]
22    pub subscriptions_to_add: ::prost::alloc::vec::Vec<
23        super::stream_plan::SubscriptionUpstreamInfo,
24    >,
25    #[prost(message, repeated, tag = "11")]
26    pub subscriptions_to_remove: ::prost::alloc::vec::Vec<
27        super::stream_plan::SubscriptionUpstreamInfo,
28    >,
29}
30/// Nested message and enum types in `InjectBarrierRequest`.
31pub mod inject_barrier_request {
32    #[derive(prost_helpers::AnyPB)]
33    #[derive(Clone, PartialEq, ::prost::Message)]
34    pub struct FragmentBuildActorInfo {
35        #[prost(uint32, tag = "1")]
36        pub fragment_id: u32,
37        #[prost(message, optional, tag = "2")]
38        pub node: ::core::option::Option<super::super::stream_plan::StreamNode>,
39        #[prost(message, repeated, tag = "3")]
40        pub actors: ::prost::alloc::vec::Vec<BuildActorInfo>,
41    }
42    #[derive(prost_helpers::AnyPB)]
43    #[derive(Clone, PartialEq, ::prost::Message)]
44    pub struct BuildActorInfo {
45        #[prost(uint32, tag = "1")]
46        pub actor_id: u32,
47        #[prost(map = "uint32, message", tag = "2")]
48        pub fragment_upstreams: ::std::collections::HashMap<
49            u32,
50            build_actor_info::UpstreamActors,
51        >,
52        #[prost(message, repeated, tag = "3")]
53        pub dispatchers: ::prost::alloc::vec::Vec<super::super::stream_plan::Dispatcher>,
54        #[prost(message, optional, tag = "4")]
55        pub vnode_bitmap: ::core::option::Option<super::super::common::Buffer>,
56        #[prost(string, tag = "5")]
57        pub mview_definition: ::prost::alloc::string::String,
58        #[prost(message, optional, tag = "6")]
59        pub expr_context: ::core::option::Option<super::super::plan_common::ExprContext>,
60    }
61    /// Nested message and enum types in `BuildActorInfo`.
62    pub mod build_actor_info {
63        #[derive(prost_helpers::AnyPB)]
64        #[derive(Clone, PartialEq, ::prost::Message)]
65        pub struct UpstreamActors {
66            #[prost(message, repeated, tag = "1")]
67            pub actors: ::prost::alloc::vec::Vec<super::super::super::common::ActorInfo>,
68        }
69    }
70}
71#[derive(prost_helpers::AnyPB)]
72#[derive(Clone, PartialEq, ::prost::Message)]
73pub struct BarrierCompleteResponse {
74    #[prost(string, tag = "1")]
75    pub request_id: ::prost::alloc::string::String,
76    #[prost(message, optional, tag = "2")]
77    pub status: ::core::option::Option<super::common::Status>,
78    #[prost(message, repeated, tag = "3")]
79    pub create_mview_progress: ::prost::alloc::vec::Vec<
80        barrier_complete_response::CreateMviewProgress,
81    >,
82    #[prost(message, repeated, tag = "4")]
83    pub synced_sstables: ::prost::alloc::vec::Vec<
84        barrier_complete_response::LocalSstableInfo,
85    >,
86    #[prost(uint32, tag = "5")]
87    pub worker_id: u32,
88    #[prost(map = "uint32, message", tag = "6")]
89    pub table_watermarks: ::std::collections::HashMap<
90        u32,
91        super::hummock::TableWatermarks,
92    >,
93    #[prost(message, repeated, tag = "7")]
94    pub old_value_sstables: ::prost::alloc::vec::Vec<super::hummock::SstableInfo>,
95    #[prost(uint32, tag = "8")]
96    pub partial_graph_id: u32,
97    /// prev_epoch of barrier
98    #[prost(uint64, tag = "9")]
99    pub epoch: u64,
100    #[prost(uint32, tag = "10")]
101    pub database_id: u32,
102}
103/// Nested message and enum types in `BarrierCompleteResponse`.
104pub mod barrier_complete_response {
105    #[derive(prost_helpers::AnyPB)]
106    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
107    pub struct CreateMviewProgress {
108        /// Note: ideally we should use `executor_id`, but `actor_id` is ok-ish.
109        /// See <<https://github.com/risingwavelabs/risingwave/issues/6236>.>
110        #[prost(uint32, tag = "1")]
111        pub backfill_actor_id: u32,
112        #[prost(bool, tag = "2")]
113        pub done: bool,
114        /// MV backfill snapshot read epoch (0 for Done / Source backfill)
115        #[prost(uint64, tag = "3")]
116        pub consumed_epoch: u64,
117        /// MV backfill snapshot read rows / Source backfilled rows
118        #[prost(uint64, tag = "4")]
119        pub consumed_rows: u64,
120        #[prost(uint64, tag = "5")]
121        pub pending_epoch_lag: u64,
122    }
123    #[derive(prost_helpers::AnyPB)]
124    #[derive(Clone, PartialEq, ::prost::Message)]
125    pub struct LocalSstableInfo {
126        #[prost(message, optional, tag = "2")]
127        pub sst: ::core::option::Option<super::super::hummock::SstableInfo>,
128        #[prost(map = "uint32, message", tag = "3")]
129        pub table_stats_map: ::std::collections::HashMap<
130            u32,
131            super::super::hummock::TableStats,
132        >,
133        #[prost(uint64, tag = "4")]
134        pub created_at: u64,
135    }
136}
137#[derive(prost_helpers::AnyPB)]
138#[derive(Clone, PartialEq, ::prost::Message)]
139pub struct StreamingControlStreamRequest {
140    #[prost(oneof = "streaming_control_stream_request::Request", tags = "1, 2, 3, 4, 5")]
141    pub request: ::core::option::Option<streaming_control_stream_request::Request>,
142}
143/// Nested message and enum types in `StreamingControlStreamRequest`.
144pub mod streaming_control_stream_request {
145    #[derive(prost_helpers::AnyPB)]
146    #[derive(Clone, PartialEq, ::prost::Message)]
147    pub struct InitialPartialGraph {
148        #[prost(uint32, tag = "1")]
149        pub partial_graph_id: u32,
150        #[prost(message, repeated, tag = "2")]
151        pub subscriptions: ::prost::alloc::vec::Vec<
152            super::super::stream_plan::SubscriptionUpstreamInfo,
153        >,
154    }
155    #[derive(prost_helpers::AnyPB)]
156    #[derive(Clone, PartialEq, ::prost::Message)]
157    pub struct DatabaseInitialPartialGraph {
158        #[prost(uint32, tag = "1")]
159        pub database_id: u32,
160        #[prost(message, repeated, tag = "2")]
161        pub graphs: ::prost::alloc::vec::Vec<InitialPartialGraph>,
162    }
163    #[derive(prost_helpers::AnyPB)]
164    #[derive(Clone, PartialEq, ::prost::Message)]
165    pub struct InitRequest {
166        #[prost(message, repeated, tag = "1")]
167        pub databases: ::prost::alloc::vec::Vec<DatabaseInitialPartialGraph>,
168        #[prost(string, tag = "2")]
169        pub term_id: ::prost::alloc::string::String,
170    }
171    #[derive(prost_helpers::AnyPB)]
172    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
173    pub struct CreatePartialGraphRequest {
174        #[prost(uint32, tag = "1")]
175        pub partial_graph_id: u32,
176        #[prost(uint32, tag = "2")]
177        pub database_id: u32,
178    }
179    #[derive(prost_helpers::AnyPB)]
180    #[derive(Clone, PartialEq, ::prost::Message)]
181    pub struct RemovePartialGraphRequest {
182        #[prost(uint32, repeated, tag = "1")]
183        pub partial_graph_ids: ::prost::alloc::vec::Vec<u32>,
184        #[prost(uint32, tag = "2")]
185        pub database_id: u32,
186    }
187    #[derive(prost_helpers::AnyPB)]
188    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
189    pub struct ResetDatabaseRequest {
190        #[prost(uint32, tag = "1")]
191        pub database_id: u32,
192        #[prost(uint32, tag = "2")]
193        pub reset_request_id: u32,
194    }
195    #[derive(prost_helpers::AnyPB)]
196    #[derive(Clone, PartialEq, ::prost::Oneof)]
197    pub enum Request {
198        #[prost(message, tag = "1")]
199        Init(InitRequest),
200        #[prost(message, tag = "2")]
201        InjectBarrier(super::InjectBarrierRequest),
202        #[prost(message, tag = "3")]
203        RemovePartialGraph(RemovePartialGraphRequest),
204        #[prost(message, tag = "4")]
205        CreatePartialGraph(CreatePartialGraphRequest),
206        #[prost(message, tag = "5")]
207        ResetDatabase(ResetDatabaseRequest),
208    }
209}
210#[derive(prost_helpers::AnyPB)]
211#[derive(Clone, PartialEq, ::prost::Message)]
212pub struct ScoredError {
213    #[prost(string, tag = "1")]
214    pub err_msg: ::prost::alloc::string::String,
215    #[prost(int32, tag = "2")]
216    pub score: i32,
217}
218#[derive(prost_helpers::AnyPB)]
219#[derive(Clone, PartialEq, ::prost::Message)]
220pub struct StreamingControlStreamResponse {
221    #[prost(
222        oneof = "streaming_control_stream_response::Response",
223        tags = "1, 2, 3, 4, 5"
224    )]
225    pub response: ::core::option::Option<streaming_control_stream_response::Response>,
226}
227/// Nested message and enum types in `StreamingControlStreamResponse`.
228pub mod streaming_control_stream_response {
229    #[derive(prost_helpers::AnyPB)]
230    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
231    pub struct InitResponse {}
232    #[derive(prost_helpers::AnyPB)]
233    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
234    pub struct ShutdownResponse {}
235    #[derive(prost_helpers::AnyPB)]
236    #[derive(Clone, Copy, PartialEq, ::prost::Message)]
237    pub struct ReportDatabaseFailureResponse {
238        #[prost(uint32, tag = "1")]
239        pub database_id: u32,
240    }
241    #[derive(prost_helpers::AnyPB)]
242    #[derive(Clone, PartialEq, ::prost::Message)]
243    pub struct ResetDatabaseResponse {
244        #[prost(uint32, tag = "1")]
245        pub database_id: u32,
246        #[prost(message, optional, tag = "2")]
247        pub root_err: ::core::option::Option<super::ScoredError>,
248        #[prost(uint32, tag = "3")]
249        pub reset_request_id: u32,
250    }
251    #[derive(prost_helpers::AnyPB)]
252    #[derive(Clone, PartialEq, ::prost::Oneof)]
253    pub enum Response {
254        #[prost(message, tag = "1")]
255        Init(InitResponse),
256        #[prost(message, tag = "2")]
257        CompleteBarrier(super::BarrierCompleteResponse),
258        #[prost(message, tag = "3")]
259        Shutdown(ShutdownResponse),
260        #[prost(message, tag = "4")]
261        ReportDatabaseFailure(ReportDatabaseFailureResponse),
262        #[prost(message, tag = "5")]
263        ResetDatabase(ResetDatabaseResponse),
264    }
265}
266#[derive(prost_helpers::AnyPB)]
267#[derive(Clone, Copy, PartialEq, ::prost::Message)]
268pub struct GetMinUncommittedObjectIdRequest {}
269#[derive(prost_helpers::AnyPB)]
270#[derive(Clone, Copy, PartialEq, ::prost::Message)]
271pub struct GetMinUncommittedObjectIdResponse {
272    #[prost(uint64, tag = "1")]
273    pub min_uncommitted_object_id: u64,
274}
275/// Generated client implementations.
276pub mod stream_service_client {
277    #![allow(
278        unused_variables,
279        dead_code,
280        missing_docs,
281        clippy::wildcard_imports,
282        clippy::let_unit_value,
283    )]
284    use tonic::codegen::*;
285    use tonic::codegen::http::Uri;
286    #[derive(Debug, Clone)]
287    pub struct StreamServiceClient<T> {
288        inner: tonic::client::Grpc<T>,
289    }
290    impl StreamServiceClient<tonic::transport::Channel> {
291        /// Attempt to create a new client by connecting to a given endpoint.
292        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
293        where
294            D: TryInto<tonic::transport::Endpoint>,
295            D::Error: Into<StdError>,
296        {
297            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
298            Ok(Self::new(conn))
299        }
300    }
301    impl<T> StreamServiceClient<T>
302    where
303        T: tonic::client::GrpcService<tonic::body::BoxBody>,
304        T::Error: Into<StdError>,
305        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
306        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
307    {
308        pub fn new(inner: T) -> Self {
309            let inner = tonic::client::Grpc::new(inner);
310            Self { inner }
311        }
312        pub fn with_origin(inner: T, origin: Uri) -> Self {
313            let inner = tonic::client::Grpc::with_origin(inner, origin);
314            Self { inner }
315        }
316        pub fn with_interceptor<F>(
317            inner: T,
318            interceptor: F,
319        ) -> StreamServiceClient<InterceptedService<T, F>>
320        where
321            F: tonic::service::Interceptor,
322            T::ResponseBody: Default,
323            T: tonic::codegen::Service<
324                http::Request<tonic::body::BoxBody>,
325                Response = http::Response<
326                    <T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody,
327                >,
328            >,
329            <T as tonic::codegen::Service<
330                http::Request<tonic::body::BoxBody>,
331            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
332        {
333            StreamServiceClient::new(InterceptedService::new(inner, interceptor))
334        }
335        /// Compress requests with the given encoding.
336        ///
337        /// This requires the server to support it otherwise it might respond with an
338        /// error.
339        #[must_use]
340        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
341            self.inner = self.inner.send_compressed(encoding);
342            self
343        }
344        /// Enable decompressing responses.
345        #[must_use]
346        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
347            self.inner = self.inner.accept_compressed(encoding);
348            self
349        }
350        /// Limits the maximum size of a decoded message.
351        ///
352        /// Default: `4MB`
353        #[must_use]
354        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
355            self.inner = self.inner.max_decoding_message_size(limit);
356            self
357        }
358        /// Limits the maximum size of an encoded message.
359        ///
360        /// Default: `usize::MAX`
361        #[must_use]
362        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
363            self.inner = self.inner.max_encoding_message_size(limit);
364            self
365        }
366        pub async fn streaming_control_stream(
367            &mut self,
368            request: impl tonic::IntoStreamingRequest<
369                Message = super::StreamingControlStreamRequest,
370            >,
371        ) -> std::result::Result<
372            tonic::Response<
373                tonic::codec::Streaming<super::StreamingControlStreamResponse>,
374            >,
375            tonic::Status,
376        > {
377            self.inner
378                .ready()
379                .await
380                .map_err(|e| {
381                    tonic::Status::unknown(
382                        format!("Service was not ready: {}", e.into()),
383                    )
384                })?;
385            let codec = tonic::codec::ProstCodec::default();
386            let path = http::uri::PathAndQuery::from_static(
387                "/stream_service.StreamService/StreamingControlStream",
388            );
389            let mut req = request.into_streaming_request();
390            req.extensions_mut()
391                .insert(
392                    GrpcMethod::new(
393                        "stream_service.StreamService",
394                        "StreamingControlStream",
395                    ),
396                );
397            self.inner.streaming(req, path, codec).await
398        }
399        pub async fn get_min_uncommitted_object_id(
400            &mut self,
401            request: impl tonic::IntoRequest<super::GetMinUncommittedObjectIdRequest>,
402        ) -> std::result::Result<
403            tonic::Response<super::GetMinUncommittedObjectIdResponse>,
404            tonic::Status,
405        > {
406            self.inner
407                .ready()
408                .await
409                .map_err(|e| {
410                    tonic::Status::unknown(
411                        format!("Service was not ready: {}", e.into()),
412                    )
413                })?;
414            let codec = tonic::codec::ProstCodec::default();
415            let path = http::uri::PathAndQuery::from_static(
416                "/stream_service.StreamService/GetMinUncommittedObjectId",
417            );
418            let mut req = request.into_request();
419            req.extensions_mut()
420                .insert(
421                    GrpcMethod::new(
422                        "stream_service.StreamService",
423                        "GetMinUncommittedObjectId",
424                    ),
425                );
426            self.inner.unary(req, path, codec).await
427        }
428    }
429}
430/// Generated server implementations.
431pub mod stream_service_server {
432    #![allow(
433        unused_variables,
434        dead_code,
435        missing_docs,
436        clippy::wildcard_imports,
437        clippy::let_unit_value,
438    )]
439    use tonic::codegen::*;
440    /// Generated trait containing gRPC methods that should be implemented for use with StreamServiceServer.
441    #[async_trait]
442    pub trait StreamService: std::marker::Send + std::marker::Sync + 'static {
443        /// Server streaming response type for the StreamingControlStream method.
444        type StreamingControlStreamStream: tonic::codegen::tokio_stream::Stream<
445                Item = std::result::Result<
446                    super::StreamingControlStreamResponse,
447                    tonic::Status,
448                >,
449            >
450            + std::marker::Send
451            + 'static;
452        async fn streaming_control_stream(
453            &self,
454            request: tonic::Request<
455                tonic::Streaming<super::StreamingControlStreamRequest>,
456            >,
457        ) -> std::result::Result<
458            tonic::Response<Self::StreamingControlStreamStream>,
459            tonic::Status,
460        >;
461        async fn get_min_uncommitted_object_id(
462            &self,
463            request: tonic::Request<super::GetMinUncommittedObjectIdRequest>,
464        ) -> std::result::Result<
465            tonic::Response<super::GetMinUncommittedObjectIdResponse>,
466            tonic::Status,
467        >;
468    }
469    #[derive(Debug)]
470    pub struct StreamServiceServer<T> {
471        inner: Arc<T>,
472        accept_compression_encodings: EnabledCompressionEncodings,
473        send_compression_encodings: EnabledCompressionEncodings,
474        max_decoding_message_size: Option<usize>,
475        max_encoding_message_size: Option<usize>,
476    }
477    impl<T> StreamServiceServer<T> {
478        pub fn new(inner: T) -> Self {
479            Self::from_arc(Arc::new(inner))
480        }
481        pub fn from_arc(inner: Arc<T>) -> Self {
482            Self {
483                inner,
484                accept_compression_encodings: Default::default(),
485                send_compression_encodings: Default::default(),
486                max_decoding_message_size: None,
487                max_encoding_message_size: None,
488            }
489        }
490        pub fn with_interceptor<F>(
491            inner: T,
492            interceptor: F,
493        ) -> InterceptedService<Self, F>
494        where
495            F: tonic::service::Interceptor,
496        {
497            InterceptedService::new(Self::new(inner), interceptor)
498        }
499        /// Enable decompressing requests with the given encoding.
500        #[must_use]
501        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
502            self.accept_compression_encodings.enable(encoding);
503            self
504        }
505        /// Compress responses with the given encoding, if the client supports it.
506        #[must_use]
507        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
508            self.send_compression_encodings.enable(encoding);
509            self
510        }
511        /// Limits the maximum size of a decoded message.
512        ///
513        /// Default: `4MB`
514        #[must_use]
515        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
516            self.max_decoding_message_size = Some(limit);
517            self
518        }
519        /// Limits the maximum size of an encoded message.
520        ///
521        /// Default: `usize::MAX`
522        #[must_use]
523        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
524            self.max_encoding_message_size = Some(limit);
525            self
526        }
527    }
528    impl<T, B> tonic::codegen::Service<http::Request<B>> for StreamServiceServer<T>
529    where
530        T: StreamService,
531        B: Body + std::marker::Send + 'static,
532        B::Error: Into<StdError> + std::marker::Send + 'static,
533    {
534        type Response = http::Response<tonic::body::BoxBody>;
535        type Error = std::convert::Infallible;
536        type Future = BoxFuture<Self::Response, Self::Error>;
537        fn poll_ready(
538            &mut self,
539            _cx: &mut Context<'_>,
540        ) -> Poll<std::result::Result<(), Self::Error>> {
541            Poll::Ready(Ok(()))
542        }
543        fn call(&mut self, req: http::Request<B>) -> Self::Future {
544            match req.uri().path() {
545                "/stream_service.StreamService/StreamingControlStream" => {
546                    #[allow(non_camel_case_types)]
547                    struct StreamingControlStreamSvc<T: StreamService>(pub Arc<T>);
548                    impl<
549                        T: StreamService,
550                    > tonic::server::StreamingService<
551                        super::StreamingControlStreamRequest,
552                    > for StreamingControlStreamSvc<T> {
553                        type Response = super::StreamingControlStreamResponse;
554                        type ResponseStream = T::StreamingControlStreamStream;
555                        type Future = BoxFuture<
556                            tonic::Response<Self::ResponseStream>,
557                            tonic::Status,
558                        >;
559                        fn call(
560                            &mut self,
561                            request: tonic::Request<
562                                tonic::Streaming<super::StreamingControlStreamRequest>,
563                            >,
564                        ) -> Self::Future {
565                            let inner = Arc::clone(&self.0);
566                            let fut = async move {
567                                <T as StreamService>::streaming_control_stream(
568                                        &inner,
569                                        request,
570                                    )
571                                    .await
572                            };
573                            Box::pin(fut)
574                        }
575                    }
576                    let accept_compression_encodings = self.accept_compression_encodings;
577                    let send_compression_encodings = self.send_compression_encodings;
578                    let max_decoding_message_size = self.max_decoding_message_size;
579                    let max_encoding_message_size = self.max_encoding_message_size;
580                    let inner = self.inner.clone();
581                    let fut = async move {
582                        let method = StreamingControlStreamSvc(inner);
583                        let codec = tonic::codec::ProstCodec::default();
584                        let mut grpc = tonic::server::Grpc::new(codec)
585                            .apply_compression_config(
586                                accept_compression_encodings,
587                                send_compression_encodings,
588                            )
589                            .apply_max_message_size_config(
590                                max_decoding_message_size,
591                                max_encoding_message_size,
592                            );
593                        let res = grpc.streaming(method, req).await;
594                        Ok(res)
595                    };
596                    Box::pin(fut)
597                }
598                "/stream_service.StreamService/GetMinUncommittedObjectId" => {
599                    #[allow(non_camel_case_types)]
600                    struct GetMinUncommittedObjectIdSvc<T: StreamService>(pub Arc<T>);
601                    impl<
602                        T: StreamService,
603                    > tonic::server::UnaryService<
604                        super::GetMinUncommittedObjectIdRequest,
605                    > for GetMinUncommittedObjectIdSvc<T> {
606                        type Response = super::GetMinUncommittedObjectIdResponse;
607                        type Future = BoxFuture<
608                            tonic::Response<Self::Response>,
609                            tonic::Status,
610                        >;
611                        fn call(
612                            &mut self,
613                            request: tonic::Request<
614                                super::GetMinUncommittedObjectIdRequest,
615                            >,
616                        ) -> Self::Future {
617                            let inner = Arc::clone(&self.0);
618                            let fut = async move {
619                                <T as StreamService>::get_min_uncommitted_object_id(
620                                        &inner,
621                                        request,
622                                    )
623                                    .await
624                            };
625                            Box::pin(fut)
626                        }
627                    }
628                    let accept_compression_encodings = self.accept_compression_encodings;
629                    let send_compression_encodings = self.send_compression_encodings;
630                    let max_decoding_message_size = self.max_decoding_message_size;
631                    let max_encoding_message_size = self.max_encoding_message_size;
632                    let inner = self.inner.clone();
633                    let fut = async move {
634                        let method = GetMinUncommittedObjectIdSvc(inner);
635                        let codec = tonic::codec::ProstCodec::default();
636                        let mut grpc = tonic::server::Grpc::new(codec)
637                            .apply_compression_config(
638                                accept_compression_encodings,
639                                send_compression_encodings,
640                            )
641                            .apply_max_message_size_config(
642                                max_decoding_message_size,
643                                max_encoding_message_size,
644                            );
645                        let res = grpc.unary(method, req).await;
646                        Ok(res)
647                    };
648                    Box::pin(fut)
649                }
650                _ => {
651                    Box::pin(async move {
652                        let mut response = http::Response::new(empty_body());
653                        let headers = response.headers_mut();
654                        headers
655                            .insert(
656                                tonic::Status::GRPC_STATUS,
657                                (tonic::Code::Unimplemented as i32).into(),
658                            );
659                        headers
660                            .insert(
661                                http::header::CONTENT_TYPE,
662                                tonic::metadata::GRPC_CONTENT_TYPE,
663                            );
664                        Ok(response)
665                    })
666                }
667            }
668        }
669    }
670    impl<T> Clone for StreamServiceServer<T> {
671        fn clone(&self) -> Self {
672            let inner = self.inner.clone();
673            Self {
674                inner,
675                accept_compression_encodings: self.accept_compression_encodings,
676                send_compression_encodings: self.send_compression_encodings,
677                max_decoding_message_size: self.max_decoding_message_size,
678                max_encoding_message_size: self.max_encoding_message_size,
679            }
680        }
681    }
682    /// Generated gRPC service name
683    pub const SERVICE_NAME: &str = "stream_service.StreamService";
684    impl<T> tonic::server::NamedService for StreamServiceServer<T> {
685        const NAME: &'static str = SERVICE_NAME;
686    }
687}
risingwave_pb/stream_service.rs

risingwave_pb/
stream_service.rs
