risingwave_pb/
connector_service.rs

1// This file is @generated by prost-build.
2#[derive(prost_helpers::AnyPB)]
3#[derive(Clone, PartialEq, ::prost::Message)]
4pub struct TableSchema {
5    #[prost(message, repeated, tag = "1")]
6    pub columns: ::prost::alloc::vec::Vec<super::plan_common::ColumnDesc>,
7    #[prost(uint32, repeated, tag = "2")]
8    pub pk_indices: ::prost::alloc::vec::Vec<u32>,
9}
10#[derive(prost_helpers::AnyPB)]
11#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
12pub struct ValidationError {
13    #[prost(string, tag = "1")]
14    pub error_message: ::prost::alloc::string::String,
15}
16#[derive(prost_helpers::AnyPB)]
17#[derive(Clone, PartialEq, ::prost::Message)]
18pub struct SinkParam {
19    #[prost(uint32, tag = "1", wrapper = "crate::id::SinkId")]
20    pub sink_id: crate::id::SinkId,
21    #[prost(btree_map = "string, string", tag = "2")]
22    pub properties: ::prost::alloc::collections::BTreeMap<
23        ::prost::alloc::string::String,
24        ::prost::alloc::string::String,
25    >,
26    #[prost(message, optional, tag = "3")]
27    pub table_schema: ::core::option::Option<TableSchema>,
28    /// to be deprecated
29    #[prost(enumeration = "super::catalog::SinkType", tag = "4")]
30    pub sink_type: i32,
31    #[prost(string, tag = "5")]
32    pub db_name: ::prost::alloc::string::String,
33    #[prost(string, tag = "6")]
34    pub sink_from_name: ::prost::alloc::string::String,
35    #[prost(message, optional, tag = "7")]
36    pub format_desc: ::core::option::Option<super::catalog::SinkFormatDesc>,
37    #[prost(string, tag = "8")]
38    pub sink_name: ::prost::alloc::string::String,
39    /// Backward-compatible replacement for `SINK_TYPE_FORCE_APPEND_ONLY`.
40    ///
41    /// Should not directly access this field. Use method `ignore_delete()` instead.
42    #[prost(bool, tag = "9")]
43    pub raw_ignore_delete: bool,
44}
45#[derive(prost_helpers::AnyPB)]
46#[derive(Clone, PartialEq, ::prost::Message)]
47pub struct SinkWriterStreamRequest {
48    #[prost(oneof = "sink_writer_stream_request::Request", tags = "1, 3, 4")]
49    pub request: ::core::option::Option<sink_writer_stream_request::Request>,
50}
51/// Nested message and enum types in `SinkWriterStreamRequest`.
52pub mod sink_writer_stream_request {
53    #[derive(prost_helpers::AnyPB)]
54    #[derive(Clone, PartialEq, ::prost::Message)]
55    pub struct StartSink {
56        #[prost(message, optional, tag = "1")]
57        pub sink_param: ::core::option::Option<super::SinkParam>,
58        #[prost(message, optional, tag = "3")]
59        pub payload_schema: ::core::option::Option<super::TableSchema>,
60    }
61    #[derive(prost_helpers::AnyPB)]
62    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
63    pub struct WriteBatch {
64        #[prost(uint64, tag = "3")]
65        pub batch_id: u64,
66        #[prost(uint64, tag = "4")]
67        pub epoch: u64,
68        #[prost(oneof = "write_batch::Payload", tags = "2, 5")]
69        pub payload: ::core::option::Option<write_batch::Payload>,
70    }
71    /// Nested message and enum types in `WriteBatch`.
72    pub mod write_batch {
73        #[derive(prost_helpers::AnyPB)]
74        #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
75        pub struct StreamChunkPayload {
76            #[prost(bytes = "vec", tag = "1")]
77            pub binary_data: ::prost::alloc::vec::Vec<u8>,
78        }
79        #[derive(prost_helpers::AnyPB)]
80        #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
81        pub enum Payload {
82            #[prost(message, tag = "2")]
83            StreamChunkPayload(StreamChunkPayload),
84            /// This is a reference pointer to a StreamChunk. The StreamChunk is owned
85            /// by the JniSinkWriterStreamRequest, which should handle the release of StreamChunk.
86            /// Index set to 5 because 3 and 4 have been occupied by `batch_id` and `epoch`
87            #[prost(int64, tag = "5")]
88            StreamChunkRefPointer(i64),
89        }
90    }
91    #[derive(prost_helpers::AnyPB)]
92    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
93    pub struct Barrier {
94        #[prost(uint64, tag = "1")]
95        pub epoch: u64,
96        #[prost(bool, tag = "2")]
97        pub is_checkpoint: bool,
98    }
99    #[derive(prost_helpers::AnyPB)]
100    #[derive(Clone, PartialEq, ::prost::Oneof)]
101    pub enum Request {
102        #[prost(message, tag = "1")]
103        Start(StartSink),
104        #[prost(message, tag = "3")]
105        WriteBatch(WriteBatch),
106        #[prost(message, tag = "4")]
107        Barrier(Barrier),
108    }
109}
110#[derive(prost_helpers::AnyPB)]
111#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
112pub struct SinkWriterStreamResponse {
113    #[prost(oneof = "sink_writer_stream_response::Response", tags = "1, 2, 3")]
114    pub response: ::core::option::Option<sink_writer_stream_response::Response>,
115}
116/// Nested message and enum types in `SinkWriterStreamResponse`.
117pub mod sink_writer_stream_response {
118    #[derive(prost_helpers::AnyPB)]
119    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
120    pub struct StartResponse {}
121    #[derive(prost_helpers::AnyPB)]
122    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
123    pub struct CommitResponse {
124        #[prost(uint64, tag = "1")]
125        pub epoch: u64,
126        #[prost(message, optional, tag = "2")]
127        pub metadata: ::core::option::Option<super::SinkMetadata>,
128    }
129    #[derive(prost_helpers::AnyPB)]
130    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
131    pub struct BatchWrittenResponse {
132        #[prost(uint64, tag = "1")]
133        pub epoch: u64,
134        #[prost(uint64, tag = "2")]
135        pub batch_id: u64,
136    }
137    #[derive(prost_helpers::AnyPB)]
138    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
139    pub enum Response {
140        #[prost(message, tag = "1")]
141        Start(StartResponse),
142        #[prost(message, tag = "2")]
143        Commit(CommitResponse),
144        #[prost(message, tag = "3")]
145        Batch(BatchWrittenResponse),
146    }
147}
148#[derive(prost_helpers::AnyPB)]
149#[derive(Clone, PartialEq, ::prost::Message)]
150pub struct ValidateSinkRequest {
151    #[prost(message, optional, tag = "1")]
152    pub sink_param: ::core::option::Option<SinkParam>,
153}
154#[derive(prost_helpers::AnyPB)]
155#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
156pub struct ValidateSinkResponse {
157    /// On validation failure, we return the error.
158    #[prost(message, optional, tag = "1")]
159    pub error: ::core::option::Option<ValidationError>,
160}
161#[derive(prost_helpers::AnyPB)]
162#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
163pub struct SinkMetadata {
164    #[prost(oneof = "sink_metadata::Metadata", tags = "1")]
165    pub metadata: ::core::option::Option<sink_metadata::Metadata>,
166}
167/// Nested message and enum types in `SinkMetadata`.
168pub mod sink_metadata {
169    #[derive(prost_helpers::AnyPB)]
170    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
171    pub struct SerializedMetadata {
172        #[prost(bytes = "vec", tag = "1")]
173        pub metadata: ::prost::alloc::vec::Vec<u8>,
174    }
175    #[derive(prost_helpers::AnyPB)]
176    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
177    pub enum Metadata {
178        #[prost(message, tag = "1")]
179        Serialized(SerializedMetadata),
180    }
181}
182#[derive(prost_helpers::AnyPB)]
183#[derive(Clone, PartialEq, ::prost::Message)]
184pub struct SinkCoordinatorStreamRequest {
185    #[prost(oneof = "sink_coordinator_stream_request::Request", tags = "1, 2")]
186    pub request: ::core::option::Option<sink_coordinator_stream_request::Request>,
187}
188/// Nested message and enum types in `SinkCoordinatorStreamRequest`.
189pub mod sink_coordinator_stream_request {
190    #[derive(prost_helpers::AnyPB)]
191    #[derive(Clone, PartialEq, ::prost::Message)]
192    pub struct StartCoordinator {
193        #[prost(message, optional, tag = "1")]
194        pub param: ::core::option::Option<super::SinkParam>,
195    }
196    #[derive(prost_helpers::AnyPB)]
197    #[derive(Clone, PartialEq, ::prost::Message)]
198    pub struct CommitMetadata {
199        #[prost(uint64, tag = "1")]
200        pub epoch: u64,
201        #[prost(message, repeated, tag = "2")]
202        pub metadata: ::prost::alloc::vec::Vec<super::SinkMetadata>,
203    }
204    #[derive(prost_helpers::AnyPB)]
205    #[derive(Clone, PartialEq, ::prost::Oneof)]
206    pub enum Request {
207        #[prost(message, tag = "1")]
208        Start(StartCoordinator),
209        #[prost(message, tag = "2")]
210        Commit(CommitMetadata),
211    }
212}
213#[derive(prost_helpers::AnyPB)]
214#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
215pub struct SinkCoordinatorStreamResponse {
216    #[prost(oneof = "sink_coordinator_stream_response::Response", tags = "1, 2")]
217    pub response: ::core::option::Option<sink_coordinator_stream_response::Response>,
218}
219/// Nested message and enum types in `SinkCoordinatorStreamResponse`.
220pub mod sink_coordinator_stream_response {
221    #[derive(prost_helpers::AnyPB)]
222    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
223    pub struct StartResponse {}
224    #[derive(prost_helpers::AnyPB)]
225    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
226    pub struct CommitResponse {
227        #[prost(uint64, tag = "1")]
228        pub epoch: u64,
229    }
230    #[derive(prost_helpers::AnyPB)]
231    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)]
232    pub enum Response {
233        #[prost(message, tag = "1")]
234        Start(StartResponse),
235        #[prost(message, tag = "2")]
236        Commit(CommitResponse),
237    }
238}
239#[derive(prost_helpers::AnyPB)]
240#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
241pub struct CdcMessage {
242    /// The value of the Debezium message
243    #[prost(string, tag = "1")]
244    pub payload: ::prost::alloc::string::String,
245    #[prost(string, tag = "2")]
246    pub partition: ::prost::alloc::string::String,
247    #[prost(string, tag = "3")]
248    pub offset: ::prost::alloc::string::String,
249    #[prost(string, tag = "4")]
250    pub full_table_name: ::prost::alloc::string::String,
251    #[prost(int64, tag = "5")]
252    pub source_ts_ms: i64,
253    #[prost(enumeration = "cdc_message::CdcMessageType", tag = "6")]
254    pub msg_type: i32,
255    /// The key of the Debezium message, which only used by `mongodb-cdc` connector.
256    #[prost(string, tag = "7")]
257    pub key: ::prost::alloc::string::String,
258    /// The CDC source type of this message. This is required to disambiguate the parsing logic for
259    /// `full_table_name` (e.g. `db.table` in MySQL vs `schema.table` in
260    /// Postgres/SQL Server/Oracle).
261    #[prost(enumeration = "SourceType", tag = "8")]
262    pub source_type: i32,
263}
264/// Nested message and enum types in `CdcMessage`.
265pub mod cdc_message {
266    #[derive(prost_helpers::AnyPB)]
267    #[derive(
268        Clone,
269        Copy,
270        Debug,
271        PartialEq,
272        Eq,
273        Hash,
274        PartialOrd,
275        Ord,
276        ::prost::Enumeration
277    )]
278    #[repr(i32)]
279    pub enum CdcMessageType {
280        Unspecified = 0,
281        Heartbeat = 1,
282        Data = 2,
283        TransactionMeta = 3,
284        SchemaChange = 4,
285    }
286    impl CdcMessageType {
287        /// String value of the enum field names used in the ProtoBuf definition.
288        ///
289        /// The values are not transformed in any way and thus are considered stable
290        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
291        pub fn as_str_name(&self) -> &'static str {
292            match self {
293                Self::Unspecified => "UNSPECIFIED",
294                Self::Heartbeat => "HEARTBEAT",
295                Self::Data => "DATA",
296                Self::TransactionMeta => "TRANSACTION_META",
297                Self::SchemaChange => "SCHEMA_CHANGE",
298            }
299        }
300        /// Creates an enum from field names used in the ProtoBuf definition.
301        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
302            match value {
303                "UNSPECIFIED" => Some(Self::Unspecified),
304                "HEARTBEAT" => Some(Self::Heartbeat),
305                "DATA" => Some(Self::Data),
306                "TRANSACTION_META" => Some(Self::TransactionMeta),
307                "SCHEMA_CHANGE" => Some(Self::SchemaChange),
308                _ => None,
309            }
310        }
311    }
312}
313#[derive(prost_helpers::AnyPB)]
314#[derive(Clone, PartialEq, ::prost::Message)]
315pub struct GetEventStreamRequest {
316    #[prost(uint64, tag = "1")]
317    pub source_id: u64,
318    #[prost(enumeration = "SourceType", tag = "2")]
319    pub source_type: i32,
320    #[prost(string, tag = "3")]
321    pub start_offset: ::prost::alloc::string::String,
322    #[prost(btree_map = "string, string", tag = "4")]
323    pub properties: ::prost::alloc::collections::BTreeMap<
324        ::prost::alloc::string::String,
325        ::prost::alloc::string::String,
326    >,
327    #[prost(bool, tag = "5")]
328    pub snapshot_done: bool,
329    #[prost(bool, tag = "6")]
330    pub is_source_job: bool,
331}
332#[derive(prost_helpers::AnyPB)]
333#[derive(Clone, PartialEq, ::prost::Message)]
334pub struct GetEventStreamResponse {
335    #[prost(uint64, tag = "1")]
336    pub source_id: u64,
337    #[prost(message, repeated, tag = "2")]
338    pub events: ::prost::alloc::vec::Vec<CdcMessage>,
339    #[prost(message, optional, tag = "3")]
340    pub control: ::core::option::Option<get_event_stream_response::ControlInfo>,
341}
342/// Nested message and enum types in `GetEventStreamResponse`.
343pub mod get_event_stream_response {
344    #[derive(prost_helpers::AnyPB)]
345    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
346    pub struct ControlInfo {
347        #[prost(bool, tag = "1")]
348        pub handshake_ok: bool,
349    }
350}
351#[derive(prost_helpers::AnyPB)]
352#[derive(Clone, PartialEq, ::prost::Message)]
353pub struct ValidateSourceRequest {
354    #[prost(uint64, tag = "1")]
355    pub source_id: u64,
356    #[prost(enumeration = "SourceType", tag = "2")]
357    pub source_type: i32,
358    #[prost(btree_map = "string, string", tag = "3")]
359    pub properties: ::prost::alloc::collections::BTreeMap<
360        ::prost::alloc::string::String,
361        ::prost::alloc::string::String,
362    >,
363    #[prost(message, optional, tag = "4")]
364    pub table_schema: ::core::option::Option<TableSchema>,
365    #[prost(bool, tag = "5")]
366    pub is_source_job: bool,
367    #[prost(bool, tag = "6")]
368    pub is_backfill_table: bool,
369}
370#[derive(prost_helpers::AnyPB)]
371#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
372pub struct ValidateSourceResponse {
373    /// On validation failure, we return the error.
374    #[prost(message, optional, tag = "1")]
375    pub error: ::core::option::Option<ValidationError>,
376}
377#[derive(prost_helpers::AnyPB)]
378#[derive(Clone, PartialEq, ::prost::Message)]
379pub struct CoordinateRequest {
380    #[prost(oneof = "coordinate_request::Msg", tags = "1, 2, 3, 4, 5")]
381    pub msg: ::core::option::Option<coordinate_request::Msg>,
382}
383/// Nested message and enum types in `CoordinateRequest`.
384pub mod coordinate_request {
385    /// The first request that starts a coordination between sink writer and coordinator.
386    /// The service will respond after sink writers of all vnodes have sent the request.
387    #[derive(prost_helpers::AnyPB)]
388    #[derive(Clone, PartialEq, ::prost::Message)]
389    pub struct StartCoordinationRequest {
390        #[prost(message, optional, tag = "1")]
391        pub vnode_bitmap: ::core::option::Option<super::super::common::Buffer>,
392        #[prost(message, optional, tag = "2")]
393        pub param: ::core::option::Option<super::SinkParam>,
394    }
395    #[derive(prost_helpers::AnyPB)]
396    #[derive(Clone, PartialEq, ::prost::Message)]
397    pub struct CommitRequest {
398        #[prost(uint64, tag = "1")]
399        pub epoch: u64,
400        #[prost(message, optional, tag = "2")]
401        pub metadata: ::core::option::Option<super::SinkMetadata>,
402        #[prost(message, optional, tag = "3")]
403        pub schema_change: ::core::option::Option<
404            super::super::stream_plan::SinkSchemaChange,
405        >,
406    }
407    #[derive(prost_helpers::AnyPB)]
408    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
409    pub struct UpdateVnodeBitmapRequest {
410        #[prost(message, optional, tag = "1")]
411        pub vnode_bitmap: ::core::option::Option<super::super::common::Buffer>,
412    }
413    #[derive(prost_helpers::AnyPB)]
414    #[derive(Clone, PartialEq, ::prost::Oneof)]
415    pub enum Msg {
416        #[prost(message, tag = "1")]
417        StartRequest(StartCoordinationRequest),
418        #[prost(message, tag = "2")]
419        CommitRequest(CommitRequest),
420        #[prost(message, tag = "3")]
421        UpdateVnodeRequest(UpdateVnodeBitmapRequest),
422        #[prost(bool, tag = "4")]
423        Stop(bool),
424        #[prost(uint64, tag = "5")]
425        AlignInitialEpochRequest(u64),
426    }
427}
428#[derive(prost_helpers::AnyPB)]
429#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
430pub struct CoordinateResponse {
431    #[prost(oneof = "coordinate_response::Msg", tags = "1, 2, 3, 4")]
432    pub msg: ::core::option::Option<coordinate_response::Msg>,
433}
434/// Nested message and enum types in `CoordinateResponse`.
435pub mod coordinate_response {
436    #[derive(prost_helpers::AnyPB)]
437    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
438    pub struct StartCoordinationResponse {
439        #[prost(uint64, optional, tag = "1")]
440        pub log_store_rewind_start_epoch: ::core::option::Option<u64>,
441    }
442    #[derive(prost_helpers::AnyPB)]
443    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
444    pub struct CommitResponse {
445        #[prost(uint64, tag = "1")]
446        pub epoch: u64,
447    }
448    #[derive(prost_helpers::AnyPB)]
449    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Oneof)]
450    pub enum Msg {
451        #[prost(message, tag = "1")]
452        StartResponse(StartCoordinationResponse),
453        #[prost(message, tag = "2")]
454        CommitResponse(CommitResponse),
455        #[prost(bool, tag = "3")]
456        Stopped(bool),
457        #[prost(uint64, tag = "4")]
458        AlignInitialEpochResponse(u64),
459    }
460}
461#[derive(prost_helpers::AnyPB)]
462#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
463pub struct IcebergPkIndexPreCommitState {
464    #[prost(bytes = "vec", tag = "1")]
465    pub agg_result: ::prost::alloc::vec::Vec<u8>,
466    #[prost(int64, tag = "2")]
467    pub snapshot_id: i64,
468}
469#[derive(prost_helpers::AnyPB)]
470#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
471#[repr(i32)]
472pub enum SourceType {
473    Unspecified = 0,
474    Mysql = 1,
475    Postgres = 2,
476    Mongodb = 4,
477    SqlServer = 5,
478    Oracle = 6,
479}
480impl SourceType {
481    /// String value of the enum field names used in the ProtoBuf definition.
482    ///
483    /// The values are not transformed in any way and thus are considered stable
484    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
485    pub fn as_str_name(&self) -> &'static str {
486        match self {
487            Self::Unspecified => "UNSPECIFIED",
488            Self::Mysql => "MYSQL",
489            Self::Postgres => "POSTGRES",
490            Self::Mongodb => "MONGODB",
491            Self::SqlServer => "SQL_SERVER",
492            Self::Oracle => "ORACLE",
493        }
494    }
495    /// Creates an enum from field names used in the ProtoBuf definition.
496    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
497        match value {
498            "UNSPECIFIED" => Some(Self::Unspecified),
499            "MYSQL" => Some(Self::Mysql),
500            "POSTGRES" => Some(Self::Postgres),
501            "MONGODB" => Some(Self::Mongodb),
502            "SQL_SERVER" => Some(Self::SqlServer),
503            "ORACLE" => Some(Self::Oracle),
504            _ => None,
505        }
506    }
507}
508/// Generated client implementations.
509pub mod connector_service_client {
510    #![allow(
511        unused_variables,
512        dead_code,
513        missing_docs,
514        clippy::wildcard_imports,
515        clippy::let_unit_value,
516    )]
517    use tonic::codegen::*;
518    use tonic::codegen::http::Uri;
519    #[derive(Debug, Clone)]
520    pub struct ConnectorServiceClient<T> {
521        inner: tonic::client::Grpc<T>,
522    }
523    impl ConnectorServiceClient<tonic::transport::Channel> {
524        /// Attempt to create a new client by connecting to a given endpoint.
525        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
526        where
527            D: TryInto<tonic::transport::Endpoint>,
528            D::Error: Into<StdError>,
529        {
530            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
531            Ok(Self::new(conn))
532        }
533    }
534    impl<T> ConnectorServiceClient<T>
535    where
536        T: tonic::client::GrpcService<tonic::body::Body>,
537        T::Error: Into<StdError>,
538        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
539        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
540    {
541        pub fn new(inner: T) -> Self {
542            let inner = tonic::client::Grpc::new(inner);
543            Self { inner }
544        }
545        pub fn with_origin(inner: T, origin: Uri) -> Self {
546            let inner = tonic::client::Grpc::with_origin(inner, origin);
547            Self { inner }
548        }
549        pub fn with_interceptor<F>(
550            inner: T,
551            interceptor: F,
552        ) -> ConnectorServiceClient<InterceptedService<T, F>>
553        where
554            F: tonic::service::Interceptor,
555            T::ResponseBody: Default,
556            T: tonic::codegen::Service<
557                http::Request<tonic::body::Body>,
558                Response = http::Response<
559                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
560                >,
561            >,
562            <T as tonic::codegen::Service<
563                http::Request<tonic::body::Body>,
564            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
565        {
566            ConnectorServiceClient::new(InterceptedService::new(inner, interceptor))
567        }
568        /// Compress requests with the given encoding.
569        ///
570        /// This requires the server to support it otherwise it might respond with an
571        /// error.
572        #[must_use]
573        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
574            self.inner = self.inner.send_compressed(encoding);
575            self
576        }
577        /// Enable decompressing responses.
578        #[must_use]
579        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
580            self.inner = self.inner.accept_compressed(encoding);
581            self
582        }
583        /// Limits the maximum size of a decoded message.
584        ///
585        /// Default: `4MB`
586        #[must_use]
587        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
588            self.inner = self.inner.max_decoding_message_size(limit);
589            self
590        }
591        /// Limits the maximum size of an encoded message.
592        ///
593        /// Default: `usize::MAX`
594        #[must_use]
595        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
596            self.inner = self.inner.max_encoding_message_size(limit);
597            self
598        }
599        pub async fn sink_writer_stream(
600            &mut self,
601            request: impl tonic::IntoStreamingRequest<
602                Message = super::SinkWriterStreamRequest,
603            >,
604        ) -> std::result::Result<
605            tonic::Response<tonic::codec::Streaming<super::SinkWriterStreamResponse>>,
606            tonic::Status,
607        > {
608            self.inner
609                .ready()
610                .await
611                .map_err(|e| {
612                    tonic::Status::unknown(
613                        format!("Service was not ready: {}", e.into()),
614                    )
615                })?;
616            let codec = tonic_prost::ProstCodec::default();
617            let path = http::uri::PathAndQuery::from_static(
618                "/connector_service.ConnectorService/SinkWriterStream",
619            );
620            let mut req = request.into_streaming_request();
621            req.extensions_mut()
622                .insert(
623                    GrpcMethod::new(
624                        "connector_service.ConnectorService",
625                        "SinkWriterStream",
626                    ),
627                );
628            self.inner.streaming(req, path, codec).await
629        }
630        pub async fn sink_coordinator_stream(
631            &mut self,
632            request: impl tonic::IntoStreamingRequest<
633                Message = super::SinkCoordinatorStreamRequest,
634            >,
635        ) -> std::result::Result<
636            tonic::Response<
637                tonic::codec::Streaming<super::SinkCoordinatorStreamResponse>,
638            >,
639            tonic::Status,
640        > {
641            self.inner
642                .ready()
643                .await
644                .map_err(|e| {
645                    tonic::Status::unknown(
646                        format!("Service was not ready: {}", e.into()),
647                    )
648                })?;
649            let codec = tonic_prost::ProstCodec::default();
650            let path = http::uri::PathAndQuery::from_static(
651                "/connector_service.ConnectorService/SinkCoordinatorStream",
652            );
653            let mut req = request.into_streaming_request();
654            req.extensions_mut()
655                .insert(
656                    GrpcMethod::new(
657                        "connector_service.ConnectorService",
658                        "SinkCoordinatorStream",
659                    ),
660                );
661            self.inner.streaming(req, path, codec).await
662        }
663        pub async fn validate_sink(
664            &mut self,
665            request: impl tonic::IntoRequest<super::ValidateSinkRequest>,
666        ) -> std::result::Result<
667            tonic::Response<super::ValidateSinkResponse>,
668            tonic::Status,
669        > {
670            self.inner
671                .ready()
672                .await
673                .map_err(|e| {
674                    tonic::Status::unknown(
675                        format!("Service was not ready: {}", e.into()),
676                    )
677                })?;
678            let codec = tonic_prost::ProstCodec::default();
679            let path = http::uri::PathAndQuery::from_static(
680                "/connector_service.ConnectorService/ValidateSink",
681            );
682            let mut req = request.into_request();
683            req.extensions_mut()
684                .insert(
685                    GrpcMethod::new("connector_service.ConnectorService", "ValidateSink"),
686                );
687            self.inner.unary(req, path, codec).await
688        }
689        pub async fn get_event_stream(
690            &mut self,
691            request: impl tonic::IntoRequest<super::GetEventStreamRequest>,
692        ) -> std::result::Result<
693            tonic::Response<tonic::codec::Streaming<super::GetEventStreamResponse>>,
694            tonic::Status,
695        > {
696            self.inner
697                .ready()
698                .await
699                .map_err(|e| {
700                    tonic::Status::unknown(
701                        format!("Service was not ready: {}", e.into()),
702                    )
703                })?;
704            let codec = tonic_prost::ProstCodec::default();
705            let path = http::uri::PathAndQuery::from_static(
706                "/connector_service.ConnectorService/GetEventStream",
707            );
708            let mut req = request.into_request();
709            req.extensions_mut()
710                .insert(
711                    GrpcMethod::new(
712                        "connector_service.ConnectorService",
713                        "GetEventStream",
714                    ),
715                );
716            self.inner.server_streaming(req, path, codec).await
717        }
718        pub async fn validate_source(
719            &mut self,
720            request: impl tonic::IntoRequest<super::ValidateSourceRequest>,
721        ) -> std::result::Result<
722            tonic::Response<super::ValidateSourceResponse>,
723            tonic::Status,
724        > {
725            self.inner
726                .ready()
727                .await
728                .map_err(|e| {
729                    tonic::Status::unknown(
730                        format!("Service was not ready: {}", e.into()),
731                    )
732                })?;
733            let codec = tonic_prost::ProstCodec::default();
734            let path = http::uri::PathAndQuery::from_static(
735                "/connector_service.ConnectorService/ValidateSource",
736            );
737            let mut req = request.into_request();
738            req.extensions_mut()
739                .insert(
740                    GrpcMethod::new(
741                        "connector_service.ConnectorService",
742                        "ValidateSource",
743                    ),
744                );
745            self.inner.unary(req, path, codec).await
746        }
747    }
748}
749/// Generated server implementations.
750pub mod connector_service_server {
751    #![allow(
752        unused_variables,
753        dead_code,
754        missing_docs,
755        clippy::wildcard_imports,
756        clippy::let_unit_value,
757    )]
758    use tonic::codegen::*;
759    /// Generated trait containing gRPC methods that should be implemented for use with ConnectorServiceServer.
760    #[async_trait]
761    pub trait ConnectorService: std::marker::Send + std::marker::Sync + 'static {
762        /// Server streaming response type for the SinkWriterStream method.
763        type SinkWriterStreamStream: tonic::codegen::tokio_stream::Stream<
764                Item = std::result::Result<
765                    super::SinkWriterStreamResponse,
766                    tonic::Status,
767                >,
768            >
769            + std::marker::Send
770            + 'static;
771        async fn sink_writer_stream(
772            &self,
773            request: tonic::Request<tonic::Streaming<super::SinkWriterStreamRequest>>,
774        ) -> std::result::Result<
775            tonic::Response<Self::SinkWriterStreamStream>,
776            tonic::Status,
777        >;
778        /// Server streaming response type for the SinkCoordinatorStream method.
779        type SinkCoordinatorStreamStream: tonic::codegen::tokio_stream::Stream<
780                Item = std::result::Result<
781                    super::SinkCoordinatorStreamResponse,
782                    tonic::Status,
783                >,
784            >
785            + std::marker::Send
786            + 'static;
787        async fn sink_coordinator_stream(
788            &self,
789            request: tonic::Request<
790                tonic::Streaming<super::SinkCoordinatorStreamRequest>,
791            >,
792        ) -> std::result::Result<
793            tonic::Response<Self::SinkCoordinatorStreamStream>,
794            tonic::Status,
795        >;
796        async fn validate_sink(
797            &self,
798            request: tonic::Request<super::ValidateSinkRequest>,
799        ) -> std::result::Result<
800            tonic::Response<super::ValidateSinkResponse>,
801            tonic::Status,
802        >;
803        /// Server streaming response type for the GetEventStream method.
804        type GetEventStreamStream: tonic::codegen::tokio_stream::Stream<
805                Item = std::result::Result<super::GetEventStreamResponse, tonic::Status>,
806            >
807            + std::marker::Send
808            + 'static;
809        async fn get_event_stream(
810            &self,
811            request: tonic::Request<super::GetEventStreamRequest>,
812        ) -> std::result::Result<
813            tonic::Response<Self::GetEventStreamStream>,
814            tonic::Status,
815        >;
816        async fn validate_source(
817            &self,
818            request: tonic::Request<super::ValidateSourceRequest>,
819        ) -> std::result::Result<
820            tonic::Response<super::ValidateSourceResponse>,
821            tonic::Status,
822        >;
823    }
824    #[derive(Debug)]
825    pub struct ConnectorServiceServer<T> {
826        inner: Arc<T>,
827        accept_compression_encodings: EnabledCompressionEncodings,
828        send_compression_encodings: EnabledCompressionEncodings,
829        max_decoding_message_size: Option<usize>,
830        max_encoding_message_size: Option<usize>,
831    }
832    impl<T> ConnectorServiceServer<T> {
833        pub fn new(inner: T) -> Self {
834            Self::from_arc(Arc::new(inner))
835        }
836        pub fn from_arc(inner: Arc<T>) -> Self {
837            Self {
838                inner,
839                accept_compression_encodings: Default::default(),
840                send_compression_encodings: Default::default(),
841                max_decoding_message_size: None,
842                max_encoding_message_size: None,
843            }
844        }
845        pub fn with_interceptor<F>(
846            inner: T,
847            interceptor: F,
848        ) -> InterceptedService<Self, F>
849        where
850            F: tonic::service::Interceptor,
851        {
852            InterceptedService::new(Self::new(inner), interceptor)
853        }
854        /// Enable decompressing requests with the given encoding.
855        #[must_use]
856        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
857            self.accept_compression_encodings.enable(encoding);
858            self
859        }
860        /// Compress responses with the given encoding, if the client supports it.
861        #[must_use]
862        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
863            self.send_compression_encodings.enable(encoding);
864            self
865        }
866        /// Limits the maximum size of a decoded message.
867        ///
868        /// Default: `4MB`
869        #[must_use]
870        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
871            self.max_decoding_message_size = Some(limit);
872            self
873        }
874        /// Limits the maximum size of an encoded message.
875        ///
876        /// Default: `usize::MAX`
877        #[must_use]
878        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
879            self.max_encoding_message_size = Some(limit);
880            self
881        }
882    }
883    impl<T, B> tonic::codegen::Service<http::Request<B>> for ConnectorServiceServer<T>
884    where
885        T: ConnectorService,
886        B: Body + std::marker::Send + 'static,
887        B::Error: Into<StdError> + std::marker::Send + 'static,
888    {
889        type Response = http::Response<tonic::body::Body>;
890        type Error = std::convert::Infallible;
891        type Future = BoxFuture<Self::Response, Self::Error>;
892        fn poll_ready(
893            &mut self,
894            _cx: &mut Context<'_>,
895        ) -> Poll<std::result::Result<(), Self::Error>> {
896            Poll::Ready(Ok(()))
897        }
898        fn call(&mut self, req: http::Request<B>) -> Self::Future {
899            match req.uri().path() {
900                "/connector_service.ConnectorService/SinkWriterStream" => {
901                    #[allow(non_camel_case_types)]
902                    struct SinkWriterStreamSvc<T: ConnectorService>(pub Arc<T>);
903                    impl<
904                        T: ConnectorService,
905                    > tonic::server::StreamingService<super::SinkWriterStreamRequest>
906                    for SinkWriterStreamSvc<T> {
907                        type Response = super::SinkWriterStreamResponse;
908                        type ResponseStream = T::SinkWriterStreamStream;
909                        type Future = BoxFuture<
910                            tonic::Response<Self::ResponseStream>,
911                            tonic::Status,
912                        >;
913                        fn call(
914                            &mut self,
915                            request: tonic::Request<
916                                tonic::Streaming<super::SinkWriterStreamRequest>,
917                            >,
918                        ) -> Self::Future {
919                            let inner = Arc::clone(&self.0);
920                            let fut = async move {
921                                <T as ConnectorService>::sink_writer_stream(&inner, request)
922                                    .await
923                            };
924                            Box::pin(fut)
925                        }
926                    }
927                    let accept_compression_encodings = self.accept_compression_encodings;
928                    let send_compression_encodings = self.send_compression_encodings;
929                    let max_decoding_message_size = self.max_decoding_message_size;
930                    let max_encoding_message_size = self.max_encoding_message_size;
931                    let inner = self.inner.clone();
932                    let fut = async move {
933                        let method = SinkWriterStreamSvc(inner);
934                        let codec = tonic_prost::ProstCodec::default();
935                        let mut grpc = tonic::server::Grpc::new(codec)
936                            .apply_compression_config(
937                                accept_compression_encodings,
938                                send_compression_encodings,
939                            )
940                            .apply_max_message_size_config(
941                                max_decoding_message_size,
942                                max_encoding_message_size,
943                            );
944                        let res = grpc.streaming(method, req).await;
945                        Ok(res)
946                    };
947                    Box::pin(fut)
948                }
949                "/connector_service.ConnectorService/SinkCoordinatorStream" => {
950                    #[allow(non_camel_case_types)]
951                    struct SinkCoordinatorStreamSvc<T: ConnectorService>(pub Arc<T>);
952                    impl<
953                        T: ConnectorService,
954                    > tonic::server::StreamingService<
955                        super::SinkCoordinatorStreamRequest,
956                    > for SinkCoordinatorStreamSvc<T> {
957                        type Response = super::SinkCoordinatorStreamResponse;
958                        type ResponseStream = T::SinkCoordinatorStreamStream;
959                        type Future = BoxFuture<
960                            tonic::Response<Self::ResponseStream>,
961                            tonic::Status,
962                        >;
963                        fn call(
964                            &mut self,
965                            request: tonic::Request<
966                                tonic::Streaming<super::SinkCoordinatorStreamRequest>,
967                            >,
968                        ) -> Self::Future {
969                            let inner = Arc::clone(&self.0);
970                            let fut = async move {
971                                <T as ConnectorService>::sink_coordinator_stream(
972                                        &inner,
973                                        request,
974                                    )
975                                    .await
976                            };
977                            Box::pin(fut)
978                        }
979                    }
980                    let accept_compression_encodings = self.accept_compression_encodings;
981                    let send_compression_encodings = self.send_compression_encodings;
982                    let max_decoding_message_size = self.max_decoding_message_size;
983                    let max_encoding_message_size = self.max_encoding_message_size;
984                    let inner = self.inner.clone();
985                    let fut = async move {
986                        let method = SinkCoordinatorStreamSvc(inner);
987                        let codec = tonic_prost::ProstCodec::default();
988                        let mut grpc = tonic::server::Grpc::new(codec)
989                            .apply_compression_config(
990                                accept_compression_encodings,
991                                send_compression_encodings,
992                            )
993                            .apply_max_message_size_config(
994                                max_decoding_message_size,
995                                max_encoding_message_size,
996                            );
997                        let res = grpc.streaming(method, req).await;
998                        Ok(res)
999                    };
1000                    Box::pin(fut)
1001                }
1002                "/connector_service.ConnectorService/ValidateSink" => {
1003                    #[allow(non_camel_case_types)]
1004                    struct ValidateSinkSvc<T: ConnectorService>(pub Arc<T>);
1005                    impl<
1006                        T: ConnectorService,
1007                    > tonic::server::UnaryService<super::ValidateSinkRequest>
1008                    for ValidateSinkSvc<T> {
1009                        type Response = super::ValidateSinkResponse;
1010                        type Future = BoxFuture<
1011                            tonic::Response<Self::Response>,
1012                            tonic::Status,
1013                        >;
1014                        fn call(
1015                            &mut self,
1016                            request: tonic::Request<super::ValidateSinkRequest>,
1017                        ) -> Self::Future {
1018                            let inner = Arc::clone(&self.0);
1019                            let fut = async move {
1020                                <T as ConnectorService>::validate_sink(&inner, request)
1021                                    .await
1022                            };
1023                            Box::pin(fut)
1024                        }
1025                    }
1026                    let accept_compression_encodings = self.accept_compression_encodings;
1027                    let send_compression_encodings = self.send_compression_encodings;
1028                    let max_decoding_message_size = self.max_decoding_message_size;
1029                    let max_encoding_message_size = self.max_encoding_message_size;
1030                    let inner = self.inner.clone();
1031                    let fut = async move {
1032                        let method = ValidateSinkSvc(inner);
1033                        let codec = tonic_prost::ProstCodec::default();
1034                        let mut grpc = tonic::server::Grpc::new(codec)
1035                            .apply_compression_config(
1036                                accept_compression_encodings,
1037                                send_compression_encodings,
1038                            )
1039                            .apply_max_message_size_config(
1040                                max_decoding_message_size,
1041                                max_encoding_message_size,
1042                            );
1043                        let res = grpc.unary(method, req).await;
1044                        Ok(res)
1045                    };
1046                    Box::pin(fut)
1047                }
1048                "/connector_service.ConnectorService/GetEventStream" => {
1049                    #[allow(non_camel_case_types)]
1050                    struct GetEventStreamSvc<T: ConnectorService>(pub Arc<T>);
1051                    impl<
1052                        T: ConnectorService,
1053                    > tonic::server::ServerStreamingService<super::GetEventStreamRequest>
1054                    for GetEventStreamSvc<T> {
1055                        type Response = super::GetEventStreamResponse;
1056                        type ResponseStream = T::GetEventStreamStream;
1057                        type Future = BoxFuture<
1058                            tonic::Response<Self::ResponseStream>,
1059                            tonic::Status,
1060                        >;
1061                        fn call(
1062                            &mut self,
1063                            request: tonic::Request<super::GetEventStreamRequest>,
1064                        ) -> Self::Future {
1065                            let inner = Arc::clone(&self.0);
1066                            let fut = async move {
1067                                <T as ConnectorService>::get_event_stream(&inner, request)
1068                                    .await
1069                            };
1070                            Box::pin(fut)
1071                        }
1072                    }
1073                    let accept_compression_encodings = self.accept_compression_encodings;
1074                    let send_compression_encodings = self.send_compression_encodings;
1075                    let max_decoding_message_size = self.max_decoding_message_size;
1076                    let max_encoding_message_size = self.max_encoding_message_size;
1077                    let inner = self.inner.clone();
1078                    let fut = async move {
1079                        let method = GetEventStreamSvc(inner);
1080                        let codec = tonic_prost::ProstCodec::default();
1081                        let mut grpc = tonic::server::Grpc::new(codec)
1082                            .apply_compression_config(
1083                                accept_compression_encodings,
1084                                send_compression_encodings,
1085                            )
1086                            .apply_max_message_size_config(
1087                                max_decoding_message_size,
1088                                max_encoding_message_size,
1089                            );
1090                        let res = grpc.server_streaming(method, req).await;
1091                        Ok(res)
1092                    };
1093                    Box::pin(fut)
1094                }
1095                "/connector_service.ConnectorService/ValidateSource" => {
1096                    #[allow(non_camel_case_types)]
1097                    struct ValidateSourceSvc<T: ConnectorService>(pub Arc<T>);
1098                    impl<
1099                        T: ConnectorService,
1100                    > tonic::server::UnaryService<super::ValidateSourceRequest>
1101                    for ValidateSourceSvc<T> {
1102                        type Response = super::ValidateSourceResponse;
1103                        type Future = BoxFuture<
1104                            tonic::Response<Self::Response>,
1105                            tonic::Status,
1106                        >;
1107                        fn call(
1108                            &mut self,
1109                            request: tonic::Request<super::ValidateSourceRequest>,
1110                        ) -> Self::Future {
1111                            let inner = Arc::clone(&self.0);
1112                            let fut = async move {
1113                                <T as ConnectorService>::validate_source(&inner, request)
1114                                    .await
1115                            };
1116                            Box::pin(fut)
1117                        }
1118                    }
1119                    let accept_compression_encodings = self.accept_compression_encodings;
1120                    let send_compression_encodings = self.send_compression_encodings;
1121                    let max_decoding_message_size = self.max_decoding_message_size;
1122                    let max_encoding_message_size = self.max_encoding_message_size;
1123                    let inner = self.inner.clone();
1124                    let fut = async move {
1125                        let method = ValidateSourceSvc(inner);
1126                        let codec = tonic_prost::ProstCodec::default();
1127                        let mut grpc = tonic::server::Grpc::new(codec)
1128                            .apply_compression_config(
1129                                accept_compression_encodings,
1130                                send_compression_encodings,
1131                            )
1132                            .apply_max_message_size_config(
1133                                max_decoding_message_size,
1134                                max_encoding_message_size,
1135                            );
1136                        let res = grpc.unary(method, req).await;
1137                        Ok(res)
1138                    };
1139                    Box::pin(fut)
1140                }
1141                _ => {
1142                    Box::pin(async move {
1143                        let mut response = http::Response::new(
1144                            tonic::body::Body::default(),
1145                        );
1146                        let headers = response.headers_mut();
1147                        headers
1148                            .insert(
1149                                tonic::Status::GRPC_STATUS,
1150                                (tonic::Code::Unimplemented as i32).into(),
1151                            );
1152                        headers
1153                            .insert(
1154                                http::header::CONTENT_TYPE,
1155                                tonic::metadata::GRPC_CONTENT_TYPE,
1156                            );
1157                        Ok(response)
1158                    })
1159                }
1160            }
1161        }
1162    }
1163    impl<T> Clone for ConnectorServiceServer<T> {
1164        fn clone(&self) -> Self {
1165            let inner = self.inner.clone();
1166            Self {
1167                inner,
1168                accept_compression_encodings: self.accept_compression_encodings,
1169                send_compression_encodings: self.send_compression_encodings,
1170                max_decoding_message_size: self.max_decoding_message_size,
1171                max_encoding_message_size: self.max_encoding_message_size,
1172            }
1173        }
1174    }
1175    /// Generated gRPC service name
1176    pub const SERVICE_NAME: &str = "connector_service.ConnectorService";
1177    impl<T> tonic::server::NamedService for ConnectorServiceServer<T> {
1178        const NAME: &'static str = SERVICE_NAME;
1179    }
1180}
1181/// Generated client implementations.
1182pub mod sink_coordination_service_client {
1183    #![allow(
1184        unused_variables,
1185        dead_code,
1186        missing_docs,
1187        clippy::wildcard_imports,
1188        clippy::let_unit_value,
1189    )]
1190    use tonic::codegen::*;
1191    use tonic::codegen::http::Uri;
1192    #[derive(Debug, Clone)]
1193    pub struct SinkCoordinationServiceClient<T> {
1194        inner: tonic::client::Grpc<T>,
1195    }
1196    impl SinkCoordinationServiceClient<tonic::transport::Channel> {
1197        /// Attempt to create a new client by connecting to a given endpoint.
1198        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
1199        where
1200            D: TryInto<tonic::transport::Endpoint>,
1201            D::Error: Into<StdError>,
1202        {
1203            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
1204            Ok(Self::new(conn))
1205        }
1206    }
1207    impl<T> SinkCoordinationServiceClient<T>
1208    where
1209        T: tonic::client::GrpcService<tonic::body::Body>,
1210        T::Error: Into<StdError>,
1211        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
1212        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
1213    {
1214        pub fn new(inner: T) -> Self {
1215            let inner = tonic::client::Grpc::new(inner);
1216            Self { inner }
1217        }
1218        pub fn with_origin(inner: T, origin: Uri) -> Self {
1219            let inner = tonic::client::Grpc::with_origin(inner, origin);
1220            Self { inner }
1221        }
1222        pub fn with_interceptor<F>(
1223            inner: T,
1224            interceptor: F,
1225        ) -> SinkCoordinationServiceClient<InterceptedService<T, F>>
1226        where
1227            F: tonic::service::Interceptor,
1228            T::ResponseBody: Default,
1229            T: tonic::codegen::Service<
1230                http::Request<tonic::body::Body>,
1231                Response = http::Response<
1232                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
1233                >,
1234            >,
1235            <T as tonic::codegen::Service<
1236                http::Request<tonic::body::Body>,
1237            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
1238        {
1239            SinkCoordinationServiceClient::new(
1240                InterceptedService::new(inner, interceptor),
1241            )
1242        }
1243        /// Compress requests with the given encoding.
1244        ///
1245        /// This requires the server to support it otherwise it might respond with an
1246        /// error.
1247        #[must_use]
1248        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1249            self.inner = self.inner.send_compressed(encoding);
1250            self
1251        }
1252        /// Enable decompressing responses.
1253        #[must_use]
1254        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
1255            self.inner = self.inner.accept_compressed(encoding);
1256            self
1257        }
1258        /// Limits the maximum size of a decoded message.
1259        ///
1260        /// Default: `4MB`
1261        #[must_use]
1262        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1263            self.inner = self.inner.max_decoding_message_size(limit);
1264            self
1265        }
1266        /// Limits the maximum size of an encoded message.
1267        ///
1268        /// Default: `usize::MAX`
1269        #[must_use]
1270        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1271            self.inner = self.inner.max_encoding_message_size(limit);
1272            self
1273        }
1274        pub async fn coordinate(
1275            &mut self,
1276            request: impl tonic::IntoStreamingRequest<Message = super::CoordinateRequest>,
1277        ) -> std::result::Result<
1278            tonic::Response<tonic::codec::Streaming<super::CoordinateResponse>>,
1279            tonic::Status,
1280        > {
1281            self.inner
1282                .ready()
1283                .await
1284                .map_err(|e| {
1285                    tonic::Status::unknown(
1286                        format!("Service was not ready: {}", e.into()),
1287                    )
1288                })?;
1289            let codec = tonic_prost::ProstCodec::default();
1290            let path = http::uri::PathAndQuery::from_static(
1291                "/connector_service.SinkCoordinationService/Coordinate",
1292            );
1293            let mut req = request.into_streaming_request();
1294            req.extensions_mut()
1295                .insert(
1296                    GrpcMethod::new(
1297                        "connector_service.SinkCoordinationService",
1298                        "Coordinate",
1299                    ),
1300                );
1301            self.inner.streaming(req, path, codec).await
1302        }
1303    }
1304}
1305/// Generated server implementations.
1306pub mod sink_coordination_service_server {
1307    #![allow(
1308        unused_variables,
1309        dead_code,
1310        missing_docs,
1311        clippy::wildcard_imports,
1312        clippy::let_unit_value,
1313    )]
1314    use tonic::codegen::*;
1315    /// Generated trait containing gRPC methods that should be implemented for use with SinkCoordinationServiceServer.
1316    #[async_trait]
1317    pub trait SinkCoordinationService: std::marker::Send + std::marker::Sync + 'static {
1318        /// Server streaming response type for the Coordinate method.
1319        type CoordinateStream: tonic::codegen::tokio_stream::Stream<
1320                Item = std::result::Result<super::CoordinateResponse, tonic::Status>,
1321            >
1322            + std::marker::Send
1323            + 'static;
1324        async fn coordinate(
1325            &self,
1326            request: tonic::Request<tonic::Streaming<super::CoordinateRequest>>,
1327        ) -> std::result::Result<tonic::Response<Self::CoordinateStream>, tonic::Status>;
1328    }
1329    #[derive(Debug)]
1330    pub struct SinkCoordinationServiceServer<T> {
1331        inner: Arc<T>,
1332        accept_compression_encodings: EnabledCompressionEncodings,
1333        send_compression_encodings: EnabledCompressionEncodings,
1334        max_decoding_message_size: Option<usize>,
1335        max_encoding_message_size: Option<usize>,
1336    }
1337    impl<T> SinkCoordinationServiceServer<T> {
1338        pub fn new(inner: T) -> Self {
1339            Self::from_arc(Arc::new(inner))
1340        }
1341        pub fn from_arc(inner: Arc<T>) -> Self {
1342            Self {
1343                inner,
1344                accept_compression_encodings: Default::default(),
1345                send_compression_encodings: Default::default(),
1346                max_decoding_message_size: None,
1347                max_encoding_message_size: None,
1348            }
1349        }
1350        pub fn with_interceptor<F>(
1351            inner: T,
1352            interceptor: F,
1353        ) -> InterceptedService<Self, F>
1354        where
1355            F: tonic::service::Interceptor,
1356        {
1357            InterceptedService::new(Self::new(inner), interceptor)
1358        }
1359        /// Enable decompressing requests with the given encoding.
1360        #[must_use]
1361        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
1362            self.accept_compression_encodings.enable(encoding);
1363            self
1364        }
1365        /// Compress responses with the given encoding, if the client supports it.
1366        #[must_use]
1367        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
1368            self.send_compression_encodings.enable(encoding);
1369            self
1370        }
1371        /// Limits the maximum size of a decoded message.
1372        ///
1373        /// Default: `4MB`
1374        #[must_use]
1375        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
1376            self.max_decoding_message_size = Some(limit);
1377            self
1378        }
1379        /// Limits the maximum size of an encoded message.
1380        ///
1381        /// Default: `usize::MAX`
1382        #[must_use]
1383        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
1384            self.max_encoding_message_size = Some(limit);
1385            self
1386        }
1387    }
1388    impl<T, B> tonic::codegen::Service<http::Request<B>>
1389    for SinkCoordinationServiceServer<T>
1390    where
1391        T: SinkCoordinationService,
1392        B: Body + std::marker::Send + 'static,
1393        B::Error: Into<StdError> + std::marker::Send + 'static,
1394    {
1395        type Response = http::Response<tonic::body::Body>;
1396        type Error = std::convert::Infallible;
1397        type Future = BoxFuture<Self::Response, Self::Error>;
1398        fn poll_ready(
1399            &mut self,
1400            _cx: &mut Context<'_>,
1401        ) -> Poll<std::result::Result<(), Self::Error>> {
1402            Poll::Ready(Ok(()))
1403        }
1404        fn call(&mut self, req: http::Request<B>) -> Self::Future {
1405            match req.uri().path() {
1406                "/connector_service.SinkCoordinationService/Coordinate" => {
1407                    #[allow(non_camel_case_types)]
1408                    struct CoordinateSvc<T: SinkCoordinationService>(pub Arc<T>);
1409                    impl<
1410                        T: SinkCoordinationService,
1411                    > tonic::server::StreamingService<super::CoordinateRequest>
1412                    for CoordinateSvc<T> {
1413                        type Response = super::CoordinateResponse;
1414                        type ResponseStream = T::CoordinateStream;
1415                        type Future = BoxFuture<
1416                            tonic::Response<Self::ResponseStream>,
1417                            tonic::Status,
1418                        >;
1419                        fn call(
1420                            &mut self,
1421                            request: tonic::Request<
1422                                tonic::Streaming<super::CoordinateRequest>,
1423                            >,
1424                        ) -> Self::Future {
1425                            let inner = Arc::clone(&self.0);
1426                            let fut = async move {
1427                                <T as SinkCoordinationService>::coordinate(&inner, request)
1428                                    .await
1429                            };
1430                            Box::pin(fut)
1431                        }
1432                    }
1433                    let accept_compression_encodings = self.accept_compression_encodings;
1434                    let send_compression_encodings = self.send_compression_encodings;
1435                    let max_decoding_message_size = self.max_decoding_message_size;
1436                    let max_encoding_message_size = self.max_encoding_message_size;
1437                    let inner = self.inner.clone();
1438                    let fut = async move {
1439                        let method = CoordinateSvc(inner);
1440                        let codec = tonic_prost::ProstCodec::default();
1441                        let mut grpc = tonic::server::Grpc::new(codec)
1442                            .apply_compression_config(
1443                                accept_compression_encodings,
1444                                send_compression_encodings,
1445                            )
1446                            .apply_max_message_size_config(
1447                                max_decoding_message_size,
1448                                max_encoding_message_size,
1449                            );
1450                        let res = grpc.streaming(method, req).await;
1451                        Ok(res)
1452                    };
1453                    Box::pin(fut)
1454                }
1455                _ => {
1456                    Box::pin(async move {
1457                        let mut response = http::Response::new(
1458                            tonic::body::Body::default(),
1459                        );
1460                        let headers = response.headers_mut();
1461                        headers
1462                            .insert(
1463                                tonic::Status::GRPC_STATUS,
1464                                (tonic::Code::Unimplemented as i32).into(),
1465                            );
1466                        headers
1467                            .insert(
1468                                http::header::CONTENT_TYPE,
1469                                tonic::metadata::GRPC_CONTENT_TYPE,
1470                            );
1471                        Ok(response)
1472                    })
1473                }
1474            }
1475        }
1476    }
1477    impl<T> Clone for SinkCoordinationServiceServer<T> {
1478        fn clone(&self) -> Self {
1479            let inner = self.inner.clone();
1480            Self {
1481                inner,
1482                accept_compression_encodings: self.accept_compression_encodings,
1483                send_compression_encodings: self.send_compression_encodings,
1484                max_decoding_message_size: self.max_decoding_message_size,
1485                max_encoding_message_size: self.max_encoding_message_size,
1486            }
1487        }
1488    }
1489    /// Generated gRPC service name
1490    pub const SERVICE_NAME: &str = "connector_service.SinkCoordinationService";
1491    impl<T> tonic::server::NamedService for SinkCoordinationServiceServer<T> {
1492        const NAME: &'static str = SERVICE_NAME;
1493    }
1494}
risingwave_pb/connector_service.rs

risingwave_pb/
connector_service.rs
