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, ::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, ::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, ::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, ::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, ::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, ::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, ::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, ::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, ::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, ::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, ::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(uint32, tag = "5", wrapper = "crate::id::DatabaseId")]
259        pub database_id: crate::id::DatabaseId,
260        #[prost(string, tag = "6")]
261        pub term_id: ::prost::alloc::string::String,
262    }
263    #[derive(prost_helpers::AnyPB)]
264    #[derive(Clone, PartialEq, ::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::BoxBody>,
314        T::Error: Into<StdError>,
315        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
316        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
317    {
318        pub fn new(inner: T) -> Self {
319            let inner = tonic::client::Grpc::new(inner);
320            Self { inner }
321        }
322        pub fn with_origin(inner: T, origin: Uri) -> Self {
323            let inner = tonic::client::Grpc::with_origin(inner, origin);
324            Self { inner }
325        }
326        pub fn with_interceptor<F>(
327            inner: T,
328            interceptor: F,
329        ) -> 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::BoxBody>,
335                Response = http::Response<
336                    <T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody,
337                >,
338            >,
339            <T as tonic::codegen::Service<
340                http::Request<tonic::body::BoxBody>,
341            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
342        {
343            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::codec::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::codec::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::codec::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::codec::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 client implementations.
477pub mod batch_exchange_service_client {
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    use tonic::codegen::http::Uri;
487    #[derive(Debug, Clone)]
488    pub struct BatchExchangeServiceClient<T> {
489        inner: tonic::client::Grpc<T>,
490    }
491    impl BatchExchangeServiceClient<tonic::transport::Channel> {
492        /// Attempt to create a new client by connecting to a given endpoint.
493        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
494        where
495            D: TryInto<tonic::transport::Endpoint>,
496            D::Error: Into<StdError>,
497        {
498            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
499            Ok(Self::new(conn))
500        }
501    }
502    impl<T> BatchExchangeServiceClient<T>
503    where
504        T: tonic::client::GrpcService<tonic::body::BoxBody>,
505        T::Error: Into<StdError>,
506        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
507        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
508    {
509        pub fn new(inner: T) -> Self {
510            let inner = tonic::client::Grpc::new(inner);
511            Self { inner }
512        }
513        pub fn with_origin(inner: T, origin: Uri) -> Self {
514            let inner = tonic::client::Grpc::with_origin(inner, origin);
515            Self { inner }
516        }
517        pub fn with_interceptor<F>(
518            inner: T,
519            interceptor: F,
520        ) -> BatchExchangeServiceClient<InterceptedService<T, F>>
521        where
522            F: tonic::service::Interceptor,
523            T::ResponseBody: Default,
524            T: tonic::codegen::Service<
525                http::Request<tonic::body::BoxBody>,
526                Response = http::Response<
527                    <T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody,
528                >,
529            >,
530            <T as tonic::codegen::Service<
531                http::Request<tonic::body::BoxBody>,
532            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
533        {
534            BatchExchangeServiceClient::new(InterceptedService::new(inner, interceptor))
535        }
536        /// Compress requests with the given encoding.
537        ///
538        /// This requires the server to support it otherwise it might respond with an
539        /// error.
540        #[must_use]
541        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
542            self.inner = self.inner.send_compressed(encoding);
543            self
544        }
545        /// Enable decompressing responses.
546        #[must_use]
547        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
548            self.inner = self.inner.accept_compressed(encoding);
549            self
550        }
551        /// Limits the maximum size of a decoded message.
552        ///
553        /// Default: `4MB`
554        #[must_use]
555        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
556            self.inner = self.inner.max_decoding_message_size(limit);
557            self
558        }
559        /// Limits the maximum size of an encoded message.
560        ///
561        /// Default: `usize::MAX`
562        #[must_use]
563        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
564            self.inner = self.inner.max_encoding_message_size(limit);
565            self
566        }
567        pub async fn get_data(
568            &mut self,
569            request: impl tonic::IntoRequest<super::GetDataRequest>,
570        ) -> std::result::Result<
571            tonic::Response<tonic::codec::Streaming<super::GetDataResponse>>,
572            tonic::Status,
573        > {
574            self.inner
575                .ready()
576                .await
577                .map_err(|e| {
578                    tonic::Status::unknown(
579                        format!("Service was not ready: {}", e.into()),
580                    )
581                })?;
582            let codec = tonic::codec::ProstCodec::default();
583            let path = http::uri::PathAndQuery::from_static(
584                "/task_service.BatchExchangeService/GetData",
585            );
586            let mut req = request.into_request();
587            req.extensions_mut()
588                .insert(GrpcMethod::new("task_service.BatchExchangeService", "GetData"));
589            self.inner.server_streaming(req, path, codec).await
590        }
591    }
592}
593/// Generated client implementations.
594pub mod stream_exchange_service_client {
595    #![allow(
596        unused_variables,
597        dead_code,
598        missing_docs,
599        clippy::wildcard_imports,
600        clippy::let_unit_value,
601    )]
602    use tonic::codegen::*;
603    use tonic::codegen::http::Uri;
604    #[derive(Debug, Clone)]
605    pub struct StreamExchangeServiceClient<T> {
606        inner: tonic::client::Grpc<T>,
607    }
608    impl StreamExchangeServiceClient<tonic::transport::Channel> {
609        /// Attempt to create a new client by connecting to a given endpoint.
610        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
611        where
612            D: TryInto<tonic::transport::Endpoint>,
613            D::Error: Into<StdError>,
614        {
615            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
616            Ok(Self::new(conn))
617        }
618    }
619    impl<T> StreamExchangeServiceClient<T>
620    where
621        T: tonic::client::GrpcService<tonic::body::BoxBody>,
622        T::Error: Into<StdError>,
623        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
624        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
625    {
626        pub fn new(inner: T) -> Self {
627            let inner = tonic::client::Grpc::new(inner);
628            Self { inner }
629        }
630        pub fn with_origin(inner: T, origin: Uri) -> Self {
631            let inner = tonic::client::Grpc::with_origin(inner, origin);
632            Self { inner }
633        }
634        pub fn with_interceptor<F>(
635            inner: T,
636            interceptor: F,
637        ) -> StreamExchangeServiceClient<InterceptedService<T, F>>
638        where
639            F: tonic::service::Interceptor,
640            T::ResponseBody: Default,
641            T: tonic::codegen::Service<
642                http::Request<tonic::body::BoxBody>,
643                Response = http::Response<
644                    <T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody,
645                >,
646            >,
647            <T as tonic::codegen::Service<
648                http::Request<tonic::body::BoxBody>,
649            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
650        {
651            StreamExchangeServiceClient::new(InterceptedService::new(inner, interceptor))
652        }
653        /// Compress requests with the given encoding.
654        ///
655        /// This requires the server to support it otherwise it might respond with an
656        /// error.
657        #[must_use]
658        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
659            self.inner = self.inner.send_compressed(encoding);
660            self
661        }
662        /// Enable decompressing responses.
663        #[must_use]
664        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
665            self.inner = self.inner.accept_compressed(encoding);
666            self
667        }
668        /// Limits the maximum size of a decoded message.
669        ///
670        /// Default: `4MB`
671        #[must_use]
672        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
673            self.inner = self.inner.max_decoding_message_size(limit);
674            self
675        }
676        /// Limits the maximum size of an encoded message.
677        ///
678        /// Default: `usize::MAX`
679        #[must_use]
680        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
681            self.inner = self.inner.max_encoding_message_size(limit);
682            self
683        }
684        pub async fn get_stream(
685            &mut self,
686            request: impl tonic::IntoStreamingRequest<Message = super::GetStreamRequest>,
687        ) -> std::result::Result<
688            tonic::Response<tonic::codec::Streaming<super::GetStreamResponse>>,
689            tonic::Status,
690        > {
691            self.inner
692                .ready()
693                .await
694                .map_err(|e| {
695                    tonic::Status::unknown(
696                        format!("Service was not ready: {}", e.into()),
697                    )
698                })?;
699            let codec = tonic::codec::ProstCodec::default();
700            let path = http::uri::PathAndQuery::from_static(
701                "/task_service.StreamExchangeService/GetStream",
702            );
703            let mut req = request.into_streaming_request();
704            req.extensions_mut()
705                .insert(
706                    GrpcMethod::new("task_service.StreamExchangeService", "GetStream"),
707                );
708            self.inner.streaming(req, path, codec).await
709        }
710    }
711}
712/// Generated server implementations.
713pub mod task_service_server {
714    #![allow(
715        unused_variables,
716        dead_code,
717        missing_docs,
718        clippy::wildcard_imports,
719        clippy::let_unit_value,
720    )]
721    use tonic::codegen::*;
722    /// Generated trait containing gRPC methods that should be implemented for use with TaskServiceServer.
723    #[async_trait]
724    pub trait TaskService: std::marker::Send + std::marker::Sync + 'static {
725        /// Server streaming response type for the CreateTask method.
726        type CreateTaskStream: tonic::codegen::tokio_stream::Stream<
727                Item = std::result::Result<super::TaskInfoResponse, tonic::Status>,
728            >
729            + std::marker::Send
730            + 'static;
731        async fn create_task(
732            &self,
733            request: tonic::Request<super::CreateTaskRequest>,
734        ) -> std::result::Result<tonic::Response<Self::CreateTaskStream>, tonic::Status>;
735        /// Cancel an already-died (self execution-failure, previous aborted, completed) task will still succeed.
736        async fn cancel_task(
737            &self,
738            request: tonic::Request<super::CancelTaskRequest>,
739        ) -> std::result::Result<
740            tonic::Response<super::CancelTaskResponse>,
741            tonic::Status,
742        >;
743        /// Server streaming response type for the Execute method.
744        type ExecuteStream: tonic::codegen::tokio_stream::Stream<
745                Item = std::result::Result<super::GetDataResponse, tonic::Status>,
746            >
747            + std::marker::Send
748            + 'static;
749        async fn execute(
750            &self,
751            request: tonic::Request<super::ExecuteRequest>,
752        ) -> std::result::Result<tonic::Response<Self::ExecuteStream>, tonic::Status>;
753        /// A lightweight version insert, only for non-pgwire insert, such as inserts from webhooks and websockets.
754        async fn fast_insert(
755            &self,
756            request: tonic::Request<super::FastInsertRequest>,
757        ) -> std::result::Result<
758            tonic::Response<super::FastInsertResponse>,
759            tonic::Status,
760        >;
761    }
762    #[derive(Debug)]
763    pub struct TaskServiceServer<T> {
764        inner: Arc<T>,
765        accept_compression_encodings: EnabledCompressionEncodings,
766        send_compression_encodings: EnabledCompressionEncodings,
767        max_decoding_message_size: Option<usize>,
768        max_encoding_message_size: Option<usize>,
769    }
770    impl<T> TaskServiceServer<T> {
771        pub fn new(inner: T) -> Self {
772            Self::from_arc(Arc::new(inner))
773        }
774        pub fn from_arc(inner: Arc<T>) -> Self {
775            Self {
776                inner,
777                accept_compression_encodings: Default::default(),
778                send_compression_encodings: Default::default(),
779                max_decoding_message_size: None,
780                max_encoding_message_size: None,
781            }
782        }
783        pub fn with_interceptor<F>(
784            inner: T,
785            interceptor: F,
786        ) -> InterceptedService<Self, F>
787        where
788            F: tonic::service::Interceptor,
789        {
790            InterceptedService::new(Self::new(inner), interceptor)
791        }
792        /// Enable decompressing requests with the given encoding.
793        #[must_use]
794        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
795            self.accept_compression_encodings.enable(encoding);
796            self
797        }
798        /// Compress responses with the given encoding, if the client supports it.
799        #[must_use]
800        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
801            self.send_compression_encodings.enable(encoding);
802            self
803        }
804        /// Limits the maximum size of a decoded message.
805        ///
806        /// Default: `4MB`
807        #[must_use]
808        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
809            self.max_decoding_message_size = Some(limit);
810            self
811        }
812        /// Limits the maximum size of an encoded message.
813        ///
814        /// Default: `usize::MAX`
815        #[must_use]
816        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
817            self.max_encoding_message_size = Some(limit);
818            self
819        }
820    }
821    impl<T, B> tonic::codegen::Service<http::Request<B>> for TaskServiceServer<T>
822    where
823        T: TaskService,
824        B: Body + std::marker::Send + 'static,
825        B::Error: Into<StdError> + std::marker::Send + 'static,
826    {
827        type Response = http::Response<tonic::body::BoxBody>;
828        type Error = std::convert::Infallible;
829        type Future = BoxFuture<Self::Response, Self::Error>;
830        fn poll_ready(
831            &mut self,
832            _cx: &mut Context<'_>,
833        ) -> Poll<std::result::Result<(), Self::Error>> {
834            Poll::Ready(Ok(()))
835        }
836        fn call(&mut self, req: http::Request<B>) -> Self::Future {
837            match req.uri().path() {
838                "/task_service.TaskService/CreateTask" => {
839                    #[allow(non_camel_case_types)]
840                    struct CreateTaskSvc<T: TaskService>(pub Arc<T>);
841                    impl<
842                        T: TaskService,
843                    > tonic::server::ServerStreamingService<super::CreateTaskRequest>
844                    for CreateTaskSvc<T> {
845                        type Response = super::TaskInfoResponse;
846                        type ResponseStream = T::CreateTaskStream;
847                        type Future = BoxFuture<
848                            tonic::Response<Self::ResponseStream>,
849                            tonic::Status,
850                        >;
851                        fn call(
852                            &mut self,
853                            request: tonic::Request<super::CreateTaskRequest>,
854                        ) -> Self::Future {
855                            let inner = Arc::clone(&self.0);
856                            let fut = async move {
857                                <T as TaskService>::create_task(&inner, request).await
858                            };
859                            Box::pin(fut)
860                        }
861                    }
862                    let accept_compression_encodings = self.accept_compression_encodings;
863                    let send_compression_encodings = self.send_compression_encodings;
864                    let max_decoding_message_size = self.max_decoding_message_size;
865                    let max_encoding_message_size = self.max_encoding_message_size;
866                    let inner = self.inner.clone();
867                    let fut = async move {
868                        let method = CreateTaskSvc(inner);
869                        let codec = tonic::codec::ProstCodec::default();
870                        let mut grpc = tonic::server::Grpc::new(codec)
871                            .apply_compression_config(
872                                accept_compression_encodings,
873                                send_compression_encodings,
874                            )
875                            .apply_max_message_size_config(
876                                max_decoding_message_size,
877                                max_encoding_message_size,
878                            );
879                        let res = grpc.server_streaming(method, req).await;
880                        Ok(res)
881                    };
882                    Box::pin(fut)
883                }
884                "/task_service.TaskService/CancelTask" => {
885                    #[allow(non_camel_case_types)]
886                    struct CancelTaskSvc<T: TaskService>(pub Arc<T>);
887                    impl<
888                        T: TaskService,
889                    > tonic::server::UnaryService<super::CancelTaskRequest>
890                    for CancelTaskSvc<T> {
891                        type Response = super::CancelTaskResponse;
892                        type Future = BoxFuture<
893                            tonic::Response<Self::Response>,
894                            tonic::Status,
895                        >;
896                        fn call(
897                            &mut self,
898                            request: tonic::Request<super::CancelTaskRequest>,
899                        ) -> Self::Future {
900                            let inner = Arc::clone(&self.0);
901                            let fut = async move {
902                                <T as TaskService>::cancel_task(&inner, request).await
903                            };
904                            Box::pin(fut)
905                        }
906                    }
907                    let accept_compression_encodings = self.accept_compression_encodings;
908                    let send_compression_encodings = self.send_compression_encodings;
909                    let max_decoding_message_size = self.max_decoding_message_size;
910                    let max_encoding_message_size = self.max_encoding_message_size;
911                    let inner = self.inner.clone();
912                    let fut = async move {
913                        let method = CancelTaskSvc(inner);
914                        let codec = tonic::codec::ProstCodec::default();
915                        let mut grpc = tonic::server::Grpc::new(codec)
916                            .apply_compression_config(
917                                accept_compression_encodings,
918                                send_compression_encodings,
919                            )
920                            .apply_max_message_size_config(
921                                max_decoding_message_size,
922                                max_encoding_message_size,
923                            );
924                        let res = grpc.unary(method, req).await;
925                        Ok(res)
926                    };
927                    Box::pin(fut)
928                }
929                "/task_service.TaskService/Execute" => {
930                    #[allow(non_camel_case_types)]
931                    struct ExecuteSvc<T: TaskService>(pub Arc<T>);
932                    impl<
933                        T: TaskService,
934                    > tonic::server::ServerStreamingService<super::ExecuteRequest>
935                    for ExecuteSvc<T> {
936                        type Response = super::GetDataResponse;
937                        type ResponseStream = T::ExecuteStream;
938                        type Future = BoxFuture<
939                            tonic::Response<Self::ResponseStream>,
940                            tonic::Status,
941                        >;
942                        fn call(
943                            &mut self,
944                            request: tonic::Request<super::ExecuteRequest>,
945                        ) -> Self::Future {
946                            let inner = Arc::clone(&self.0);
947                            let fut = async move {
948                                <T as TaskService>::execute(&inner, request).await
949                            };
950                            Box::pin(fut)
951                        }
952                    }
953                    let accept_compression_encodings = self.accept_compression_encodings;
954                    let send_compression_encodings = self.send_compression_encodings;
955                    let max_decoding_message_size = self.max_decoding_message_size;
956                    let max_encoding_message_size = self.max_encoding_message_size;
957                    let inner = self.inner.clone();
958                    let fut = async move {
959                        let method = ExecuteSvc(inner);
960                        let codec = tonic::codec::ProstCodec::default();
961                        let mut grpc = tonic::server::Grpc::new(codec)
962                            .apply_compression_config(
963                                accept_compression_encodings,
964                                send_compression_encodings,
965                            )
966                            .apply_max_message_size_config(
967                                max_decoding_message_size,
968                                max_encoding_message_size,
969                            );
970                        let res = grpc.server_streaming(method, req).await;
971                        Ok(res)
972                    };
973                    Box::pin(fut)
974                }
975                "/task_service.TaskService/FastInsert" => {
976                    #[allow(non_camel_case_types)]
977                    struct FastInsertSvc<T: TaskService>(pub Arc<T>);
978                    impl<
979                        T: TaskService,
980                    > tonic::server::UnaryService<super::FastInsertRequest>
981                    for FastInsertSvc<T> {
982                        type Response = super::FastInsertResponse;
983                        type Future = BoxFuture<
984                            tonic::Response<Self::Response>,
985                            tonic::Status,
986                        >;
987                        fn call(
988                            &mut self,
989                            request: tonic::Request<super::FastInsertRequest>,
990                        ) -> Self::Future {
991                            let inner = Arc::clone(&self.0);
992                            let fut = async move {
993                                <T as TaskService>::fast_insert(&inner, request).await
994                            };
995                            Box::pin(fut)
996                        }
997                    }
998                    let accept_compression_encodings = self.accept_compression_encodings;
999                    let send_compression_encodings = self.send_compression_encodings;
1000                    let max_decoding_message_size = self.max_decoding_message_size;
1001                    let max_encoding_message_size = self.max_encoding_message_size;
1002                    let inner = self.inner.clone();
1003                    let fut = async move {
1004                        let method = FastInsertSvc(inner);
1005                        let codec = tonic::codec::ProstCodec::default();
1006                        let mut grpc = tonic::server::Grpc::new(codec)
1007                            .apply_compression_config(
1008                                accept_compression_encodings,
1009                                send_compression_encodings,
1010                            )
1011                            .apply_max_message_size_config(
1012                                max_decoding_message_size,
1013                                max_encoding_message_size,
1014                            );
1015                        let res = grpc.unary(method, req).await;
1016                        Ok(res)
1017                    };
1018                    Box::pin(fut)
1019                }
1020                _ => {
1021                    Box::pin(async move {
1022                        let mut response = http::Response::new(empty_body());
1023                        let headers = response.headers_mut();
1024                        headers
1025                            .insert(
1026                                tonic::Status::GRPC_STATUS,
1027                                (tonic::Code::Unimplemented as i32).into(),
1028                            );
1029                        headers
1030                            .insert(
1031                                http::header::CONTENT_TYPE,
1032                                tonic::metadata::GRPC_CONTENT_TYPE,
1033                            );
1034                        Ok(response)
1035                    })
1036                }
1037            }
1038        }
1039    }
1040    impl<T> Clone for TaskServiceServer<T> {
1041        fn clone(&self) -> Self {
1042            let inner = self.inner.clone();
1043            Self {
1044                inner,
1045                accept_compression_encodings: self.accept_compression_encodings,
1046                send_compression_encodings: self.send_compression_encodings,
1047                max_decoding_message_size: self.max_decoding_message_size,
1048                max_encoding_message_size: self.max_encoding_message_size,
1049            }
1050        }
1051    }
1052    /// Generated gRPC service name
1053    pub const SERVICE_NAME: &str = "task_service.TaskService";
1054    impl<T> tonic::server::NamedService for TaskServiceServer<T> {
1055        const NAME: &'static str = SERVICE_NAME;
1056    }
1057}
1058/// Generated server implementations.
1059pub mod batch_exchange_service_server {
1060    #![allow(
1061        unused_variables,
1062        dead_code,
1063        missing_docs,
1064        clippy::wildcard_imports,
1065        clippy::let_unit_value,
1066    )]
1067    use tonic::codegen::*;
1068    /// Generated trait containing gRPC methods that should be implemented for use with BatchExchangeServiceServer.
1069    #[async_trait]
1070    pub trait BatchExchangeService: std::marker::Send + std::marker::Sync + 'static {
1071        /// Server streaming response type for the GetData method.
1072        type GetDataStream: tonic::codegen::tokio_stream::Stream<
1073                Item = std::result::Result<super::GetDataResponse, tonic::Status>,
1074            >
1075            + std::marker::Send
1076            + 'static;
1077        async fn get_data(
1078            &self,
1079            request: tonic::Request<super::GetDataRequest>,
1080        ) -> std::result::Result<tonic::Response<Self::GetDataStream>, tonic::Status>;
1081    }
1082    #[derive(Debug)]
1083    pub struct BatchExchangeServiceServer<T> {
1084        inner: Arc<T>,
1085        accept_compression_encodings: EnabledCompressionEncodings,
1086        send_compression_encodings: EnabledCompressionEncodings,
1087        max_decoding_message_size: Option<usize>,
1088        max_encoding_message_size: Option<usize>,
1089    }
1090    impl<T> BatchExchangeServiceServer<T> {
1091        pub fn new(inner: T) -> Self {
1092            Self::from_arc(Arc::new(inner))
1093        }
1094        pub fn from_arc(inner: Arc<T>) -> Self {
1095            Self {
1096                inner,
1097                accept_compression_encodings: Default::default(),
1098                send_compression_encodings: Default::default(),
1099                max_decoding_message_size: None,
1100                max_encoding_message_size: None,
1101            }
1102        }
1103        pub fn with_interceptor<F>(
1104            inner: T,
1105            interceptor: F,
1106        ) -> InterceptedService<Self, F>
1107        where
1108            F: tonic::service::Interceptor,
1109        {
1110            InterceptedService::new(Self::new(inner), interceptor)
1111        }
1112        /// Enable decompressing requests with the given encoding.
1113        #[must_use]
1114        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
1115            self.accept_compression_encodings.enable(encoding);
1116            self
1117        }
1118        /// Compress responses with the given encoding, if the client supports it.
1119        #[must_use]
1120        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1121            self.send_compression_encodings.enable(encoding);
1122            self
1123        }
1124        /// Limits the maximum size of a decoded message.
1125        ///
1126        /// Default: `4MB`
1127        #[must_use]
1128        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1129            self.max_decoding_message_size = Some(limit);
1130            self
1131        }
1132        /// Limits the maximum size of an encoded message.
1133        ///
1134        /// Default: `usize::MAX`
1135        #[must_use]
1136        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1137            self.max_encoding_message_size = Some(limit);
1138            self
1139        }
1140    }
1141    impl<T, B> tonic::codegen::Service<http::Request<B>>
1142    for BatchExchangeServiceServer<T>
1143    where
1144        T: BatchExchangeService,
1145        B: Body + std::marker::Send + 'static,
1146        B::Error: Into<StdError> + std::marker::Send + 'static,
1147    {
1148        type Response = http::Response<tonic::body::BoxBody>;
1149        type Error = std::convert::Infallible;
1150        type Future = BoxFuture<Self::Response, Self::Error>;
1151        fn poll_ready(
1152            &mut self,
1153            _cx: &mut Context<'_>,
1154        ) -> Poll<std::result::Result<(), Self::Error>> {
1155            Poll::Ready(Ok(()))
1156        }
1157        fn call(&mut self, req: http::Request<B>) -> Self::Future {
1158            match req.uri().path() {
1159                "/task_service.BatchExchangeService/GetData" => {
1160                    #[allow(non_camel_case_types)]
1161                    struct GetDataSvc<T: BatchExchangeService>(pub Arc<T>);
1162                    impl<
1163                        T: BatchExchangeService,
1164                    > tonic::server::ServerStreamingService<super::GetDataRequest>
1165                    for GetDataSvc<T> {
1166                        type Response = super::GetDataResponse;
1167                        type ResponseStream = T::GetDataStream;
1168                        type Future = BoxFuture<
1169                            tonic::Response<Self::ResponseStream>,
1170                            tonic::Status,
1171                        >;
1172                        fn call(
1173                            &mut self,
1174                            request: tonic::Request<super::GetDataRequest>,
1175                        ) -> Self::Future {
1176                            let inner = Arc::clone(&self.0);
1177                            let fut = async move {
1178                                <T as BatchExchangeService>::get_data(&inner, request).await
1179                            };
1180                            Box::pin(fut)
1181                        }
1182                    }
1183                    let accept_compression_encodings = self.accept_compression_encodings;
1184                    let send_compression_encodings = self.send_compression_encodings;
1185                    let max_decoding_message_size = self.max_decoding_message_size;
1186                    let max_encoding_message_size = self.max_encoding_message_size;
1187                    let inner = self.inner.clone();
1188                    let fut = async move {
1189                        let method = GetDataSvc(inner);
1190                        let codec = tonic::codec::ProstCodec::default();
1191                        let mut grpc = tonic::server::Grpc::new(codec)
1192                            .apply_compression_config(
1193                                accept_compression_encodings,
1194                                send_compression_encodings,
1195                            )
1196                            .apply_max_message_size_config(
1197                                max_decoding_message_size,
1198                                max_encoding_message_size,
1199                            );
1200                        let res = grpc.server_streaming(method, req).await;
1201                        Ok(res)
1202                    };
1203                    Box::pin(fut)
1204                }
1205                _ => {
1206                    Box::pin(async move {
1207                        let mut response = http::Response::new(empty_body());
1208                        let headers = response.headers_mut();
1209                        headers
1210                            .insert(
1211                                tonic::Status::GRPC_STATUS,
1212                                (tonic::Code::Unimplemented as i32).into(),
1213                            );
1214                        headers
1215                            .insert(
1216                                http::header::CONTENT_TYPE,
1217                                tonic::metadata::GRPC_CONTENT_TYPE,
1218                            );
1219                        Ok(response)
1220                    })
1221                }
1222            }
1223        }
1224    }
1225    impl<T> Clone for BatchExchangeServiceServer<T> {
1226        fn clone(&self) -> Self {
1227            let inner = self.inner.clone();
1228            Self {
1229                inner,
1230                accept_compression_encodings: self.accept_compression_encodings,
1231                send_compression_encodings: self.send_compression_encodings,
1232                max_decoding_message_size: self.max_decoding_message_size,
1233                max_encoding_message_size: self.max_encoding_message_size,
1234            }
1235        }
1236    }
1237    /// Generated gRPC service name
1238    pub const SERVICE_NAME: &str = "task_service.BatchExchangeService";
1239    impl<T> tonic::server::NamedService for BatchExchangeServiceServer<T> {
1240        const NAME: &'static str = SERVICE_NAME;
1241    }
1242}
1243/// Generated server implementations.
1244pub mod stream_exchange_service_server {
1245    #![allow(
1246        unused_variables,
1247        dead_code,
1248        missing_docs,
1249        clippy::wildcard_imports,
1250        clippy::let_unit_value,
1251    )]
1252    use tonic::codegen::*;
1253    /// Generated trait containing gRPC methods that should be implemented for use with StreamExchangeServiceServer.
1254    #[async_trait]
1255    pub trait StreamExchangeService: std::marker::Send + std::marker::Sync + 'static {
1256        /// Server streaming response type for the GetStream method.
1257        type GetStreamStream: tonic::codegen::tokio_stream::Stream<
1258                Item = std::result::Result<super::GetStreamResponse, tonic::Status>,
1259            >
1260            + std::marker::Send
1261            + 'static;
1262        async fn get_stream(
1263            &self,
1264            request: tonic::Request<tonic::Streaming<super::GetStreamRequest>>,
1265        ) -> std::result::Result<tonic::Response<Self::GetStreamStream>, tonic::Status>;
1266    }
1267    #[derive(Debug)]
1268    pub struct StreamExchangeServiceServer<T> {
1269        inner: Arc<T>,
1270        accept_compression_encodings: EnabledCompressionEncodings,
1271        send_compression_encodings: EnabledCompressionEncodings,
1272        max_decoding_message_size: Option<usize>,
1273        max_encoding_message_size: Option<usize>,
1274    }
1275    impl<T> StreamExchangeServiceServer<T> {
1276        pub fn new(inner: T) -> Self {
1277            Self::from_arc(Arc::new(inner))
1278        }
1279        pub fn from_arc(inner: Arc<T>) -> Self {
1280            Self {
1281                inner,
1282                accept_compression_encodings: Default::default(),
1283                send_compression_encodings: Default::default(),
1284                max_decoding_message_size: None,
1285                max_encoding_message_size: None,
1286            }
1287        }
1288        pub fn with_interceptor<F>(
1289            inner: T,
1290            interceptor: F,
1291        ) -> InterceptedService<Self, F>
1292        where
1293            F: tonic::service::Interceptor,
1294        {
1295            InterceptedService::new(Self::new(inner), interceptor)
1296        }
1297        /// Enable decompressing requests with the given encoding.
1298        #[must_use]
1299        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
1300            self.accept_compression_encodings.enable(encoding);
1301            self
1302        }
1303        /// Compress responses with the given encoding, if the client supports it.
1304        #[must_use]
1305        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1306            self.send_compression_encodings.enable(encoding);
1307            self
1308        }
1309        /// Limits the maximum size of a decoded message.
1310        ///
1311        /// Default: `4MB`
1312        #[must_use]
1313        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1314            self.max_decoding_message_size = Some(limit);
1315            self
1316        }
1317        /// Limits the maximum size of an encoded message.
1318        ///
1319        /// Default: `usize::MAX`
1320        #[must_use]
1321        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1322            self.max_encoding_message_size = Some(limit);
1323            self
1324        }
1325    }
1326    impl<T, B> tonic::codegen::Service<http::Request<B>>
1327    for StreamExchangeServiceServer<T>
1328    where
1329        T: StreamExchangeService,
1330        B: Body + std::marker::Send + 'static,
1331        B::Error: Into<StdError> + std::marker::Send + 'static,
1332    {
1333        type Response = http::Response<tonic::body::BoxBody>;
1334        type Error = std::convert::Infallible;
1335        type Future = BoxFuture<Self::Response, Self::Error>;
1336        fn poll_ready(
1337            &mut self,
1338            _cx: &mut Context<'_>,
1339        ) -> Poll<std::result::Result<(), Self::Error>> {
1340            Poll::Ready(Ok(()))
1341        }
1342        fn call(&mut self, req: http::Request<B>) -> Self::Future {
1343            match req.uri().path() {
1344                "/task_service.StreamExchangeService/GetStream" => {
1345                    #[allow(non_camel_case_types)]
1346                    struct GetStreamSvc<T: StreamExchangeService>(pub Arc<T>);
1347                    impl<
1348                        T: StreamExchangeService,
1349                    > tonic::server::StreamingService<super::GetStreamRequest>
1350                    for GetStreamSvc<T> {
1351                        type Response = super::GetStreamResponse;
1352                        type ResponseStream = T::GetStreamStream;
1353                        type Future = BoxFuture<
1354                            tonic::Response<Self::ResponseStream>,
1355                            tonic::Status,
1356                        >;
1357                        fn call(
1358                            &mut self,
1359                            request: tonic::Request<
1360                                tonic::Streaming<super::GetStreamRequest>,
1361                            >,
1362                        ) -> Self::Future {
1363                            let inner = Arc::clone(&self.0);
1364                            let fut = async move {
1365                                <T as StreamExchangeService>::get_stream(&inner, request)
1366                                    .await
1367                            };
1368                            Box::pin(fut)
1369                        }
1370                    }
1371                    let accept_compression_encodings = self.accept_compression_encodings;
1372                    let send_compression_encodings = self.send_compression_encodings;
1373                    let max_decoding_message_size = self.max_decoding_message_size;
1374                    let max_encoding_message_size = self.max_encoding_message_size;
1375                    let inner = self.inner.clone();
1376                    let fut = async move {
1377                        let method = GetStreamSvc(inner);
1378                        let codec = tonic::codec::ProstCodec::default();
1379                        let mut grpc = tonic::server::Grpc::new(codec)
1380                            .apply_compression_config(
1381                                accept_compression_encodings,
1382                                send_compression_encodings,
1383                            )
1384                            .apply_max_message_size_config(
1385                                max_decoding_message_size,
1386                                max_encoding_message_size,
1387                            );
1388                        let res = grpc.streaming(method, req).await;
1389                        Ok(res)
1390                    };
1391                    Box::pin(fut)
1392                }
1393                _ => {
1394                    Box::pin(async move {
1395                        let mut response = http::Response::new(empty_body());
1396                        let headers = response.headers_mut();
1397                        headers
1398                            .insert(
1399                                tonic::Status::GRPC_STATUS,
1400                                (tonic::Code::Unimplemented as i32).into(),
1401                            );
1402                        headers
1403                            .insert(
1404                                http::header::CONTENT_TYPE,
1405                                tonic::metadata::GRPC_CONTENT_TYPE,
1406                            );
1407                        Ok(response)
1408                    })
1409                }
1410            }
1411        }
1412    }
1413    impl<T> Clone for StreamExchangeServiceServer<T> {
1414        fn clone(&self) -> Self {
1415            let inner = self.inner.clone();
1416            Self {
1417                inner,
1418                accept_compression_encodings: self.accept_compression_encodings,
1419                send_compression_encodings: self.send_compression_encodings,
1420                max_decoding_message_size: self.max_decoding_message_size,
1421                max_encoding_message_size: self.max_encoding_message_size,
1422            }
1423        }
1424    }
1425    /// Generated gRPC service name
1426    pub const SERVICE_NAME: &str = "task_service.StreamExchangeService";
1427    impl<T> tonic::server::NamedService for StreamExchangeServiceServer<T> {
1428        const NAME: &'static str = SERVICE_NAME;
1429    }
1430}
risingwave_pb/task_service.rs

risingwave_pb/
task_service.rs
