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

risingwave_pb/
stream_service.rs
