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

risingwave_pb/
stream_service.rs
