volo_grpc/client/
mod.rs

1//! gRPC client for Volo.
2//!
3//! Users should not use this module directly.
4//! Instead, they should use the `Builder` type in the generated code.
5//!
6//! For users need to specify some options at call time, they may use [`CallOpt`].
7
8mod callopt;
9pub mod dns;
10mod meta;
11
12use std::{cell::RefCell, marker::PhantomData, sync::Arc, time::Duration};
13
14pub use callopt::CallOpt;
15pub use meta::MetaService;
16use motore::{
17    ServiceExt,
18    layer::{Identity, Layer, Stack},
19    service::{BoxCloneService, Service},
20};
21use volo::{
22    FastStr,
23    client::{MkClient, WithOptService},
24    context::{Endpoint, Role, RpcInfo},
25    discovery::Discover,
26    loadbalance::{MkLbLayer, random::WeightedRandomBalance},
27    net::Address,
28};
29
30use self::{dns::DnsResolver, layer::timeout::TimeoutLayer};
31use crate::{
32    Request, Response, Status,
33    codec::compression::CompressionEncoding,
34    context::{ClientContext, Config},
35    layer::loadbalance::LbConfig,
36    transport::ClientTransport,
37};
38pub mod layer;
39
40/// [`ClientBuilder`] provides a builder-like interface to construct a [`Client`].
41pub struct ClientBuilder<IL, OL, C, LB, T, U> {
42    http2_config: Http2Config,
43    rpc_config: Config,
44    callee_name: FastStr,
45    caller_name: FastStr,
46    // Maybe address use Arc avoid memory alloc.
47    target: Option<Address>,
48    inner_layer: IL,
49    outer_layer: OL,
50    mk_client: C,
51    mk_lb: LB,
52    _marker: PhantomData<fn(T, U)>,
53
54    #[cfg(feature = "__tls")]
55    tls_config: Option<volo::net::tls::ClientTlsConfig>,
56}
57
58impl<C, T, U>
59    ClientBuilder<
60        Identity,
61        Identity,
62        C,
63        LbConfig<WeightedRandomBalance<<DnsResolver as Discover>::Key>, DnsResolver>,
64        T,
65        U,
66    >
67{
68    /// Creates a new [`ClientBuilder`].
69    pub fn new(service_client: C, service_name: impl AsRef<str>) -> Self {
70        Self {
71            http2_config: Default::default(),
72            rpc_config: Default::default(),
73            callee_name: FastStr::new(service_name),
74            caller_name: "".into(),
75            target: None,
76            inner_layer: Identity::new(),
77            outer_layer: Identity::new(),
78            mk_client: service_client,
79            mk_lb: LbConfig::new(WeightedRandomBalance::new(), DnsResolver::default()),
80            _marker: PhantomData,
81
82            #[cfg(feature = "__tls")]
83            tls_config: None,
84        }
85    }
86}
87
88impl<IL, OL, C, LB, T, U, DISC> ClientBuilder<IL, OL, C, LbConfig<LB, DISC>, T, U> {
89    pub fn load_balance<NLB>(
90        self,
91        load_balance: NLB,
92    ) -> ClientBuilder<IL, OL, C, LbConfig<NLB, DISC>, T, U> {
93        ClientBuilder {
94            http2_config: self.http2_config,
95            rpc_config: self.rpc_config,
96            callee_name: self.callee_name,
97            caller_name: self.caller_name,
98            target: self.target,
99            inner_layer: self.inner_layer,
100            outer_layer: self.outer_layer,
101            mk_client: self.mk_client,
102            mk_lb: self.mk_lb.load_balance(load_balance),
103            _marker: PhantomData,
104
105            #[cfg(feature = "__tls")]
106            tls_config: self.tls_config,
107        }
108    }
109
110    pub fn discover<NDISC>(
111        self,
112        discover: NDISC,
113    ) -> ClientBuilder<IL, OL, C, LbConfig<LB, NDISC>, T, U> {
114        ClientBuilder {
115            http2_config: self.http2_config,
116            rpc_config: self.rpc_config,
117            callee_name: self.callee_name,
118            caller_name: self.caller_name,
119            target: self.target,
120            inner_layer: self.inner_layer,
121            outer_layer: self.outer_layer,
122            mk_client: self.mk_client,
123            mk_lb: self.mk_lb.discover(discover),
124            _marker: PhantomData,
125
126            #[cfg(feature = "__tls")]
127            tls_config: self.tls_config,
128        }
129    }
130}
131
132impl<IL, OL, C, LB, T, U> ClientBuilder<IL, OL, C, LB, T, U> {
133    /// Sets the rpc timeout for the client.
134    ///
135    /// The default value is 1 second.
136    ///
137    /// Users can set this to `None` to disable the timeout.
138    pub fn rpc_timeout(mut self, timeout: Option<Duration>) -> Self {
139        self.rpc_config.set_rpc_timeout(timeout);
140        self
141    }
142    /// Sets the `SETTINGS_INITIAL_WINDOW_SIZE` option for HTTP2
143    /// stream-level flow control.
144    ///
145    /// Default is `2MB`.
146    pub fn http2_init_stream_window_size(mut self, sz: impl Into<u32>) -> Self {
147        self.http2_config.init_stream_window_size = sz.into();
148        self
149    }
150
151    /// Sets the max connection-level flow control for HTTP2.
152    ///
153    /// Default is `5MB`.
154    pub fn http2_init_connection_window_size(mut self, sz: impl Into<u32>) -> Self {
155        self.http2_config.init_connection_window_size = sz.into();
156        self
157    }
158
159    /// Sets whether to use an adaptive flow control.
160    ///
161    /// Enabling this will override the limits set in
162    /// `http2_initial_stream_window_size` and
163    /// `http2_initial_connection_window_size`.
164    ///
165    /// Default is `false`.
166    pub fn http2_adaptive_window(mut self, enabled: bool) -> Self {
167        self.http2_config.adaptive_window = enabled;
168        self
169    }
170
171    /// Sets the maximum frame size to use for HTTP2.
172    ///
173    /// Default is `16KB`.
174    pub fn http2_max_frame_size(mut self, sz: impl Into<u32>) -> Self {
175        self.http2_config.max_frame_size = sz.into();
176        self
177    }
178
179    /// Sets an interval for HTTP2 Ping frames should be sent to keep a
180    /// connection alive.
181    ///
182    /// Default is disabled.
183    pub fn http2_keepalive_interval(mut self, interval: impl Into<Option<Duration>>) -> Self {
184        self.http2_config.http2_keepalive_interval = interval.into();
185        self
186    }
187
188    /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
189    ///
190    /// If the ping is not acknowledged within the timeout, the connection will
191    /// be closed. Does nothing if `http2_keepalive_interval` is disabled.
192    ///
193    /// Default is `20` seconds.
194    pub fn http2_keepalive_timeout(mut self, timeout: Duration) -> Self {
195        self.http2_config.http2_keepalive_timeout = timeout;
196        self
197    }
198
199    /// Sets whether HTTP2 keep-alive should apply while the connection is idle.
200    ///
201    /// If disabled, keep-alive pings are only sent while there are open
202    /// request/responses streams. If enabled, pings are also sent when no
203    /// streams are active. Does nothing if `http2_keepalive_interval` is
204    /// disabled.
205    ///
206    /// Default is `false`.
207    pub fn http2_keepalive_while_idle(mut self, enabled: bool) -> Self {
208        self.http2_config.http2_keepalive_while_idle = enabled;
209        self
210    }
211
212    /// Sets the maximum number of HTTP2 concurrent locally reset streams.
213    ///
214    /// Default is `10`.
215    pub fn http2_max_concurrent_reset_streams(mut self, sz: impl Into<usize>) -> Self {
216        self.http2_config.max_concurrent_reset_streams = sz.into();
217        self
218    }
219
220    /// Set the maximum write buffer size for each HTTP/2 stream.
221    ///
222    /// Default is currently 1MB, but may change.
223    ///
224    /// The value must be no larger than `u32::MAX`.
225    pub fn http2_max_send_buf_size(mut self, max: impl Into<usize>) -> Self {
226        self.http2_config.max_send_buf_size = max.into();
227        self
228    }
229
230    /// Sets whether to retry requests that get disrupted before ever starting
231    /// to write.
232    ///
233    /// Default is `true`.
234    #[deprecated(
235        since = "0.9.0",
236        note = "`retry_canceled_requests` has been removed in `hyper`"
237    )]
238    pub fn retry_canceled_requests(self, _enabled: bool) -> Self {
239        self
240    }
241
242    /// Sets whether the connection **must** use HTTP/2.
243    ///
244    /// Default is `false`.
245    #[deprecated(
246        since = "0.9.0",
247        note = "accepting http1 connection was not supported by `hyper`"
248    )]
249    pub fn accept_http1(self, _accept_http1: bool) -> Self {
250        self
251    }
252
253    /// Sets the timeout for connecting to a URL.
254    ///
255    /// Default is no timeout.
256    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
257        self.rpc_config.connect_timeout = Some(timeout);
258        self
259    }
260
261    /// Sets the timeout for the response.
262    ///
263    /// Default is no timeout.
264    pub fn read_timeout(mut self, timeout: Duration) -> Self {
265        self.rpc_config.read_timeout = Some(timeout);
266        self
267    }
268
269    /// Sets the timeout for the request.
270    ///
271    /// Default is no timeout.
272    pub fn write_timeout(mut self, timeout: Duration) -> Self {
273        self.rpc_config.write_timeout = Some(timeout);
274        self
275    }
276
277    /// Sets the caller name for the client.
278    ///
279    /// Default is the empty string.
280    pub fn caller_name(mut self, name: impl AsRef<str>) -> Self {
281        self.caller_name = FastStr::new(name);
282        self
283    }
284
285    /// Sets the send compression encodings for the request, and will self-adaptive with config of
286    /// the server.
287    ///
288    /// Default is disable the send compression.
289    pub fn send_compressions(mut self, config: Vec<CompressionEncoding>) -> Self {
290        self.rpc_config.send_compressions = Some(config);
291        self
292    }
293
294    /// Sets the accept compression encodings for the request, and will self-adaptive with config of
295    /// the server.
296    ///
297    /// Default is disable the accept decompression.
298    pub fn accept_compressions(mut self, config: Vec<CompressionEncoding>) -> Self {
299        self.rpc_config.accept_compressions = Some(config);
300        self
301    }
302
303    pub fn mk_load_balance<NLB>(self, mk_load_balance: NLB) -> ClientBuilder<IL, OL, C, NLB, T, U> {
304        ClientBuilder {
305            http2_config: self.http2_config,
306            rpc_config: self.rpc_config,
307            callee_name: self.callee_name,
308            caller_name: self.caller_name,
309            target: self.target,
310            inner_layer: self.inner_layer,
311            outer_layer: self.outer_layer,
312            mk_client: self.mk_client,
313            mk_lb: mk_load_balance,
314            _marker: PhantomData,
315
316            #[cfg(feature = "__tls")]
317            tls_config: self.tls_config,
318        }
319    }
320
321    /// Sets the address for the rpc call.
322    ///
323    /// If the address is set, the call will be sent to the address directly.
324    ///
325    /// The client will skip the discovery and loadbalance Service if this is set.
326    pub fn address<A: Into<Address>>(mut self, target: A) -> Self {
327        self.target = Some(target.into());
328        self
329    }
330
331    /// Adds a new inner layer to the client.
332    ///
333    /// The layer's `Service` should be `Send + Sync + Clone + 'static`.
334    ///
335    /// # Order
336    ///
337    /// Assume we already have two layers: foo and bar. We want to add a new layer baz.
338    ///
339    /// The current order is: foo -> bar (the request will come to foo first, and then bar).
340    ///
341    /// After we call `.layer_inner(baz)`, we will get: foo -> bar -> baz.
342    ///
343    /// The overall order for layers is: outer -> LoadBalance -> \[inner\] -> transport.
344    pub fn layer_inner<Inner>(
345        self,
346        layer: Inner,
347    ) -> ClientBuilder<Stack<Inner, IL>, OL, C, LB, T, U> {
348        ClientBuilder {
349            http2_config: self.http2_config,
350            rpc_config: self.rpc_config,
351            callee_name: self.callee_name,
352            caller_name: self.caller_name,
353            target: self.target,
354            inner_layer: Stack::new(layer, self.inner_layer),
355            outer_layer: self.outer_layer,
356            mk_client: self.mk_client,
357            mk_lb: self.mk_lb,
358            _marker: self._marker,
359
360            #[cfg(feature = "__tls")]
361            tls_config: self.tls_config,
362        }
363    }
364
365    /// Adds a new inner layer to the client.
366    ///
367    /// The layer's `Service` should be `Send + Sync + Clone + 'static`.
368    ///
369    /// # Order
370    ///
371    /// Assume we already have two layers: foo and bar. We want to add a new layer baz.
372    ///
373    /// The current order is: foo -> bar (the request will come to foo first, and then bar).
374    ///
375    /// After we call `.layer_inner_front(baz)`, we will get: baz -> foo -> bar.
376    ///
377    /// The overall order for layers is: outer -> LoadBalance -> \[inner\] -> transport.
378    pub fn layer_inner_front<Inner>(
379        self,
380        layer: Inner,
381    ) -> ClientBuilder<Stack<IL, Inner>, OL, C, LB, T, U> {
382        ClientBuilder {
383            http2_config: self.http2_config,
384            rpc_config: self.rpc_config,
385            callee_name: self.callee_name,
386            caller_name: self.caller_name,
387            target: self.target,
388            inner_layer: Stack::new(self.inner_layer, layer),
389            outer_layer: self.outer_layer,
390            mk_client: self.mk_client,
391            mk_lb: self.mk_lb,
392            _marker: self._marker,
393
394            #[cfg(feature = "__tls")]
395            tls_config: self.tls_config,
396        }
397    }
398
399    /// Adds a new outer layer to the client.
400    ///
401    /// The layer's `Service` should be `Send + Sync + Clone + 'static`.
402    ///
403    /// # Order
404    ///
405    /// Assume we already have two layers: foo and bar. We want to add a new layer baz.
406    ///
407    /// The current order is: foo -> bar (the request will come to foo first, and then bar).
408    ///
409    /// After we call `.layer_outer(baz)`, we will get: foo -> bar -> baz.
410    ///
411    /// The overall order for layers is: \[outer\] -> LoadBalance -> inner -> transport.
412    pub fn layer_outer<Outer>(
413        self,
414        layer: Outer,
415    ) -> ClientBuilder<IL, Stack<Outer, OL>, C, LB, T, U> {
416        ClientBuilder {
417            http2_config: self.http2_config,
418            rpc_config: self.rpc_config,
419            callee_name: self.callee_name,
420            caller_name: self.caller_name,
421            target: self.target,
422            inner_layer: self.inner_layer,
423            outer_layer: Stack::new(layer, self.outer_layer),
424            mk_client: self.mk_client,
425            mk_lb: self.mk_lb,
426            _marker: self._marker,
427
428            #[cfg(feature = "__tls")]
429            tls_config: self.tls_config,
430        }
431    }
432
433    /// Adds a new outer layer to the client.
434    ///
435    /// The layer's `Service` should be `Send + Sync + Clone + 'static`.
436    ///
437    /// # Order
438    ///
439    /// Assume we already have two layers: foo and bar. We want to add a new layer baz.
440    ///
441    /// The current order is: foo -> bar (the request will come to foo first, and then bar).
442    ///
443    /// After we call `.layer_outer_front(baz)`, we will get: baz -> foo -> bar.
444    ///
445    /// The overall order for layers is: \[outer\] -> LoadBalance -> inner -> transport.
446    pub fn layer_outer_front<Outer>(
447        self,
448        layer: Outer,
449    ) -> ClientBuilder<IL, Stack<OL, Outer>, C, LB, T, U> {
450        ClientBuilder {
451            http2_config: self.http2_config,
452            rpc_config: self.rpc_config,
453            callee_name: self.callee_name,
454            caller_name: self.caller_name,
455            target: self.target,
456            inner_layer: self.inner_layer,
457            outer_layer: Stack::new(self.outer_layer, layer),
458            mk_client: self.mk_client,
459            mk_lb: self.mk_lb,
460            _marker: self._marker,
461
462            #[cfg(feature = "__tls")]
463            tls_config: self.tls_config,
464        }
465    }
466
467    /// Sets the [`ClientTlsConfig`][ClientTlsConfig] for the client.
468    ///
469    /// [ClientTlsConfig]: volo::net::tls::ClientTlsConfig
470    #[cfg(feature = "__tls")]
471    #[cfg_attr(docsrs, doc(cfg(any(feature = "rustls", feature = "native-tls"))))]
472    pub fn tls_config(mut self, tls_config: volo::net::tls::ClientTlsConfig) -> Self {
473        self.tls_config = Some(tls_config);
474        self
475    }
476}
477
478impl<IL, OL, C, LB, T, U> ClientBuilder<IL, OL, C, LB, T, U>
479where
480    C: MkClient<Client<BoxCloneService<ClientContext, Request<T>, Response<U>, Status>>>,
481    LB: MkLbLayer,
482    LB::Layer: Layer<IL::Service>,
483    <LB::Layer as Layer<IL::Service>>::Service:
484        Service<ClientContext, Request<T>, Response = Response<U>> + 'static + Send + Clone + Sync,
485    <<LB::Layer as Layer<IL::Service>>::Service as Service<ClientContext, Request<T>>>::Error:
486        Into<Status>,
487    IL: Layer<MetaService<ClientTransport<U>>>,
488    IL::Service:
489        Service<ClientContext, Request<T>, Response = Response<U>> + 'static + Send + Clone + Sync,
490    <IL::Service as Service<ClientContext, Request<T>>>::Error: Into<Status>,
491    OL:
492        Layer<
493            BoxCloneService<
494                ClientContext,
495                Request<T>,
496                Response<U>,
497                <<LB::Layer as Layer<IL::Service>>::Service as Service<
498                    ClientContext,
499                    Request<T>,
500                >>::Error,
501            >,
502        >,
503    OL::Service:
504        Service<ClientContext, Request<T>, Response = Response<U>> + 'static + Send + Clone + Sync,
505    <OL::Service as Service<ClientContext, Request<T>>>::Error: Send + Into<Status>,
506    T: 'static + Send,
507{
508    /// Builds a new [`Client`].
509    pub fn build(self) -> C::Target {
510        #[cfg(not(feature = "__tls"))]
511        let transport =
512            MetaService::new(ClientTransport::new(&self.http2_config, &self.rpc_config));
513        #[cfg(feature = "__tls")]
514        let transport = match self.tls_config {
515            Some(tls_config) => MetaService::new(ClientTransport::new_with_tls(
516                &self.http2_config,
517                &self.rpc_config,
518                tls_config,
519            )),
520            None => MetaService::new(ClientTransport::new(&self.http2_config, &self.rpc_config)),
521        };
522
523        let transport = self.outer_layer.layer(BoxCloneService::new(
524            self.mk_lb.make().layer(self.inner_layer.layer(transport)),
525        ));
526
527        let transport = transport.map_err(|err| err.into());
528        let transport = TimeoutLayer::new().layer(transport);
529        let transport = BoxCloneService::new(transport);
530
531        self.mk_client.mk_client(Client {
532            inner: Arc::new(ClientInner {
533                callee_name: self.callee_name,
534                caller_name: self.caller_name,
535                rpc_config: self.rpc_config,
536                target: self.target,
537            }),
538            transport,
539        })
540    }
541}
542
543#[derive(Debug)]
544/// A struct indicating the rpc configuration of the client.
545struct ClientInner {
546    callee_name: FastStr,
547    caller_name: FastStr,
548    rpc_config: Config,
549    target: Option<Address>,
550}
551
552/// A client for a gRPC service.
553///
554/// `Client` is designed to "clone and use", so it's cheap to clone it.
555/// One important thing is that the `CallOpt` will not be cloned, because
556/// it's designed to be per-request.
557#[derive(Clone)]
558pub struct Client<S> {
559    transport: S,
560    inner: Arc<ClientInner>,
561}
562
563impl<S> Client<S> {
564    pub fn make_cx(&self, path: &'static str) -> ClientContext {
565        ClientContext::new(self.make_rpc_info(path))
566    }
567
568    fn make_rpc_info(&self, method: &'static str) -> RpcInfo<Config> {
569        let caller = Endpoint::new(self.inner.caller_name.clone());
570        let mut callee = Endpoint::new(self.inner.callee_name.clone());
571        if let Some(target) = &self.inner.target {
572            callee.set_address(target.clone());
573        }
574        RpcInfo::new(
575            Role::Client,
576            method.into(),
577            caller,
578            callee,
579            self.inner.rpc_config.clone(),
580        )
581    }
582
583    pub fn with_opt<Opt>(self, opt: Opt) -> Client<WithOptService<S, Opt>> {
584        Client {
585            transport: WithOptService::new(self.transport, opt),
586            inner: self.inner,
587        }
588    }
589}
590
591macro_rules! impl_client {
592    (($self: ident, &mut $cx:ident, $req: ident) => async move $e: tt ) => {
593        impl<S, Req: Send + 'static>
594            volo::service::Service<crate::context::ClientContext, Req> for Client<S>
595        where
596            S: volo::service::Service<
597                    crate::context::ClientContext,
598                    Req,
599                    Error = crate::Status,
600                > + Sync
601                + Send
602                + 'static,
603        {
604            type Response = S::Response;
605            type Error = S::Error;
606
607            async fn call(
608                &$self,
609                $cx: &mut crate::context::ClientContext,
610                $req: Req,
611            ) -> Result<Self::Response, Self::Error> {
612                $e
613            }
614        }
615
616        impl<S, Req: Send + 'static>
617            volo::client::OneShotService<crate::context::ClientContext, Req> for Client<S>
618        where
619            S: volo::client::OneShotService<
620                    crate::context::ClientContext,
621                    Req,
622                    Error = crate::Status,
623                > + Sync
624                + Send
625                + 'static,
626        {
627            type Response = S::Response;
628            type Error = S::Error;
629
630            async fn call(
631                $self,
632                $cx: &mut crate::context::ClientContext,
633                $req: Req,
634            ) -> Result<Self::Response, Self::Error> {
635                $e
636            }
637        }
638    };
639}
640
641impl_client!((self, &mut cx, req) => async move {
642    let has_metainfo = metainfo::METAINFO.try_with(|_| {}).is_ok();
643
644    let mk_call = async { self.transport.call(cx, req).await };
645
646    if has_metainfo {
647        mk_call.await
648    } else {
649        metainfo::METAINFO
650            .scope(RefCell::new(metainfo::MetaInfo::default()), mk_call)
651            .await
652    }
653});
654
655const DEFAULT_STREAM_WINDOW_SIZE: u32 = 1024 * 1024 * 2; // 2MB
656const DEFAULT_CONN_WINDOW_SIZE: u32 = 1024 * 1024 * 5; // 5MB
657const DEFAULT_MAX_FRAME_SIZE: u32 = 1024 * 16; // 16KB
658const DEFAULT_MAX_SEND_BUF_SIZE: usize = 1024 * 1024; // 1MB
659const DEFAULT_KEEPALIVE_TIMEOUT_SECS: Duration = Duration::from_secs(20); // 20s
660const DEFAULT_MAX_CONCURRENT_RESET_STREAMS: usize = 10;
661
662/// Configuration for the underlying h2 connection.
663#[derive(Debug, Clone, Copy)]
664pub struct Http2Config {
665    pub(crate) init_stream_window_size: u32,
666    pub(crate) init_connection_window_size: u32,
667    pub(crate) adaptive_window: bool,
668    pub(crate) max_frame_size: u32,
669    pub(crate) http2_keepalive_interval: Option<Duration>,
670    pub(crate) http2_keepalive_timeout: Duration,
671    pub(crate) http2_keepalive_while_idle: bool,
672    pub(crate) max_concurrent_reset_streams: usize,
673    pub(crate) max_send_buf_size: usize,
674}
675
676impl Default for Http2Config {
677    fn default() -> Self {
678        Self {
679            init_stream_window_size: DEFAULT_STREAM_WINDOW_SIZE,
680            init_connection_window_size: DEFAULT_CONN_WINDOW_SIZE,
681            adaptive_window: false,
682            max_frame_size: DEFAULT_MAX_FRAME_SIZE,
683            http2_keepalive_interval: None,
684            http2_keepalive_timeout: DEFAULT_KEEPALIVE_TIMEOUT_SECS,
685            http2_keepalive_while_idle: false,
686            max_concurrent_reset_streams: DEFAULT_MAX_CONCURRENT_RESET_STREAMS,
687            max_send_buf_size: DEFAULT_MAX_SEND_BUF_SIZE,
688        }
689    }
690}
volo_grpc/client/mod.rs

volo_grpc/client/
mod.rs
