risingwave_pb/
task_service.rs

1// This file is @generated by prost-build.
2/// Task is a running instance of Stage.
3#[derive(prost_helpers::AnyPB)]
4#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
5pub struct TaskId {
6    #[prost(string, tag = "1")]
7    pub query_id: ::prost::alloc::string::String,
8    #[prost(uint32, tag = "2")]
9    pub stage_id: u32,
10    #[prost(uint32, tag = "3")]
11    pub task_id: u32,
12}
13#[derive(prost_helpers::AnyPB)]
14#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
15pub struct TaskInfoResponse {
16    #[prost(message, optional, tag = "1")]
17    pub task_id: ::core::option::Option<super::batch_plan::TaskId>,
18    #[prost(enumeration = "task_info_response::TaskStatus", tag = "2")]
19    pub task_status: i32,
20    /// Optional error message for failed task.
21    #[prost(string, tag = "3")]
22    pub error_message: ::prost::alloc::string::String,
23}
24/// Nested message and enum types in `TaskInfoResponse`.
25pub mod task_info_response {
26    #[derive(prost_helpers::AnyPB)]
27    #[derive(
28        Clone,
29        Copy,
30        Debug,
31        PartialEq,
32        Eq,
33        Hash,
34        PartialOrd,
35        Ord,
36        ::prost::Enumeration
37    )]
38    #[repr(i32)]
39    pub enum TaskStatus {
40        /// Note: Requirement of proto3: first enum must be 0.
41        Unspecified = 0,
42        Pending = 2,
43        Running = 3,
44        Finished = 6,
45        Failed = 7,
46        Aborted = 8,
47        Cancelled = 9,
48        Ping = 10,
49    }
50    impl TaskStatus {
51        /// String value of the enum field names used in the ProtoBuf definition.
52        ///
53        /// The values are not transformed in any way and thus are considered stable
54        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
55        pub fn as_str_name(&self) -> &'static str {
56            match self {
57                Self::Unspecified => "UNSPECIFIED",
58                Self::Pending => "PENDING",
59                Self::Running => "RUNNING",
60                Self::Finished => "FINISHED",
61                Self::Failed => "FAILED",
62                Self::Aborted => "ABORTED",
63                Self::Cancelled => "CANCELLED",
64                Self::Ping => "PING",
65            }
66        }
67        /// Creates an enum from field names used in the ProtoBuf definition.
68        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
69            match value {
70                "UNSPECIFIED" => Some(Self::Unspecified),
71                "PENDING" => Some(Self::Pending),
72                "RUNNING" => Some(Self::Running),
73                "FINISHED" => Some(Self::Finished),
74                "FAILED" => Some(Self::Failed),
75                "ABORTED" => Some(Self::Aborted),
76                "CANCELLED" => Some(Self::Cancelled),
77                "PING" => Some(Self::Ping),
78                _ => None,
79            }
80        }
81    }
82}
83#[derive(prost_helpers::AnyPB)]
84#[derive(Clone, PartialEq, ::prost::Message)]
85pub struct CreateTaskRequest {
86    #[prost(message, optional, tag = "1")]
87    pub task_id: ::core::option::Option<super::batch_plan::TaskId>,
88    #[prost(message, optional, tag = "2")]
89    pub plan: ::core::option::Option<super::batch_plan::PlanFragment>,
90    #[prost(map = "string, string", tag = "4")]
91    pub tracing_context: ::std::collections::HashMap<
92        ::prost::alloc::string::String,
93        ::prost::alloc::string::String,
94    >,
95    #[prost(message, optional, tag = "5")]
96    pub expr_context: ::core::option::Option<super::plan_common::ExprContext>,
97}
98#[derive(prost_helpers::AnyPB)]
99#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
100pub struct CancelTaskRequest {
101    #[prost(message, optional, tag = "1")]
102    pub task_id: ::core::option::Option<super::batch_plan::TaskId>,
103}
104#[derive(prost_helpers::AnyPB)]
105#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
106pub struct CancelTaskResponse {
107    #[prost(message, optional, tag = "1")]
108    pub status: ::core::option::Option<super::common::Status>,
109}
110#[derive(prost_helpers::AnyPB)]
111#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
112pub struct GetTaskInfoRequest {
113    #[prost(message, optional, tag = "1")]
114    pub task_id: ::core::option::Option<super::batch_plan::TaskId>,
115}
116#[derive(prost_helpers::AnyPB)]
117#[derive(Clone, PartialEq, ::prost::Message)]
118pub struct GetDataResponse {
119    #[prost(message, optional, tag = "2")]
120    pub record_batch: ::core::option::Option<super::data::DataChunk>,
121}
122#[derive(prost_helpers::AnyPB)]
123#[derive(Clone, PartialEq, ::prost::Message)]
124pub struct FastInsertRequest {
125    /// Id of the table to perform inserting.
126    #[prost(uint32, tag = "1", wrapper = "crate::id::TableId")]
127    pub table_id: crate::id::TableId,
128    /// Version of the table.
129    #[prost(uint64, tag = "2")]
130    pub table_version_id: u64,
131    #[prost(message, optional, tag = "4")]
132    pub data_chunk: ::core::option::Option<super::data::DataChunk>,
133    /// An optional field and will be `None` for tables without user-defined pk.
134    /// The `BatchInsertExecutor` should add a column with NULL value which will
135    /// be filled in streaming.
136    #[prost(uint32, optional, tag = "5")]
137    pub row_id_index: ::core::option::Option<u32>,
138    /// Use this number to assign the insert req to different worker nodes and dml channels.
139    #[prost(uint32, tag = "6")]
140    pub request_id: u32,
141    /// TODO(kexiang): add support for default columns. plan_common.ExprContext expr_context is needed for it.
142    #[prost(bool, tag = "7")]
143    pub wait_for_persistence: bool,
144}
145#[derive(prost_helpers::AnyPB)]
146#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
147pub struct FastInsertResponse {
148    #[prost(enumeration = "fast_insert_response::Status", tag = "1")]
149    pub status: i32,
150    #[prost(string, tag = "2")]
151    pub error_message: ::prost::alloc::string::String,
152}
153/// Nested message and enum types in `FastInsertResponse`.
154pub mod fast_insert_response {
155    #[derive(prost_helpers::AnyPB)]
156    #[derive(
157        Clone,
158        Copy,
159        Debug,
160        PartialEq,
161        Eq,
162        Hash,
163        PartialOrd,
164        Ord,
165        ::prost::Enumeration
166    )]
167    #[repr(i32)]
168    pub enum Status {
169        Unspecified = 0,
170        Succeeded = 1,
171        DmlFailed = 2,
172    }
173    impl Status {
174        /// String value of the enum field names used in the ProtoBuf definition.
175        ///
176        /// The values are not transformed in any way and thus are considered stable
177        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
178        pub fn as_str_name(&self) -> &'static str {
179            match self {
180                Self::Unspecified => "UNSPECIFIED",
181                Self::Succeeded => "SUCCEEDED",
182                Self::DmlFailed => "DML_FAILED",
183            }
184        }
185        /// Creates an enum from field names used in the ProtoBuf definition.
186        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
187            match value {
188                "UNSPECIFIED" => Some(Self::Unspecified),
189                "SUCCEEDED" => Some(Self::Succeeded),
190                "DML_FAILED" => Some(Self::DmlFailed),
191                _ => None,
192            }
193        }
194    }
195}
196#[derive(prost_helpers::AnyPB)]
197#[derive(Clone, PartialEq, ::prost::Message)]
198pub struct ExecuteRequest {
199    #[prost(message, optional, tag = "1")]
200    pub task_id: ::core::option::Option<super::batch_plan::TaskId>,
201    #[prost(message, optional, tag = "2")]
202    pub plan: ::core::option::Option<super::batch_plan::PlanFragment>,
203    #[prost(map = "string, string", tag = "4")]
204    pub tracing_context: ::std::collections::HashMap<
205        ::prost::alloc::string::String,
206        ::prost::alloc::string::String,
207    >,
208    #[prost(message, optional, tag = "5")]
209    pub expr_context: ::core::option::Option<super::plan_common::ExprContext>,
210}
211#[derive(prost_helpers::AnyPB)]
212#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
213pub struct GetDataRequest {
214    #[prost(message, optional, tag = "1")]
215    pub task_output_id: ::core::option::Option<super::batch_plan::TaskOutputId>,
216}
217/// The structure for permit-based back-pressure.
218#[derive(prost_helpers::AnyPB)]
219#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
220pub struct Permits {
221    #[prost(oneof = "permits::Value", tags = "1, 2")]
222    pub value: ::core::option::Option<permits::Value>,
223}
224/// Nested message and enum types in `Permits`.
225pub mod permits {
226    #[derive(prost_helpers::AnyPB)]
227    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)]
228    pub enum Value {
229        /// The permits required for a chunk, i.e. the cardinality of the chunk.
230        #[prost(uint32, tag = "1")]
231        Record(u32),
232        /// The permits required for a barrier, typically 1.
233        #[prost(uint32, tag = "2")]
234        Barrier(u32),
235    }
236}
237#[derive(prost_helpers::AnyPB)]
238#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
239pub struct GetStreamRequest {
240    #[prost(oneof = "get_stream_request::Value", tags = "1, 2")]
241    pub value: ::core::option::Option<get_stream_request::Value>,
242}
243/// Nested message and enum types in `GetStreamRequest`.
244pub mod get_stream_request {
245    #[derive(prost_helpers::AnyPB)]
246    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
247    pub struct Get {
248        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
249        pub up_actor_id: crate::id::ActorId,
250        #[prost(uint32, tag = "2", wrapper = "crate::id::ActorId")]
251        pub down_actor_id: crate::id::ActorId,
252        #[prost(uint32, tag = "3", wrapper = "crate::id::FragmentId")]
253        pub up_fragment_id: crate::id::FragmentId,
254        #[prost(uint32, tag = "4", wrapper = "crate::id::FragmentId")]
255        pub down_fragment_id: crate::id::FragmentId,
256        #[prost(uint64, tag = "5", wrapper = "crate::id::PartialGraphId")]
257        pub up_partial_graph_id: crate::id::PartialGraphId,
258        #[prost(string, tag = "6")]
259        pub term_id: ::prost::alloc::string::String,
260    }
261    #[derive(prost_helpers::AnyPB)]
262    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
263    pub enum Value {
264        /// The first message, which tells the upstream which channel this exchange stream is for.
265        #[prost(message, tag = "1")]
266        Get(Get),
267        /// The following messages, which adds the permits back to the upstream to achieve back-pressure.
268        #[prost(message, tag = "2")]
269        AddPermits(super::Permits),
270    }
271}
272#[derive(prost_helpers::AnyPB)]
273#[derive(Clone, PartialEq, ::prost::Message)]
274pub struct GetStreamResponse {
275    #[prost(message, optional, tag = "1")]
276    pub message: ::core::option::Option<super::stream_plan::StreamMessageBatch>,
277    /// The number of permits acquired for this message, which should be sent back to the upstream with `add_permits`.
278    /// In theory, this can also be derived from the message itself by the receiver. Here we send it explicitly to
279    /// avoid any sense of inconsistency for the derivation logic, so the receiver can simply send it back verbatim.
280    #[prost(message, optional, tag = "2")]
281    pub permits: ::core::option::Option<Permits>,
282}
283/// Generated client implementations.
284pub mod task_service_client {
285    #![allow(
286        unused_variables,
287        dead_code,
288        missing_docs,
289        clippy::wildcard_imports,
290        clippy::let_unit_value,
291    )]
292    use tonic::codegen::*;
293    use tonic::codegen::http::Uri;
294    #[derive(Debug, Clone)]
295    pub struct TaskServiceClient<T> {
296        inner: tonic::client::Grpc<T>,
297    }
298    impl TaskServiceClient<tonic::transport::Channel> {
299        /// Attempt to create a new client by connecting to a given endpoint.
300        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
301        where
302            D: TryInto<tonic::transport::Endpoint>,
303            D::Error: Into<StdError>,
304        {
305            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
306            Ok(Self::new(conn))
307        }
308    }
309    impl<T> TaskServiceClient<T>
310    where
311        T: tonic::client::GrpcService<tonic::body::Body>,
312        T::Error: Into<StdError>,
313        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
314        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
315    {
316        pub fn new(inner: T) -> Self {
317            let inner = tonic::client::Grpc::new(inner);
318            Self { inner }
319        }
320        pub fn with_origin(inner: T, origin: Uri) -> Self {
321            let inner = tonic::client::Grpc::with_origin(inner, origin);
322            Self { inner }
323        }
324        pub fn with_interceptor<F>(
325            inner: T,
326            interceptor: F,
327        ) -> TaskServiceClient<InterceptedService<T, F>>
328        where
329            F: tonic::service::Interceptor,
330            T::ResponseBody: Default,
331            T: tonic::codegen::Service<
332                http::Request<tonic::body::Body>,
333                Response = http::Response<
334                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
335                >,
336            >,
337            <T as tonic::codegen::Service<
338                http::Request<tonic::body::Body>,
339            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
340        {
341            TaskServiceClient::new(InterceptedService::new(inner, interceptor))
342        }
343        /// Compress requests with the given encoding.
344        ///
345        /// This requires the server to support it otherwise it might respond with an
346        /// error.
347        #[must_use]
348        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
349            self.inner = self.inner.send_compressed(encoding);
350            self
351        }
352        /// Enable decompressing responses.
353        #[must_use]
354        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
355            self.inner = self.inner.accept_compressed(encoding);
356            self
357        }
358        /// Limits the maximum size of a decoded message.
359        ///
360        /// Default: `4MB`
361        #[must_use]
362        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
363            self.inner = self.inner.max_decoding_message_size(limit);
364            self
365        }
366        /// Limits the maximum size of an encoded message.
367        ///
368        /// Default: `usize::MAX`
369        #[must_use]
370        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
371            self.inner = self.inner.max_encoding_message_size(limit);
372            self
373        }
374        pub async fn create_task(
375            &mut self,
376            request: impl tonic::IntoRequest<super::CreateTaskRequest>,
377        ) -> std::result::Result<
378            tonic::Response<tonic::codec::Streaming<super::TaskInfoResponse>>,
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_prost::ProstCodec::default();
390            let path = http::uri::PathAndQuery::from_static(
391                "/task_service.TaskService/CreateTask",
392            );
393            let mut req = request.into_request();
394            req.extensions_mut()
395                .insert(GrpcMethod::new("task_service.TaskService", "CreateTask"));
396            self.inner.server_streaming(req, path, codec).await
397        }
398        /// Cancel an already-died (self execution-failure, previous aborted, completed) task will still succeed.
399        pub async fn cancel_task(
400            &mut self,
401            request: impl tonic::IntoRequest<super::CancelTaskRequest>,
402        ) -> std::result::Result<
403            tonic::Response<super::CancelTaskResponse>,
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_prost::ProstCodec::default();
415            let path = http::uri::PathAndQuery::from_static(
416                "/task_service.TaskService/CancelTask",
417            );
418            let mut req = request.into_request();
419            req.extensions_mut()
420                .insert(GrpcMethod::new("task_service.TaskService", "CancelTask"));
421            self.inner.unary(req, path, codec).await
422        }
423        pub async fn execute(
424            &mut self,
425            request: impl tonic::IntoRequest<super::ExecuteRequest>,
426        ) -> std::result::Result<
427            tonic::Response<tonic::codec::Streaming<super::GetDataResponse>>,
428            tonic::Status,
429        > {
430            self.inner
431                .ready()
432                .await
433                .map_err(|e| {
434                    tonic::Status::unknown(
435                        format!("Service was not ready: {}", e.into()),
436                    )
437                })?;
438            let codec = tonic_prost::ProstCodec::default();
439            let path = http::uri::PathAndQuery::from_static(
440                "/task_service.TaskService/Execute",
441            );
442            let mut req = request.into_request();
443            req.extensions_mut()
444                .insert(GrpcMethod::new("task_service.TaskService", "Execute"));
445            self.inner.server_streaming(req, path, codec).await
446        }
447        /// A lightweight version insert, only for non-pgwire insert, such as inserts from webhooks and websockets.
448        pub async fn fast_insert(
449            &mut self,
450            request: impl tonic::IntoRequest<super::FastInsertRequest>,
451        ) -> std::result::Result<
452            tonic::Response<super::FastInsertResponse>,
453            tonic::Status,
454        > {
455            self.inner
456                .ready()
457                .await
458                .map_err(|e| {
459                    tonic::Status::unknown(
460                        format!("Service was not ready: {}", e.into()),
461                    )
462                })?;
463            let codec = tonic_prost::ProstCodec::default();
464            let path = http::uri::PathAndQuery::from_static(
465                "/task_service.TaskService/FastInsert",
466            );
467            let mut req = request.into_request();
468            req.extensions_mut()
469                .insert(GrpcMethod::new("task_service.TaskService", "FastInsert"));
470            self.inner.unary(req, path, codec).await
471        }
472    }
473}
474/// Generated server implementations.
475pub mod task_service_server {
476    #![allow(
477        unused_variables,
478        dead_code,
479        missing_docs,
480        clippy::wildcard_imports,
481        clippy::let_unit_value,
482    )]
483    use tonic::codegen::*;
484    /// Generated trait containing gRPC methods that should be implemented for use with TaskServiceServer.
485    #[async_trait]
486    pub trait TaskService: std::marker::Send + std::marker::Sync + 'static {
487        /// Server streaming response type for the CreateTask method.
488        type CreateTaskStream: tonic::codegen::tokio_stream::Stream<
489                Item = std::result::Result<super::TaskInfoResponse, tonic::Status>,
490            >
491            + std::marker::Send
492            + 'static;
493        async fn create_task(
494            &self,
495            request: tonic::Request<super::CreateTaskRequest>,
496        ) -> std::result::Result<tonic::Response<Self::CreateTaskStream>, tonic::Status>;
497        /// Cancel an already-died (self execution-failure, previous aborted, completed) task will still succeed.
498        async fn cancel_task(
499            &self,
500            request: tonic::Request<super::CancelTaskRequest>,
501        ) -> std::result::Result<
502            tonic::Response<super::CancelTaskResponse>,
503            tonic::Status,
504        >;
505        /// Server streaming response type for the Execute method.
506        type ExecuteStream: tonic::codegen::tokio_stream::Stream<
507                Item = std::result::Result<super::GetDataResponse, tonic::Status>,
508            >
509            + std::marker::Send
510            + 'static;
511        async fn execute(
512            &self,
513            request: tonic::Request<super::ExecuteRequest>,
514        ) -> std::result::Result<tonic::Response<Self::ExecuteStream>, tonic::Status>;
515        /// A lightweight version insert, only for non-pgwire insert, such as inserts from webhooks and websockets.
516        async fn fast_insert(
517            &self,
518            request: tonic::Request<super::FastInsertRequest>,
519        ) -> std::result::Result<
520            tonic::Response<super::FastInsertResponse>,
521            tonic::Status,
522        >;
523    }
524    #[derive(Debug)]
525    pub struct TaskServiceServer<T> {
526        inner: Arc<T>,
527        accept_compression_encodings: EnabledCompressionEncodings,
528        send_compression_encodings: EnabledCompressionEncodings,
529        max_decoding_message_size: Option<usize>,
530        max_encoding_message_size: Option<usize>,
531    }
532    impl<T> TaskServiceServer<T> {
533        pub fn new(inner: T) -> Self {
534            Self::from_arc(Arc::new(inner))
535        }
536        pub fn from_arc(inner: Arc<T>) -> Self {
537            Self {
538                inner,
539                accept_compression_encodings: Default::default(),
540                send_compression_encodings: Default::default(),
541                max_decoding_message_size: None,
542                max_encoding_message_size: None,
543            }
544        }
545        pub fn with_interceptor<F>(
546            inner: T,
547            interceptor: F,
548        ) -> InterceptedService<Self, F>
549        where
550            F: tonic::service::Interceptor,
551        {
552            InterceptedService::new(Self::new(inner), interceptor)
553        }
554        /// Enable decompressing requests with the given encoding.
555        #[must_use]
556        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
557            self.accept_compression_encodings.enable(encoding);
558            self
559        }
560        /// Compress responses with the given encoding, if the client supports it.
561        #[must_use]
562        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
563            self.send_compression_encodings.enable(encoding);
564            self
565        }
566        /// Limits the maximum size of a decoded message.
567        ///
568        /// Default: `4MB`
569        #[must_use]
570        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
571            self.max_decoding_message_size = Some(limit);
572            self
573        }
574        /// Limits the maximum size of an encoded message.
575        ///
576        /// Default: `usize::MAX`
577        #[must_use]
578        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
579            self.max_encoding_message_size = Some(limit);
580            self
581        }
582    }
583    impl<T, B> tonic::codegen::Service<http::Request<B>> for TaskServiceServer<T>
584    where
585        T: TaskService,
586        B: Body + std::marker::Send + 'static,
587        B::Error: Into<StdError> + std::marker::Send + 'static,
588    {
589        type Response = http::Response<tonic::body::Body>;
590        type Error = std::convert::Infallible;
591        type Future = BoxFuture<Self::Response, Self::Error>;
592        fn poll_ready(
593            &mut self,
594            _cx: &mut Context<'_>,
595        ) -> Poll<std::result::Result<(), Self::Error>> {
596            Poll::Ready(Ok(()))
597        }
598        fn call(&mut self, req: http::Request<B>) -> Self::Future {
599            match req.uri().path() {
600                "/task_service.TaskService/CreateTask" => {
601                    #[allow(non_camel_case_types)]
602                    struct CreateTaskSvc<T: TaskService>(pub Arc<T>);
603                    impl<
604                        T: TaskService,
605                    > tonic::server::ServerStreamingService<super::CreateTaskRequest>
606                    for CreateTaskSvc<T> {
607                        type Response = super::TaskInfoResponse;
608                        type ResponseStream = T::CreateTaskStream;
609                        type Future = BoxFuture<
610                            tonic::Response<Self::ResponseStream>,
611                            tonic::Status,
612                        >;
613                        fn call(
614                            &mut self,
615                            request: tonic::Request<super::CreateTaskRequest>,
616                        ) -> Self::Future {
617                            let inner = Arc::clone(&self.0);
618                            let fut = async move {
619                                <T as TaskService>::create_task(&inner, request).await
620                            };
621                            Box::pin(fut)
622                        }
623                    }
624                    let accept_compression_encodings = self.accept_compression_encodings;
625                    let send_compression_encodings = self.send_compression_encodings;
626                    let max_decoding_message_size = self.max_decoding_message_size;
627                    let max_encoding_message_size = self.max_encoding_message_size;
628                    let inner = self.inner.clone();
629                    let fut = async move {
630                        let method = CreateTaskSvc(inner);
631                        let codec = tonic_prost::ProstCodec::default();
632                        let mut grpc = tonic::server::Grpc::new(codec)
633                            .apply_compression_config(
634                                accept_compression_encodings,
635                                send_compression_encodings,
636                            )
637                            .apply_max_message_size_config(
638                                max_decoding_message_size,
639                                max_encoding_message_size,
640                            );
641                        let res = grpc.server_streaming(method, req).await;
642                        Ok(res)
643                    };
644                    Box::pin(fut)
645                }
646                "/task_service.TaskService/CancelTask" => {
647                    #[allow(non_camel_case_types)]
648                    struct CancelTaskSvc<T: TaskService>(pub Arc<T>);
649                    impl<
650                        T: TaskService,
651                    > tonic::server::UnaryService<super::CancelTaskRequest>
652                    for CancelTaskSvc<T> {
653                        type Response = super::CancelTaskResponse;
654                        type Future = BoxFuture<
655                            tonic::Response<Self::Response>,
656                            tonic::Status,
657                        >;
658                        fn call(
659                            &mut self,
660                            request: tonic::Request<super::CancelTaskRequest>,
661                        ) -> Self::Future {
662                            let inner = Arc::clone(&self.0);
663                            let fut = async move {
664                                <T as TaskService>::cancel_task(&inner, request).await
665                            };
666                            Box::pin(fut)
667                        }
668                    }
669                    let accept_compression_encodings = self.accept_compression_encodings;
670                    let send_compression_encodings = self.send_compression_encodings;
671                    let max_decoding_message_size = self.max_decoding_message_size;
672                    let max_encoding_message_size = self.max_encoding_message_size;
673                    let inner = self.inner.clone();
674                    let fut = async move {
675                        let method = CancelTaskSvc(inner);
676                        let codec = tonic_prost::ProstCodec::default();
677                        let mut grpc = tonic::server::Grpc::new(codec)
678                            .apply_compression_config(
679                                accept_compression_encodings,
680                                send_compression_encodings,
681                            )
682                            .apply_max_message_size_config(
683                                max_decoding_message_size,
684                                max_encoding_message_size,
685                            );
686                        let res = grpc.unary(method, req).await;
687                        Ok(res)
688                    };
689                    Box::pin(fut)
690                }
691                "/task_service.TaskService/Execute" => {
692                    #[allow(non_camel_case_types)]
693                    struct ExecuteSvc<T: TaskService>(pub Arc<T>);
694                    impl<
695                        T: TaskService,
696                    > tonic::server::ServerStreamingService<super::ExecuteRequest>
697                    for ExecuteSvc<T> {
698                        type Response = super::GetDataResponse;
699                        type ResponseStream = T::ExecuteStream;
700                        type Future = BoxFuture<
701                            tonic::Response<Self::ResponseStream>,
702                            tonic::Status,
703                        >;
704                        fn call(
705                            &mut self,
706                            request: tonic::Request<super::ExecuteRequest>,
707                        ) -> Self::Future {
708                            let inner = Arc::clone(&self.0);
709                            let fut = async move {
710                                <T as TaskService>::execute(&inner, request).await
711                            };
712                            Box::pin(fut)
713                        }
714                    }
715                    let accept_compression_encodings = self.accept_compression_encodings;
716                    let send_compression_encodings = self.send_compression_encodings;
717                    let max_decoding_message_size = self.max_decoding_message_size;
718                    let max_encoding_message_size = self.max_encoding_message_size;
719                    let inner = self.inner.clone();
720                    let fut = async move {
721                        let method = ExecuteSvc(inner);
722                        let codec = tonic_prost::ProstCodec::default();
723                        let mut grpc = tonic::server::Grpc::new(codec)
724                            .apply_compression_config(
725                                accept_compression_encodings,
726                                send_compression_encodings,
727                            )
728                            .apply_max_message_size_config(
729                                max_decoding_message_size,
730                                max_encoding_message_size,
731                            );
732                        let res = grpc.server_streaming(method, req).await;
733                        Ok(res)
734                    };
735                    Box::pin(fut)
736                }
737                "/task_service.TaskService/FastInsert" => {
738                    #[allow(non_camel_case_types)]
739                    struct FastInsertSvc<T: TaskService>(pub Arc<T>);
740                    impl<
741                        T: TaskService,
742                    > tonic::server::UnaryService<super::FastInsertRequest>
743                    for FastInsertSvc<T> {
744                        type Response = super::FastInsertResponse;
745                        type Future = BoxFuture<
746                            tonic::Response<Self::Response>,
747                            tonic::Status,
748                        >;
749                        fn call(
750                            &mut self,
751                            request: tonic::Request<super::FastInsertRequest>,
752                        ) -> Self::Future {
753                            let inner = Arc::clone(&self.0);
754                            let fut = async move {
755                                <T as TaskService>::fast_insert(&inner, request).await
756                            };
757                            Box::pin(fut)
758                        }
759                    }
760                    let accept_compression_encodings = self.accept_compression_encodings;
761                    let send_compression_encodings = self.send_compression_encodings;
762                    let max_decoding_message_size = self.max_decoding_message_size;
763                    let max_encoding_message_size = self.max_encoding_message_size;
764                    let inner = self.inner.clone();
765                    let fut = async move {
766                        let method = FastInsertSvc(inner);
767                        let codec = tonic_prost::ProstCodec::default();
768                        let mut grpc = tonic::server::Grpc::new(codec)
769                            .apply_compression_config(
770                                accept_compression_encodings,
771                                send_compression_encodings,
772                            )
773                            .apply_max_message_size_config(
774                                max_decoding_message_size,
775                                max_encoding_message_size,
776                            );
777                        let res = grpc.unary(method, req).await;
778                        Ok(res)
779                    };
780                    Box::pin(fut)
781                }
782                _ => {
783                    Box::pin(async move {
784                        let mut response = http::Response::new(
785                            tonic::body::Body::default(),
786                        );
787                        let headers = response.headers_mut();
788                        headers
789                            .insert(
790                                tonic::Status::GRPC_STATUS,
791                                (tonic::Code::Unimplemented as i32).into(),
792                            );
793                        headers
794                            .insert(
795                                http::header::CONTENT_TYPE,
796                                tonic::metadata::GRPC_CONTENT_TYPE,
797                            );
798                        Ok(response)
799                    })
800                }
801            }
802        }
803    }
804    impl<T> Clone for TaskServiceServer<T> {
805        fn clone(&self) -> Self {
806            let inner = self.inner.clone();
807            Self {
808                inner,
809                accept_compression_encodings: self.accept_compression_encodings,
810                send_compression_encodings: self.send_compression_encodings,
811                max_decoding_message_size: self.max_decoding_message_size,
812                max_encoding_message_size: self.max_encoding_message_size,
813            }
814        }
815    }
816    /// Generated gRPC service name
817    pub const SERVICE_NAME: &str = "task_service.TaskService";
818    impl<T> tonic::server::NamedService for TaskServiceServer<T> {
819        const NAME: &'static str = SERVICE_NAME;
820    }
821}
822/// Generated client implementations.
823pub mod batch_exchange_service_client {
824    #![allow(
825        unused_variables,
826        dead_code,
827        missing_docs,
828        clippy::wildcard_imports,
829        clippy::let_unit_value,
830    )]
831    use tonic::codegen::*;
832    use tonic::codegen::http::Uri;
833    #[derive(Debug, Clone)]
834    pub struct BatchExchangeServiceClient<T> {
835        inner: tonic::client::Grpc<T>,
836    }
837    impl BatchExchangeServiceClient<tonic::transport::Channel> {
838        /// Attempt to create a new client by connecting to a given endpoint.
839        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
840        where
841            D: TryInto<tonic::transport::Endpoint>,
842            D::Error: Into<StdError>,
843        {
844            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
845            Ok(Self::new(conn))
846        }
847    }
848    impl<T> BatchExchangeServiceClient<T>
849    where
850        T: tonic::client::GrpcService<tonic::body::Body>,
851        T::Error: Into<StdError>,
852        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
853        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
854    {
855        pub fn new(inner: T) -> Self {
856            let inner = tonic::client::Grpc::new(inner);
857            Self { inner }
858        }
859        pub fn with_origin(inner: T, origin: Uri) -> Self {
860            let inner = tonic::client::Grpc::with_origin(inner, origin);
861            Self { inner }
862        }
863        pub fn with_interceptor<F>(
864            inner: T,
865            interceptor: F,
866        ) -> BatchExchangeServiceClient<InterceptedService<T, F>>
867        where
868            F: tonic::service::Interceptor,
869            T::ResponseBody: Default,
870            T: tonic::codegen::Service<
871                http::Request<tonic::body::Body>,
872                Response = http::Response<
873                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
874                >,
875            >,
876            <T as tonic::codegen::Service<
877                http::Request<tonic::body::Body>,
878            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
879        {
880            BatchExchangeServiceClient::new(InterceptedService::new(inner, interceptor))
881        }
882        /// Compress requests with the given encoding.
883        ///
884        /// This requires the server to support it otherwise it might respond with an
885        /// error.
886        #[must_use]
887        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
888            self.inner = self.inner.send_compressed(encoding);
889            self
890        }
891        /// Enable decompressing responses.
892        #[must_use]
893        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
894            self.inner = self.inner.accept_compressed(encoding);
895            self
896        }
897        /// Limits the maximum size of a decoded message.
898        ///
899        /// Default: `4MB`
900        #[must_use]
901        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
902            self.inner = self.inner.max_decoding_message_size(limit);
903            self
904        }
905        /// Limits the maximum size of an encoded message.
906        ///
907        /// Default: `usize::MAX`
908        #[must_use]
909        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
910            self.inner = self.inner.max_encoding_message_size(limit);
911            self
912        }
913        pub async fn get_data(
914            &mut self,
915            request: impl tonic::IntoRequest<super::GetDataRequest>,
916        ) -> std::result::Result<
917            tonic::Response<tonic::codec::Streaming<super::GetDataResponse>>,
918            tonic::Status,
919        > {
920            self.inner
921                .ready()
922                .await
923                .map_err(|e| {
924                    tonic::Status::unknown(
925                        format!("Service was not ready: {}", e.into()),
926                    )
927                })?;
928            let codec = tonic_prost::ProstCodec::default();
929            let path = http::uri::PathAndQuery::from_static(
930                "/task_service.BatchExchangeService/GetData",
931            );
932            let mut req = request.into_request();
933            req.extensions_mut()
934                .insert(GrpcMethod::new("task_service.BatchExchangeService", "GetData"));
935            self.inner.server_streaming(req, path, codec).await
936        }
937    }
938}
939/// Generated server implementations.
940pub mod batch_exchange_service_server {
941    #![allow(
942        unused_variables,
943        dead_code,
944        missing_docs,
945        clippy::wildcard_imports,
946        clippy::let_unit_value,
947    )]
948    use tonic::codegen::*;
949    /// Generated trait containing gRPC methods that should be implemented for use with BatchExchangeServiceServer.
950    #[async_trait]
951    pub trait BatchExchangeService: std::marker::Send + std::marker::Sync + 'static {
952        /// Server streaming response type for the GetData method.
953        type GetDataStream: tonic::codegen::tokio_stream::Stream<
954                Item = std::result::Result<super::GetDataResponse, tonic::Status>,
955            >
956            + std::marker::Send
957            + 'static;
958        async fn get_data(
959            &self,
960            request: tonic::Request<super::GetDataRequest>,
961        ) -> std::result::Result<tonic::Response<Self::GetDataStream>, tonic::Status>;
962    }
963    #[derive(Debug)]
964    pub struct BatchExchangeServiceServer<T> {
965        inner: Arc<T>,
966        accept_compression_encodings: EnabledCompressionEncodings,
967        send_compression_encodings: EnabledCompressionEncodings,
968        max_decoding_message_size: Option<usize>,
969        max_encoding_message_size: Option<usize>,
970    }
971    impl<T> BatchExchangeServiceServer<T> {
972        pub fn new(inner: T) -> Self {
973            Self::from_arc(Arc::new(inner))
974        }
975        pub fn from_arc(inner: Arc<T>) -> Self {
976            Self {
977                inner,
978                accept_compression_encodings: Default::default(),
979                send_compression_encodings: Default::default(),
980                max_decoding_message_size: None,
981                max_encoding_message_size: None,
982            }
983        }
984        pub fn with_interceptor<F>(
985            inner: T,
986            interceptor: F,
987        ) -> InterceptedService<Self, F>
988        where
989            F: tonic::service::Interceptor,
990        {
991            InterceptedService::new(Self::new(inner), interceptor)
992        }
993        /// Enable decompressing requests with the given encoding.
994        #[must_use]
995        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
996            self.accept_compression_encodings.enable(encoding);
997            self
998        }
999        /// Compress responses with the given encoding, if the client supports it.
1000        #[must_use]
1001        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1002            self.send_compression_encodings.enable(encoding);
1003            self
1004        }
1005        /// Limits the maximum size of a decoded message.
1006        ///
1007        /// Default: `4MB`
1008        #[must_use]
1009        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1010            self.max_decoding_message_size = Some(limit);
1011            self
1012        }
1013        /// Limits the maximum size of an encoded message.
1014        ///
1015        /// Default: `usize::MAX`
1016        #[must_use]
1017        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1018            self.max_encoding_message_size = Some(limit);
1019            self
1020        }
1021    }
1022    impl<T, B> tonic::codegen::Service<http::Request<B>>
1023    for BatchExchangeServiceServer<T>
1024    where
1025        T: BatchExchangeService,
1026        B: Body + std::marker::Send + 'static,
1027        B::Error: Into<StdError> + std::marker::Send + 'static,
1028    {
1029        type Response = http::Response<tonic::body::Body>;
1030        type Error = std::convert::Infallible;
1031        type Future = BoxFuture<Self::Response, Self::Error>;
1032        fn poll_ready(
1033            &mut self,
1034            _cx: &mut Context<'_>,
1035        ) -> Poll<std::result::Result<(), Self::Error>> {
1036            Poll::Ready(Ok(()))
1037        }
1038        fn call(&mut self, req: http::Request<B>) -> Self::Future {
1039            match req.uri().path() {
1040                "/task_service.BatchExchangeService/GetData" => {
1041                    #[allow(non_camel_case_types)]
1042                    struct GetDataSvc<T: BatchExchangeService>(pub Arc<T>);
1043                    impl<
1044                        T: BatchExchangeService,
1045                    > tonic::server::ServerStreamingService<super::GetDataRequest>
1046                    for GetDataSvc<T> {
1047                        type Response = super::GetDataResponse;
1048                        type ResponseStream = T::GetDataStream;
1049                        type Future = BoxFuture<
1050                            tonic::Response<Self::ResponseStream>,
1051                            tonic::Status,
1052                        >;
1053                        fn call(
1054                            &mut self,
1055                            request: tonic::Request<super::GetDataRequest>,
1056                        ) -> Self::Future {
1057                            let inner = Arc::clone(&self.0);
1058                            let fut = async move {
1059                                <T as BatchExchangeService>::get_data(&inner, request).await
1060                            };
1061                            Box::pin(fut)
1062                        }
1063                    }
1064                    let accept_compression_encodings = self.accept_compression_encodings;
1065                    let send_compression_encodings = self.send_compression_encodings;
1066                    let max_decoding_message_size = self.max_decoding_message_size;
1067                    let max_encoding_message_size = self.max_encoding_message_size;
1068                    let inner = self.inner.clone();
1069                    let fut = async move {
1070                        let method = GetDataSvc(inner);
1071                        let codec = tonic_prost::ProstCodec::default();
1072                        let mut grpc = tonic::server::Grpc::new(codec)
1073                            .apply_compression_config(
1074                                accept_compression_encodings,
1075                                send_compression_encodings,
1076                            )
1077                            .apply_max_message_size_config(
1078                                max_decoding_message_size,
1079                                max_encoding_message_size,
1080                            );
1081                        let res = grpc.server_streaming(method, req).await;
1082                        Ok(res)
1083                    };
1084                    Box::pin(fut)
1085                }
1086                _ => {
1087                    Box::pin(async move {
1088                        let mut response = http::Response::new(
1089                            tonic::body::Body::default(),
1090                        );
1091                        let headers = response.headers_mut();
1092                        headers
1093                            .insert(
1094                                tonic::Status::GRPC_STATUS,
1095                                (tonic::Code::Unimplemented as i32).into(),
1096                            );
1097                        headers
1098                            .insert(
1099                                http::header::CONTENT_TYPE,
1100                                tonic::metadata::GRPC_CONTENT_TYPE,
1101                            );
1102                        Ok(response)
1103                    })
1104                }
1105            }
1106        }
1107    }
1108    impl<T> Clone for BatchExchangeServiceServer<T> {
1109        fn clone(&self) -> Self {
1110            let inner = self.inner.clone();
1111            Self {
1112                inner,
1113                accept_compression_encodings: self.accept_compression_encodings,
1114                send_compression_encodings: self.send_compression_encodings,
1115                max_decoding_message_size: self.max_decoding_message_size,
1116                max_encoding_message_size: self.max_encoding_message_size,
1117            }
1118        }
1119    }
1120    /// Generated gRPC service name
1121    pub const SERVICE_NAME: &str = "task_service.BatchExchangeService";
1122    impl<T> tonic::server::NamedService for BatchExchangeServiceServer<T> {
1123        const NAME: &'static str = SERVICE_NAME;
1124    }
1125}
1126/// Generated client implementations.
1127pub mod stream_exchange_service_client {
1128    #![allow(
1129        unused_variables,
1130        dead_code,
1131        missing_docs,
1132        clippy::wildcard_imports,
1133        clippy::let_unit_value,
1134    )]
1135    use tonic::codegen::*;
1136    use tonic::codegen::http::Uri;
1137    #[derive(Debug, Clone)]
1138    pub struct StreamExchangeServiceClient<T> {
1139        inner: tonic::client::Grpc<T>,
1140    }
1141    impl StreamExchangeServiceClient<tonic::transport::Channel> {
1142        /// Attempt to create a new client by connecting to a given endpoint.
1143        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
1144        where
1145            D: TryInto<tonic::transport::Endpoint>,
1146            D::Error: Into<StdError>,
1147        {
1148            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
1149            Ok(Self::new(conn))
1150        }
1151    }
1152    impl<T> StreamExchangeServiceClient<T>
1153    where
1154        T: tonic::client::GrpcService<tonic::body::Body>,
1155        T::Error: Into<StdError>,
1156        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
1157        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
1158    {
1159        pub fn new(inner: T) -> Self {
1160            let inner = tonic::client::Grpc::new(inner);
1161            Self { inner }
1162        }
1163        pub fn with_origin(inner: T, origin: Uri) -> Self {
1164            let inner = tonic::client::Grpc::with_origin(inner, origin);
1165            Self { inner }
1166        }
1167        pub fn with_interceptor<F>(
1168            inner: T,
1169            interceptor: F,
1170        ) -> StreamExchangeServiceClient<InterceptedService<T, F>>
1171        where
1172            F: tonic::service::Interceptor,
1173            T::ResponseBody: Default,
1174            T: tonic::codegen::Service<
1175                http::Request<tonic::body::Body>,
1176                Response = http::Response<
1177                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
1178                >,
1179            >,
1180            <T as tonic::codegen::Service<
1181                http::Request<tonic::body::Body>,
1182            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
1183        {
1184            StreamExchangeServiceClient::new(InterceptedService::new(inner, interceptor))
1185        }
1186        /// Compress requests with the given encoding.
1187        ///
1188        /// This requires the server to support it otherwise it might respond with an
1189        /// error.
1190        #[must_use]
1191        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1192            self.inner = self.inner.send_compressed(encoding);
1193            self
1194        }
1195        /// Enable decompressing responses.
1196        #[must_use]
1197        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
1198            self.inner = self.inner.accept_compressed(encoding);
1199            self
1200        }
1201        /// Limits the maximum size of a decoded message.
1202        ///
1203        /// Default: `4MB`
1204        #[must_use]
1205        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1206            self.inner = self.inner.max_decoding_message_size(limit);
1207            self
1208        }
1209        /// Limits the maximum size of an encoded message.
1210        ///
1211        /// Default: `usize::MAX`
1212        #[must_use]
1213        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1214            self.inner = self.inner.max_encoding_message_size(limit);
1215            self
1216        }
1217        pub async fn get_stream(
1218            &mut self,
1219            request: impl tonic::IntoStreamingRequest<Message = super::GetStreamRequest>,
1220        ) -> std::result::Result<
1221            tonic::Response<tonic::codec::Streaming<super::GetStreamResponse>>,
1222            tonic::Status,
1223        > {
1224            self.inner
1225                .ready()
1226                .await
1227                .map_err(|e| {
1228                    tonic::Status::unknown(
1229                        format!("Service was not ready: {}", e.into()),
1230                    )
1231                })?;
1232            let codec = tonic_prost::ProstCodec::default();
1233            let path = http::uri::PathAndQuery::from_static(
1234                "/task_service.StreamExchangeService/GetStream",
1235            );
1236            let mut req = request.into_streaming_request();
1237            req.extensions_mut()
1238                .insert(
1239                    GrpcMethod::new("task_service.StreamExchangeService", "GetStream"),
1240                );
1241            self.inner.streaming(req, path, codec).await
1242        }
1243    }
1244}
1245/// Generated server implementations.
1246pub mod stream_exchange_service_server {
1247    #![allow(
1248        unused_variables,
1249        dead_code,
1250        missing_docs,
1251        clippy::wildcard_imports,
1252        clippy::let_unit_value,
1253    )]
1254    use tonic::codegen::*;
1255    /// Generated trait containing gRPC methods that should be implemented for use with StreamExchangeServiceServer.
1256    #[async_trait]
1257    pub trait StreamExchangeService: std::marker::Send + std::marker::Sync + 'static {
1258        /// Server streaming response type for the GetStream method.
1259        type GetStreamStream: tonic::codegen::tokio_stream::Stream<
1260                Item = std::result::Result<super::GetStreamResponse, tonic::Status>,
1261            >
1262            + std::marker::Send
1263            + 'static;
1264        async fn get_stream(
1265            &self,
1266            request: tonic::Request<tonic::Streaming<super::GetStreamRequest>>,
1267        ) -> std::result::Result<tonic::Response<Self::GetStreamStream>, tonic::Status>;
1268    }
1269    #[derive(Debug)]
1270    pub struct StreamExchangeServiceServer<T> {
1271        inner: Arc<T>,
1272        accept_compression_encodings: EnabledCompressionEncodings,
1273        send_compression_encodings: EnabledCompressionEncodings,
1274        max_decoding_message_size: Option<usize>,
1275        max_encoding_message_size: Option<usize>,
1276    }
1277    impl<T> StreamExchangeServiceServer<T> {
1278        pub fn new(inner: T) -> Self {
1279            Self::from_arc(Arc::new(inner))
1280        }
1281        pub fn from_arc(inner: Arc<T>) -> Self {
1282            Self {
1283                inner,
1284                accept_compression_encodings: Default::default(),
1285                send_compression_encodings: Default::default(),
1286                max_decoding_message_size: None,
1287                max_encoding_message_size: None,
1288            }
1289        }
1290        pub fn with_interceptor<F>(
1291            inner: T,
1292            interceptor: F,
1293        ) -> InterceptedService<Self, F>
1294        where
1295            F: tonic::service::Interceptor,
1296        {
1297            InterceptedService::new(Self::new(inner), interceptor)
1298        }
1299        /// Enable decompressing requests with the given encoding.
1300        #[must_use]
1301        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
1302            self.accept_compression_encodings.enable(encoding);
1303            self
1304        }
1305        /// Compress responses with the given encoding, if the client supports it.
1306        #[must_use]
1307        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1308            self.send_compression_encodings.enable(encoding);
1309            self
1310        }
1311        /// Limits the maximum size of a decoded message.
1312        ///
1313        /// Default: `4MB`
1314        #[must_use]
1315        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1316            self.max_decoding_message_size = Some(limit);
1317            self
1318        }
1319        /// Limits the maximum size of an encoded message.
1320        ///
1321        /// Default: `usize::MAX`
1322        #[must_use]
1323        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1324            self.max_encoding_message_size = Some(limit);
1325            self
1326        }
1327    }
1328    impl<T, B> tonic::codegen::Service<http::Request<B>>
1329    for StreamExchangeServiceServer<T>
1330    where
1331        T: StreamExchangeService,
1332        B: Body + std::marker::Send + 'static,
1333        B::Error: Into<StdError> + std::marker::Send + 'static,
1334    {
1335        type Response = http::Response<tonic::body::Body>;
1336        type Error = std::convert::Infallible;
1337        type Future = BoxFuture<Self::Response, Self::Error>;
1338        fn poll_ready(
1339            &mut self,
1340            _cx: &mut Context<'_>,
1341        ) -> Poll<std::result::Result<(), Self::Error>> {
1342            Poll::Ready(Ok(()))
1343        }
1344        fn call(&mut self, req: http::Request<B>) -> Self::Future {
1345            match req.uri().path() {
1346                "/task_service.StreamExchangeService/GetStream" => {
1347                    #[allow(non_camel_case_types)]
1348                    struct GetStreamSvc<T: StreamExchangeService>(pub Arc<T>);
1349                    impl<
1350                        T: StreamExchangeService,
1351                    > tonic::server::StreamingService<super::GetStreamRequest>
1352                    for GetStreamSvc<T> {
1353                        type Response = super::GetStreamResponse;
1354                        type ResponseStream = T::GetStreamStream;
1355                        type Future = BoxFuture<
1356                            tonic::Response<Self::ResponseStream>,
1357                            tonic::Status,
1358                        >;
1359                        fn call(
1360                            &mut self,
1361                            request: tonic::Request<
1362                                tonic::Streaming<super::GetStreamRequest>,
1363                            >,
1364                        ) -> Self::Future {
1365                            let inner = Arc::clone(&self.0);
1366                            let fut = async move {
1367                                <T as StreamExchangeService>::get_stream(&inner, request)
1368                                    .await
1369                            };
1370                            Box::pin(fut)
1371                        }
1372                    }
1373                    let accept_compression_encodings = self.accept_compression_encodings;
1374                    let send_compression_encodings = self.send_compression_encodings;
1375                    let max_decoding_message_size = self.max_decoding_message_size;
1376                    let max_encoding_message_size = self.max_encoding_message_size;
1377                    let inner = self.inner.clone();
1378                    let fut = async move {
1379                        let method = GetStreamSvc(inner);
1380                        let codec = tonic_prost::ProstCodec::default();
1381                        let mut grpc = tonic::server::Grpc::new(codec)
1382                            .apply_compression_config(
1383                                accept_compression_encodings,
1384                                send_compression_encodings,
1385                            )
1386                            .apply_max_message_size_config(
1387                                max_decoding_message_size,
1388                                max_encoding_message_size,
1389                            );
1390                        let res = grpc.streaming(method, req).await;
1391                        Ok(res)
1392                    };
1393                    Box::pin(fut)
1394                }
1395                _ => {
1396                    Box::pin(async move {
1397                        let mut response = http::Response::new(
1398                            tonic::body::Body::default(),
1399                        );
1400                        let headers = response.headers_mut();
1401                        headers
1402                            .insert(
1403                                tonic::Status::GRPC_STATUS,
1404                                (tonic::Code::Unimplemented as i32).into(),
1405                            );
1406                        headers
1407                            .insert(
1408                                http::header::CONTENT_TYPE,
1409                                tonic::metadata::GRPC_CONTENT_TYPE,
1410                            );
1411                        Ok(response)
1412                    })
1413                }
1414            }
1415        }
1416    }
1417    impl<T> Clone for StreamExchangeServiceServer<T> {
1418        fn clone(&self) -> Self {
1419            let inner = self.inner.clone();
1420            Self {
1421                inner,
1422                accept_compression_encodings: self.accept_compression_encodings,
1423                send_compression_encodings: self.send_compression_encodings,
1424                max_decoding_message_size: self.max_decoding_message_size,
1425                max_encoding_message_size: self.max_encoding_message_size,
1426            }
1427        }
1428    }
1429    /// Generated gRPC service name
1430    pub const SERVICE_NAME: &str = "task_service.StreamExchangeService";
1431    impl<T> tonic::server::NamedService for StreamExchangeServiceServer<T> {
1432        const NAME: &'static str = SERVICE_NAME;
1433    }
1434}
risingwave_pb/task_service.rs

risingwave_pb/
task_service.rs
