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

risingwave_pb/
task_service.rs
