risingwave_pb/
stream_service.rs

1// This file is @generated by prost-build.
2#[derive(prost_helpers::AnyPB)]
3#[derive(Clone, PartialEq, ::prost::Message)]
4pub struct InjectBarrierRequest {
5    #[prost(string, tag = "1")]
6    pub request_id: ::prost::alloc::string::String,
7    #[prost(message, optional, tag = "2")]
8    pub barrier: ::core::option::Option<super::stream_plan::Barrier>,
9    #[prost(uint32, repeated, tag = "4", wrapper = "crate::id::ActorId")]
10    pub actor_ids_to_collect: ::prost::alloc::vec::Vec<crate::id::ActorId>,
11    #[prost(uint32, repeated, tag = "5", wrapper = "crate::id::TableId")]
12    pub table_ids_to_sync: ::prost::alloc::vec::Vec<crate::id::TableId>,
13    #[prost(uint64, tag = "6", wrapper = "crate::id::PartialGraphId")]
14    pub partial_graph_id: crate::id::PartialGraphId,
15    #[prost(message, repeated, tag = "9")]
16    pub actors_to_build: ::prost::alloc::vec::Vec<
17        inject_barrier_request::FragmentBuildActorInfo,
18    >,
19}
20/// Nested message and enum types in `InjectBarrierRequest`.
21pub mod inject_barrier_request {
22    #[derive(prost_helpers::AnyPB)]
23    #[derive(Clone, PartialEq, ::prost::Message)]
24    pub struct FragmentBuildActorInfo {
25        #[prost(uint32, tag = "1", wrapper = "crate::id::FragmentId")]
26        pub fragment_id: crate::id::FragmentId,
27        #[prost(message, optional, tag = "2")]
28        pub node: ::core::option::Option<super::super::stream_plan::StreamNode>,
29        #[prost(message, repeated, tag = "3")]
30        pub actors: ::prost::alloc::vec::Vec<BuildActorInfo>,
31    }
32    #[derive(prost_helpers::AnyPB)]
33    #[derive(Clone, PartialEq, ::prost::Message)]
34    pub struct BuildActorInfo {
35        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
36        pub actor_id: crate::id::ActorId,
37        #[prost(map = "uint32, message", tag = "2", wrapper = "crate::id::FragmentId")]
38        pub fragment_upstreams: ::std::collections::HashMap<
39            crate::id::FragmentId,
40            build_actor_info::UpstreamActors,
41        >,
42        #[prost(message, repeated, tag = "3")]
43        pub dispatchers: ::prost::alloc::vec::Vec<super::super::stream_plan::Dispatcher>,
44        #[prost(message, optional, tag = "4")]
45        pub vnode_bitmap: ::core::option::Option<super::super::common::Buffer>,
46        #[prost(string, tag = "5")]
47        pub mview_definition: ::prost::alloc::string::String,
48        #[prost(message, optional, tag = "6")]
49        pub expr_context: ::core::option::Option<super::super::plan_common::ExprContext>,
50        #[prost(string, tag = "9")]
51        pub config_override: ::prost::alloc::string::String,
52        #[prost(uint32, repeated, tag = "7", wrapper = "crate::id::SubscriberId")]
53        pub initial_subscriber_ids: ::prost::alloc::vec::Vec<crate::id::SubscriberId>,
54    }
55    /// Nested message and enum types in `BuildActorInfo`.
56    pub mod build_actor_info {
57        #[derive(prost_helpers::AnyPB)]
58        #[derive(Clone, PartialEq, ::prost::Message)]
59        pub struct UpstreamActors {
60            #[prost(message, repeated, tag = "1")]
61            pub actors: ::prost::alloc::vec::Vec<super::super::super::common::ActorInfo>,
62        }
63    }
64}
65#[derive(prost_helpers::AnyPB)]
66#[derive(Clone, PartialEq, ::prost::Message)]
67pub struct BarrierCompleteResponse {
68    #[prost(string, tag = "1")]
69    pub request_id: ::prost::alloc::string::String,
70    #[prost(message, optional, tag = "2")]
71    pub status: ::core::option::Option<super::common::Status>,
72    #[prost(message, repeated, tag = "3")]
73    pub create_mview_progress: ::prost::alloc::vec::Vec<
74        barrier_complete_response::CreateMviewProgress,
75    >,
76    #[prost(message, repeated, tag = "4")]
77    pub synced_sstables: ::prost::alloc::vec::Vec<
78        barrier_complete_response::LocalSstableInfo,
79    >,
80    #[prost(uint32, tag = "5", wrapper = "crate::id::WorkerId")]
81    pub worker_id: crate::id::WorkerId,
82    #[prost(map = "uint32, message", tag = "6", wrapper = "crate::id::TableId")]
83    pub table_watermarks: ::std::collections::HashMap<
84        crate::id::TableId,
85        super::hummock::TableWatermarks,
86    >,
87    #[prost(message, repeated, tag = "7")]
88    pub old_value_sstables: ::prost::alloc::vec::Vec<super::hummock::SstableInfo>,
89    #[prost(uint64, tag = "8", wrapper = "crate::id::PartialGraphId")]
90    pub partial_graph_id: crate::id::PartialGraphId,
91    /// prev_epoch of barrier
92    #[prost(uint64, tag = "9")]
93    pub epoch: u64,
94    #[prost(message, repeated, tag = "11")]
95    pub load_finished_sources: ::prost::alloc::vec::Vec<
96        barrier_complete_response::LoadFinishedSource,
97    >,
98    #[prost(map = "uint32, message", tag = "12", wrapper = "crate::id::TableId")]
99    pub vector_index_adds: ::std::collections::HashMap<
100        crate::id::TableId,
101        super::hummock::vector_index_delta::VectorIndexAdds,
102    >,
103    #[prost(message, repeated, tag = "13")]
104    pub cdc_table_backfill_progress: ::prost::alloc::vec::Vec<
105        barrier_complete_response::CdcTableBackfillProgress,
106    >,
107    /// Used for truncating tables in storage layer.
108    /// MaterializeExecutor reports the tables to truncate, and then
109    /// meta will apply truncate operation in next commit epoch.
110    #[prost(uint32, repeated, tag = "14", wrapper = "crate::id::TableId")]
111    pub truncate_tables: ::prost::alloc::vec::Vec<crate::id::TableId>,
112    /// Used for reporting materialized view refresh completion.
113    /// MaterializeExecutor reports when refresh has finished, and then
114    /// meta will update the table's refresh state to Finished.
115    #[prost(uint32, repeated, tag = "15", wrapper = "crate::id::TableId")]
116    pub refresh_finished_tables: ::prost::alloc::vec::Vec<crate::id::TableId>,
117    #[prost(message, repeated, tag = "16")]
118    pub list_finished_sources: ::prost::alloc::vec::Vec<
119        barrier_complete_response::ListFinishedSource,
120    >,
121    #[prost(message, repeated, tag = "17")]
122    pub cdc_source_offset_updated: ::prost::alloc::vec::Vec<
123        barrier_complete_response::CdcSourceOffsetUpdated,
124    >,
125    #[prost(message, repeated, tag = "18")]
126    pub iceberg_pk_index_sink_metadata: ::prost::alloc::vec::Vec<
127        barrier_complete_response::IcebergPkIndexSinkMetadata,
128    >,
129}
130/// Nested message and enum types in `BarrierCompleteResponse`.
131pub mod barrier_complete_response {
132    #[derive(prost_helpers::AnyPB)]
133    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
134    pub struct CreateMviewProgress {
135        /// Note: ideally we should use `executor_id`, but `actor_id` is ok-ish.
136        /// See <<https://github.com/risingwavelabs/risingwave/issues/6236>.>
137        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
138        pub backfill_actor_id: crate::id::ActorId,
139        #[prost(bool, tag = "2")]
140        pub done: bool,
141        /// MV backfill snapshot read epoch (0 for Done / Source backfill)
142        #[prost(uint64, tag = "3")]
143        pub consumed_epoch: u64,
144        /// MV backfill snapshot read rows / Source backfilled rows
145        #[prost(uint64, tag = "4")]
146        pub consumed_rows: u64,
147        #[prost(uint64, tag = "5")]
148        pub pending_epoch_lag: u64,
149        /// Buffered rows that are yet to be consumed (used by locality backfill to report precise progress)
150        #[prost(uint64, tag = "6")]
151        pub buffered_rows: u64,
152        #[prost(uint32, tag = "7", wrapper = "crate::id::FragmentId")]
153        pub fragment_id: crate::id::FragmentId,
154    }
155    #[derive(prost_helpers::AnyPB)]
156    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
157    pub struct CdcTableBackfillProgress {
158        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
159        pub actor_id: crate::id::ActorId,
160        #[prost(uint64, tag = "2")]
161        pub epoch: u64,
162        #[prost(bool, tag = "3")]
163        pub done: bool,
164        #[prost(int64, tag = "4")]
165        pub split_id_start_inclusive: i64,
166        #[prost(int64, tag = "5")]
167        pub split_id_end_inclusive: i64,
168        #[prost(uint64, tag = "6")]
169        pub generation: u64,
170        #[prost(uint32, tag = "7", wrapper = "crate::id::FragmentId")]
171        pub fragment_id: crate::id::FragmentId,
172    }
173    #[derive(prost_helpers::AnyPB)]
174    #[derive(Clone, PartialEq, ::prost::Message)]
175    pub struct LocalSstableInfo {
176        #[prost(message, optional, tag = "2")]
177        pub sst: ::core::option::Option<super::super::hummock::SstableInfo>,
178        #[prost(map = "uint32, message", tag = "3", wrapper = "crate::id::TableId")]
179        pub table_stats_map: ::std::collections::HashMap<
180            crate::id::TableId,
181            super::super::hummock::TableStats,
182        >,
183        #[prost(uint64, tag = "4")]
184        pub created_at: u64,
185    }
186    /// Used for refreshable batch source.
187    /// SourceExecutor reports the source load is finished, and then
188    /// meta will issue a LoadFinish barrier to notify MaterializeExecutor to start diff calculation.
189    #[derive(prost_helpers::AnyPB)]
190    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
191    pub struct LoadFinishedSource {
192        /// The actor that reported the completion event.
193        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
194        pub reporter_actor_id: crate::id::ActorId,
195        /// The table ID for the refreshable batch source.
196        #[prost(uint32, tag = "2", wrapper = "crate::id::TableId")]
197        pub table_id: crate::id::TableId,
198        /// The source identifier associated with the finished load.
199        #[prost(uint32, tag = "3", wrapper = "crate::id::SourceId")]
200        pub associated_source_id: crate::id::SourceId,
201    }
202    /// SourceExecutor reports the source list is finished, and then
203    /// meta will issue a ListFinish barrier to notify the source to start loading.
204    #[derive(prost_helpers::AnyPB)]
205    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
206    pub struct ListFinishedSource {
207        /// The actor that reported the completion event.
208        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
209        pub reporter_actor_id: crate::id::ActorId,
210        /// The table ID for the refreshable batch source.
211        #[prost(uint32, tag = "2", wrapper = "crate::id::TableId")]
212        pub table_id: crate::id::TableId,
213        /// The source identifier associated with the completed listing.
214        #[prost(uint32, tag = "3", wrapper = "crate::id::SourceId")]
215        pub associated_source_id: crate::id::SourceId,
216    }
217    /// SourceExecutor reports that CDC source has updated offset at least once.
218    /// Meta will mark the CDC source as created only after receiving this notification.
219    #[derive(prost_helpers::AnyPB)]
220    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
221    pub struct CdcSourceOffsetUpdated {
222        /// The actor that reported the offset update.
223        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
224        pub reporter_actor_id: crate::id::ActorId,
225        /// The source identifier for the CDC source.
226        #[prost(uint32, tag = "2", wrapper = "crate::id::SourceId")]
227        pub source_id: crate::id::SourceId,
228    }
229    #[derive(prost_helpers::AnyPB)]
230    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
231    pub struct IcebergPkIndexSinkMetadata {
232        #[prost(uint32, tag = "1", wrapper = "crate::id::ActorId")]
233        pub reporter_actor_id: crate::id::ActorId,
234        #[prost(uint32, tag = "2", wrapper = "crate::id::SinkId")]
235        pub sink_id: crate::id::SinkId,
236        #[prost(uint64, tag = "3")]
237        pub prev_epoch: u64,
238        #[prost(enumeration = "super::PbIcebergPkIndexSinkRole", tag = "4")]
239        pub role: i32,
240        #[prost(message, optional, tag = "5")]
241        pub metadata: ::core::option::Option<
242            super::super::connector_service::SinkMetadata,
243        >,
244    }
245}
246#[derive(prost_helpers::AnyPB)]
247#[derive(Clone, PartialEq, ::prost::Message)]
248pub struct StreamingControlStreamRequest {
249    #[prost(oneof = "streaming_control_stream_request::Request", tags = "1, 2, 3, 4, 5")]
250    pub request: ::core::option::Option<streaming_control_stream_request::Request>,
251}
252/// Nested message and enum types in `StreamingControlStreamRequest`.
253pub mod streaming_control_stream_request {
254    #[derive(prost_helpers::AnyPB)]
255    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
256    pub struct InitRequest {
257        #[prost(string, tag = "1")]
258        pub term_id: ::prost::alloc::string::String,
259    }
260    #[derive(prost_helpers::AnyPB)]
261    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
262    pub struct CreatePartialGraphRequest {
263        #[prost(uint64, tag = "1", wrapper = "crate::id::PartialGraphId")]
264        pub partial_graph_id: crate::id::PartialGraphId,
265    }
266    #[derive(prost_helpers::AnyPB)]
267    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
268    pub struct RemovePartialGraphRequest {
269        #[prost(uint64, repeated, tag = "1", wrapper = "crate::id::PartialGraphId")]
270        pub partial_graph_ids: ::prost::alloc::vec::Vec<crate::id::PartialGraphId>,
271    }
272    #[derive(prost_helpers::AnyPB)]
273    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
274    pub struct ResetPartialGraphsRequest {
275        #[prost(uint64, repeated, tag = "1", wrapper = "crate::id::PartialGraphId")]
276        pub partial_graph_ids: ::prost::alloc::vec::Vec<crate::id::PartialGraphId>,
277    }
278    #[derive(prost_helpers::AnyPB)]
279    #[derive(Clone, PartialEq, ::prost::Oneof)]
280    pub enum Request {
281        #[prost(message, tag = "1")]
282        Init(InitRequest),
283        #[prost(message, tag = "2")]
284        InjectBarrier(super::InjectBarrierRequest),
285        #[prost(message, tag = "3")]
286        RemovePartialGraph(RemovePartialGraphRequest),
287        #[prost(message, tag = "4")]
288        CreatePartialGraph(CreatePartialGraphRequest),
289        #[prost(message, tag = "5")]
290        ResetPartialGraphs(ResetPartialGraphsRequest),
291    }
292}
293#[derive(prost_helpers::AnyPB)]
294#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
295pub struct ScoredError {
296    #[prost(string, tag = "1")]
297    pub err_msg: ::prost::alloc::string::String,
298    #[prost(int32, tag = "2")]
299    pub score: i32,
300}
301#[derive(prost_helpers::AnyPB)]
302#[derive(Clone, PartialEq, ::prost::Message)]
303pub struct StreamingControlStreamResponse {
304    #[prost(
305        oneof = "streaming_control_stream_response::Response",
306        tags = "1, 2, 3, 4, 5"
307    )]
308    pub response: ::core::option::Option<streaming_control_stream_response::Response>,
309}
310/// Nested message and enum types in `StreamingControlStreamResponse`.
311pub mod streaming_control_stream_response {
312    #[derive(prost_helpers::AnyPB)]
313    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
314    pub struct InitResponse {}
315    #[derive(prost_helpers::AnyPB)]
316    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
317    pub struct ShutdownResponse {}
318    #[derive(prost_helpers::AnyPB)]
319    #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
320    pub struct ReportPartialGraphFailureResponse {
321        #[prost(uint64, tag = "1", wrapper = "crate::id::PartialGraphId")]
322        pub partial_graph_id: crate::id::PartialGraphId,
323    }
324    #[derive(prost_helpers::AnyPB)]
325    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
326    pub struct ResetPartialGraphResponse {
327        #[prost(uint64, tag = "1", wrapper = "crate::id::PartialGraphId")]
328        pub partial_graph_id: crate::id::PartialGraphId,
329        #[prost(message, optional, tag = "2")]
330        pub root_err: ::core::option::Option<super::ScoredError>,
331    }
332    #[derive(prost_helpers::AnyPB)]
333    #[derive(Clone, PartialEq, ::prost::Oneof)]
334    pub enum Response {
335        #[prost(message, tag = "1")]
336        Init(InitResponse),
337        #[prost(message, tag = "2")]
338        CompleteBarrier(super::BarrierCompleteResponse),
339        #[prost(message, tag = "3")]
340        Shutdown(ShutdownResponse),
341        #[prost(message, tag = "4")]
342        ReportPartialGraphFailure(ReportPartialGraphFailureResponse),
343        #[prost(message, tag = "5")]
344        ResetPartialGraph(ResetPartialGraphResponse),
345    }
346}
347#[derive(prost_helpers::AnyPB)]
348#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
349pub struct GetMinUncommittedObjectIdRequest {}
350#[derive(prost_helpers::AnyPB)]
351#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
352pub struct GetMinUncommittedObjectIdResponse {
353    #[prost(uint64, tag = "1", wrapper = "crate::id::HummockRawObjectId")]
354    pub min_uncommitted_object_id: crate::id::HummockRawObjectId,
355}
356#[derive(prost_helpers::AnyPB)]
357#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
358#[repr(i32)]
359pub enum PbIcebergPkIndexSinkRole {
360    Unspecified = 0,
361    Writer = 1,
362    PositionDeleteMerger = 2,
363    CompactionResolver = 3,
364}
365impl PbIcebergPkIndexSinkRole {
366    /// String value of the enum field names used in the ProtoBuf definition.
367    ///
368    /// The values are not transformed in any way and thus are considered stable
369    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
370    pub fn as_str_name(&self) -> &'static str {
371        match self {
372            Self::Unspecified => "PB_ICEBERG_PK_INDEX_SINK_ROLE_UNSPECIFIED",
373            Self::Writer => "PB_ICEBERG_PK_INDEX_SINK_ROLE_WRITER",
374            Self::PositionDeleteMerger => {
375                "PB_ICEBERG_PK_INDEX_SINK_ROLE_POSITION_DELETE_MERGER"
376            }
377            Self::CompactionResolver => {
378                "PB_ICEBERG_PK_INDEX_SINK_ROLE_COMPACTION_RESOLVER"
379            }
380        }
381    }
382    /// Creates an enum from field names used in the ProtoBuf definition.
383    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
384        match value {
385            "PB_ICEBERG_PK_INDEX_SINK_ROLE_UNSPECIFIED" => Some(Self::Unspecified),
386            "PB_ICEBERG_PK_INDEX_SINK_ROLE_WRITER" => Some(Self::Writer),
387            "PB_ICEBERG_PK_INDEX_SINK_ROLE_POSITION_DELETE_MERGER" => {
388                Some(Self::PositionDeleteMerger)
389            }
390            "PB_ICEBERG_PK_INDEX_SINK_ROLE_COMPACTION_RESOLVER" => {
391                Some(Self::CompactionResolver)
392            }
393            _ => None,
394        }
395    }
396}
397/// Generated client implementations.
398pub mod stream_service_client {
399    #![allow(
400        unused_variables,
401        dead_code,
402        missing_docs,
403        clippy::wildcard_imports,
404        clippy::let_unit_value,
405    )]
406    use tonic::codegen::*;
407    use tonic::codegen::http::Uri;
408    #[derive(Debug, Clone)]
409    pub struct StreamServiceClient<T> {
410        inner: tonic::client::Grpc<T>,
411    }
412    impl StreamServiceClient<tonic::transport::Channel> {
413        /// Attempt to create a new client by connecting to a given endpoint.
414        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
415        where
416            D: TryInto<tonic::transport::Endpoint>,
417            D::Error: Into<StdError>,
418        {
419            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
420            Ok(Self::new(conn))
421        }
422    }
423    impl<T> StreamServiceClient<T>
424    where
425        T: tonic::client::GrpcService<tonic::body::Body>,
426        T::Error: Into<StdError>,
427        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
428        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
429    {
430        pub fn new(inner: T) -> Self {
431            let inner = tonic::client::Grpc::new(inner);
432            Self { inner }
433        }
434        pub fn with_origin(inner: T, origin: Uri) -> Self {
435            let inner = tonic::client::Grpc::with_origin(inner, origin);
436            Self { inner }
437        }
438        pub fn with_interceptor<F>(
439            inner: T,
440            interceptor: F,
441        ) -> StreamServiceClient<InterceptedService<T, F>>
442        where
443            F: tonic::service::Interceptor,
444            T::ResponseBody: Default,
445            T: tonic::codegen::Service<
446                http::Request<tonic::body::Body>,
447                Response = http::Response<
448                    <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
449                >,
450            >,
451            <T as tonic::codegen::Service<
452                http::Request<tonic::body::Body>,
453            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
454        {
455            StreamServiceClient::new(InterceptedService::new(inner, interceptor))
456        }
457        /// Compress requests with the given encoding.
458        ///
459        /// This requires the server to support it otherwise it might respond with an
460        /// error.
461        #[must_use]
462        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
463            self.inner = self.inner.send_compressed(encoding);
464            self
465        }
466        /// Enable decompressing responses.
467        #[must_use]
468        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
469            self.inner = self.inner.accept_compressed(encoding);
470            self
471        }
472        /// Limits the maximum size of a decoded message.
473        ///
474        /// Default: `4MB`
475        #[must_use]
476        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
477            self.inner = self.inner.max_decoding_message_size(limit);
478            self
479        }
480        /// Limits the maximum size of an encoded message.
481        ///
482        /// Default: `usize::MAX`
483        #[must_use]
484        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
485            self.inner = self.inner.max_encoding_message_size(limit);
486            self
487        }
488        pub async fn streaming_control_stream(
489            &mut self,
490            request: impl tonic::IntoStreamingRequest<
491                Message = super::StreamingControlStreamRequest,
492            >,
493        ) -> std::result::Result<
494            tonic::Response<
495                tonic::codec::Streaming<super::StreamingControlStreamResponse>,
496            >,
497            tonic::Status,
498        > {
499            self.inner
500                .ready()
501                .await
502                .map_err(|e| {
503                    tonic::Status::unknown(
504                        format!("Service was not ready: {}", e.into()),
505                    )
506                })?;
507            let codec = tonic_prost::ProstCodec::default();
508            let path = http::uri::PathAndQuery::from_static(
509                "/stream_service.StreamService/StreamingControlStream",
510            );
511            let mut req = request.into_streaming_request();
512            req.extensions_mut()
513                .insert(
514                    GrpcMethod::new(
515                        "stream_service.StreamService",
516                        "StreamingControlStream",
517                    ),
518                );
519            self.inner.streaming(req, path, codec).await
520        }
521        pub async fn get_min_uncommitted_object_id(
522            &mut self,
523            request: impl tonic::IntoRequest<super::GetMinUncommittedObjectIdRequest>,
524        ) -> std::result::Result<
525            tonic::Response<super::GetMinUncommittedObjectIdResponse>,
526            tonic::Status,
527        > {
528            self.inner
529                .ready()
530                .await
531                .map_err(|e| {
532                    tonic::Status::unknown(
533                        format!("Service was not ready: {}", e.into()),
534                    )
535                })?;
536            let codec = tonic_prost::ProstCodec::default();
537            let path = http::uri::PathAndQuery::from_static(
538                "/stream_service.StreamService/GetMinUncommittedObjectId",
539            );
540            let mut req = request.into_request();
541            req.extensions_mut()
542                .insert(
543                    GrpcMethod::new(
544                        "stream_service.StreamService",
545                        "GetMinUncommittedObjectId",
546                    ),
547                );
548            self.inner.unary(req, path, codec).await
549        }
550    }
551}
552/// Generated server implementations.
553pub mod stream_service_server {
554    #![allow(
555        unused_variables,
556        dead_code,
557        missing_docs,
558        clippy::wildcard_imports,
559        clippy::let_unit_value,
560    )]
561    use tonic::codegen::*;
562    /// Generated trait containing gRPC methods that should be implemented for use with StreamServiceServer.
563    #[async_trait]
564    pub trait StreamService: std::marker::Send + std::marker::Sync + 'static {
565        /// Server streaming response type for the StreamingControlStream method.
566        type StreamingControlStreamStream: tonic::codegen::tokio_stream::Stream<
567                Item = std::result::Result<
568                    super::StreamingControlStreamResponse,
569                    tonic::Status,
570                >,
571            >
572            + std::marker::Send
573            + 'static;
574        async fn streaming_control_stream(
575            &self,
576            request: tonic::Request<
577                tonic::Streaming<super::StreamingControlStreamRequest>,
578            >,
579        ) -> std::result::Result<
580            tonic::Response<Self::StreamingControlStreamStream>,
581            tonic::Status,
582        >;
583        async fn get_min_uncommitted_object_id(
584            &self,
585            request: tonic::Request<super::GetMinUncommittedObjectIdRequest>,
586        ) -> std::result::Result<
587            tonic::Response<super::GetMinUncommittedObjectIdResponse>,
588            tonic::Status,
589        >;
590    }
591    #[derive(Debug)]
592    pub struct StreamServiceServer<T> {
593        inner: Arc<T>,
594        accept_compression_encodings: EnabledCompressionEncodings,
595        send_compression_encodings: EnabledCompressionEncodings,
596        max_decoding_message_size: Option<usize>,
597        max_encoding_message_size: Option<usize>,
598    }
599    impl<T> StreamServiceServer<T> {
600        pub fn new(inner: T) -> Self {
601            Self::from_arc(Arc::new(inner))
602        }
603        pub fn from_arc(inner: Arc<T>) -> Self {
604            Self {
605                inner,
606                accept_compression_encodings: Default::default(),
607                send_compression_encodings: Default::default(),
608                max_decoding_message_size: None,
609                max_encoding_message_size: None,
610            }
611        }
612        pub fn with_interceptor<F>(
613            inner: T,
614            interceptor: F,
615        ) -> InterceptedService<Self, F>
616        where
617            F: tonic::service::Interceptor,
618        {
619            InterceptedService::new(Self::new(inner), interceptor)
620        }
621        /// Enable decompressing requests with the given encoding.
622        #[must_use]
623        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
624            self.accept_compression_encodings.enable(encoding);
625            self
626        }
627        /// Compress responses with the given encoding, if the client supports it.
628        #[must_use]
629        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
630            self.send_compression_encodings.enable(encoding);
631            self
632        }
633        /// Limits the maximum size of a decoded message.
634        ///
635        /// Default: `4MB`
636        #[must_use]
637        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
638            self.max_decoding_message_size = Some(limit);
639            self
640        }
641        /// Limits the maximum size of an encoded message.
642        ///
643        /// Default: `usize::MAX`
644        #[must_use]
645        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
646            self.max_encoding_message_size = Some(limit);
647            self
648        }
649    }
650    impl<T, B> tonic::codegen::Service<http::Request<B>> for StreamServiceServer<T>
651    where
652        T: StreamService,
653        B: Body + std::marker::Send + 'static,
654        B::Error: Into<StdError> + std::marker::Send + 'static,
655    {
656        type Response = http::Response<tonic::body::Body>;
657        type Error = std::convert::Infallible;
658        type Future = BoxFuture<Self::Response, Self::Error>;
659        fn poll_ready(
660            &mut self,
661            _cx: &mut Context<'_>,
662        ) -> Poll<std::result::Result<(), Self::Error>> {
663            Poll::Ready(Ok(()))
664        }
665        fn call(&mut self, req: http::Request<B>) -> Self::Future {
666            match req.uri().path() {
667                "/stream_service.StreamService/StreamingControlStream" => {
668                    #[allow(non_camel_case_types)]
669                    struct StreamingControlStreamSvc<T: StreamService>(pub Arc<T>);
670                    impl<
671                        T: StreamService,
672                    > tonic::server::StreamingService<
673                        super::StreamingControlStreamRequest,
674                    > for StreamingControlStreamSvc<T> {
675                        type Response = super::StreamingControlStreamResponse;
676                        type ResponseStream = T::StreamingControlStreamStream;
677                        type Future = BoxFuture<
678                            tonic::Response<Self::ResponseStream>,
679                            tonic::Status,
680                        >;
681                        fn call(
682                            &mut self,
683                            request: tonic::Request<
684                                tonic::Streaming<super::StreamingControlStreamRequest>,
685                            >,
686                        ) -> Self::Future {
687                            let inner = Arc::clone(&self.0);
688                            let fut = async move {
689                                <T as StreamService>::streaming_control_stream(
690                                        &inner,
691                                        request,
692                                    )
693                                    .await
694                            };
695                            Box::pin(fut)
696                        }
697                    }
698                    let accept_compression_encodings = self.accept_compression_encodings;
699                    let send_compression_encodings = self.send_compression_encodings;
700                    let max_decoding_message_size = self.max_decoding_message_size;
701                    let max_encoding_message_size = self.max_encoding_message_size;
702                    let inner = self.inner.clone();
703                    let fut = async move {
704                        let method = StreamingControlStreamSvc(inner);
705                        let codec = tonic_prost::ProstCodec::default();
706                        let mut grpc = tonic::server::Grpc::new(codec)
707                            .apply_compression_config(
708                                accept_compression_encodings,
709                                send_compression_encodings,
710                            )
711                            .apply_max_message_size_config(
712                                max_decoding_message_size,
713                                max_encoding_message_size,
714                            );
715                        let res = grpc.streaming(method, req).await;
716                        Ok(res)
717                    };
718                    Box::pin(fut)
719                }
720                "/stream_service.StreamService/GetMinUncommittedObjectId" => {
721                    #[allow(non_camel_case_types)]
722                    struct GetMinUncommittedObjectIdSvc<T: StreamService>(pub Arc<T>);
723                    impl<
724                        T: StreamService,
725                    > tonic::server::UnaryService<
726                        super::GetMinUncommittedObjectIdRequest,
727                    > for GetMinUncommittedObjectIdSvc<T> {
728                        type Response = super::GetMinUncommittedObjectIdResponse;
729                        type Future = BoxFuture<
730                            tonic::Response<Self::Response>,
731                            tonic::Status,
732                        >;
733                        fn call(
734                            &mut self,
735                            request: tonic::Request<
736                                super::GetMinUncommittedObjectIdRequest,
737                            >,
738                        ) -> Self::Future {
739                            let inner = Arc::clone(&self.0);
740                            let fut = async move {
741                                <T as StreamService>::get_min_uncommitted_object_id(
742                                        &inner,
743                                        request,
744                                    )
745                                    .await
746                            };
747                            Box::pin(fut)
748                        }
749                    }
750                    let accept_compression_encodings = self.accept_compression_encodings;
751                    let send_compression_encodings = self.send_compression_encodings;
752                    let max_decoding_message_size = self.max_decoding_message_size;
753                    let max_encoding_message_size = self.max_encoding_message_size;
754                    let inner = self.inner.clone();
755                    let fut = async move {
756                        let method = GetMinUncommittedObjectIdSvc(inner);
757                        let codec = tonic_prost::ProstCodec::default();
758                        let mut grpc = tonic::server::Grpc::new(codec)
759                            .apply_compression_config(
760                                accept_compression_encodings,
761                                send_compression_encodings,
762                            )
763                            .apply_max_message_size_config(
764                                max_decoding_message_size,
765                                max_encoding_message_size,
766                            );
767                        let res = grpc.unary(method, req).await;
768                        Ok(res)
769                    };
770                    Box::pin(fut)
771                }
772                _ => {
773                    Box::pin(async move {
774                        let mut response = http::Response::new(
775                            tonic::body::Body::default(),
776                        );
777                        let headers = response.headers_mut();
778                        headers
779                            .insert(
780                                tonic::Status::GRPC_STATUS,
781                                (tonic::Code::Unimplemented as i32).into(),
782                            );
783                        headers
784                            .insert(
785                                http::header::CONTENT_TYPE,
786                                tonic::metadata::GRPC_CONTENT_TYPE,
787                            );
788                        Ok(response)
789                    })
790                }
791            }
792        }
793    }
794    impl<T> Clone for StreamServiceServer<T> {
795        fn clone(&self) -> Self {
796            let inner = self.inner.clone();
797            Self {
798                inner,
799                accept_compression_encodings: self.accept_compression_encodings,
800                send_compression_encodings: self.send_compression_encodings,
801                max_decoding_message_size: self.max_decoding_message_size,
802                max_encoding_message_size: self.max_encoding_message_size,
803            }
804        }
805    }
806    /// Generated gRPC service name
807    pub const SERVICE_NAME: &str = "stream_service.StreamService";
808    impl<T> tonic::server::NamedService for StreamServiceServer<T> {
809        const NAME: &'static str = SERVICE_NAME;
810    }
811}
risingwave_pb/stream_service.rs

risingwave_pb/
stream_service.rs
