tower/util/mod.rs
1//! Various utility types and functions that are generally used with Tower.
2
3mod and_then;
4mod boxed;
5mod boxed_clone;
6mod boxed_clone_sync;
7mod call_all;
8mod either;
9
10mod future_service;
11mod map_err;
12mod map_request;
13mod map_response;
14mod map_result;
15
16mod map_future;
17mod oneshot;
18mod optional;
19mod optional_layer;
20mod ready;
21mod service_fn;
22mod then;
23
24pub mod rng;
25
26pub use self::{
27 and_then::{AndThen, AndThenLayer},
28 boxed::{
29 BoxCloneServiceLayer, BoxCloneSyncServiceLayer, BoxLayer, BoxService, UnsyncBoxService,
30 },
31 boxed_clone::BoxCloneService,
32 boxed_clone_sync::BoxCloneSyncService,
33 either::Either,
34 future_service::{future_service, FutureService},
35 map_err::{MapErr, MapErrLayer},
36 map_future::{MapFuture, MapFutureLayer},
37 map_request::{MapRequest, MapRequestLayer},
38 map_response::{MapResponse, MapResponseLayer},
39 map_result::{MapResult, MapResultLayer},
40 oneshot::Oneshot,
41 optional::Optional,
42 optional_layer::{OptionLayer, OptionService},
43 ready::{Ready, ReadyOneshot},
44 service_fn::{service_fn, ServiceFn},
45 then::{Then, ThenLayer},
46};
47
48pub use self::call_all::{CallAll, CallAllUnordered};
49use std::future::Future;
50
51#[cfg(feature = "buffer")]
52use crate::buffer::Buffer;
53
54#[cfg(feature = "retry")]
55use crate::retry::Retry;
56
57pub mod error {
58 //! Error types
59
60 pub use super::optional::error as optional;
61}
62
63pub mod future {
64 //! Future types
65
66 pub use super::and_then::AndThenFuture;
67 pub use super::either::EitherResponseFuture;
68 pub use super::map_err::MapErrFuture;
69 pub use super::map_response::MapResponseFuture;
70 pub use super::map_result::MapResultFuture;
71 pub use super::optional::future as optional;
72 pub use super::optional_layer::ResponseFuture as OptionResponseFuture;
73 pub use super::then::ThenFuture;
74}
75
76/// An extension trait for `Service`s that provides a variety of convenient
77/// adapters
78pub trait ServiceExt<Request>: tower_service::Service<Request> {
79 /// Yields a mutable reference to the service when it is ready to accept a request.
80 fn ready(&mut self) -> Ready<'_, Self, Request>
81 where
82 Self: Sized,
83 {
84 Ready::new(self)
85 }
86
87 /// Yields the service when it is ready to accept a request.
88 fn ready_oneshot(self) -> ReadyOneshot<Self, Request>
89 where
90 Self: Sized,
91 {
92 ReadyOneshot::new(self)
93 }
94
95 /// Consume this `Service`, calling it with the provided request once it is ready.
96 fn oneshot(self, req: Request) -> Oneshot<Self, Request>
97 where
98 Self: Sized,
99 {
100 Oneshot::new(self, req)
101 }
102
103 /// Process all requests from the given [`Stream`], and produce a [`Stream`] of their responses.
104 ///
105 /// This is essentially [`Stream<Item = Request>`][stream] + `Self` => [`Stream<Item =
106 /// Response>`][stream]. See the documentation for [`CallAll`] for
107 /// details.
108 ///
109 /// [`Stream`]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html
110 /// [stream]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html
111 fn call_all<S>(self, reqs: S) -> CallAll<Self, S>
112 where
113 Self: Sized,
114 S: futures_core::Stream<Item = Request>,
115 {
116 CallAll::new(self, reqs)
117 }
118
119 /// Executes a new future after this service's future resolves. This does
120 /// not alter the behaviour of the [`poll_ready`] method.
121 ///
122 /// This method can be used to change the [`Response`] type of the service
123 /// into a different type. You can use this method to chain along a computation once the
124 /// service's response has been resolved.
125 ///
126 /// [`Response`]: crate::Service::Response
127 /// [`poll_ready`]: crate::Service::poll_ready
128 ///
129 /// # Example
130 /// ```
131 /// # use std::task::{Poll, Context};
132 /// # use tower::{Service, ServiceExt};
133 /// #
134 /// # struct DatabaseService;
135 /// # impl DatabaseService {
136 /// # fn new(address: &str) -> Self {
137 /// # DatabaseService
138 /// # }
139 /// # }
140 /// #
141 /// # struct Record {
142 /// # pub name: String,
143 /// # pub age: u16
144 /// # }
145 /// #
146 /// # impl Service<u32> for DatabaseService {
147 /// # type Response = Record;
148 /// # type Error = u8;
149 /// # type Future = std::future::Ready<Result<Record, u8>>;
150 /// #
151 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
152 /// # Poll::Ready(Ok(()))
153 /// # }
154 /// #
155 /// # fn call(&mut self, request: u32) -> Self::Future {
156 /// # std::future::ready(Ok(Record { name: "Jack".into(), age: 32 }))
157 /// # }
158 /// # }
159 /// #
160 /// # async fn avatar_lookup(name: String) -> Result<Vec<u8>, u8> { Ok(vec![]) }
161 /// #
162 /// # fn main() {
163 /// # async {
164 /// // A service returning Result<Record, _>
165 /// let service = DatabaseService::new("127.0.0.1:8080");
166 ///
167 /// // Map the response into a new response
168 /// let mut new_service = service.and_then(|record: Record| async move {
169 /// let name = record.name;
170 /// avatar_lookup(name).await
171 /// });
172 ///
173 /// // Call the new service
174 /// let id = 13;
175 /// let avatar = new_service.call(id).await.unwrap();
176 /// # };
177 /// # }
178 /// ```
179 fn and_then<F>(self, f: F) -> AndThen<Self, F>
180 where
181 Self: Sized,
182 F: Clone,
183 {
184 AndThen::new(self, f)
185 }
186
187 /// Maps this service's response value to a different value. This does not
188 /// alter the behaviour of the [`poll_ready`] method.
189 ///
190 /// This method can be used to change the [`Response`] type of the service
191 /// into a different type. It is similar to the [`Result::map`]
192 /// method. You can use this method to chain along a computation once the
193 /// service's response has been resolved.
194 ///
195 /// [`Response`]: crate::Service::Response
196 /// [`poll_ready`]: crate::Service::poll_ready
197 ///
198 /// # Example
199 /// ```
200 /// # use std::task::{Poll, Context};
201 /// # use tower::{Service, ServiceExt};
202 /// #
203 /// # struct DatabaseService;
204 /// # impl DatabaseService {
205 /// # fn new(address: &str) -> Self {
206 /// # DatabaseService
207 /// # }
208 /// # }
209 /// #
210 /// # struct Record {
211 /// # pub name: String,
212 /// # pub age: u16
213 /// # }
214 /// #
215 /// # impl Service<u32> for DatabaseService {
216 /// # type Response = Record;
217 /// # type Error = u8;
218 /// # type Future = std::future::Ready<Result<Record, u8>>;
219 /// #
220 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
221 /// # Poll::Ready(Ok(()))
222 /// # }
223 /// #
224 /// # fn call(&mut self, request: u32) -> Self::Future {
225 /// # std::future::ready(Ok(Record { name: "Jack".into(), age: 32 }))
226 /// # }
227 /// # }
228 /// #
229 /// # fn main() {
230 /// # async {
231 /// // A service returning Result<Record, _>
232 /// let service = DatabaseService::new("127.0.0.1:8080");
233 ///
234 /// // Map the response into a new response
235 /// let mut new_service = service.map_response(|record| record.name);
236 ///
237 /// // Call the new service
238 /// let id = 13;
239 /// let name = new_service
240 /// .ready()
241 /// .await?
242 /// .call(id)
243 /// .await?;
244 /// # Ok::<(), u8>(())
245 /// # };
246 /// # }
247 /// ```
248 fn map_response<F, Response>(self, f: F) -> MapResponse<Self, F>
249 where
250 Self: Sized,
251 F: FnOnce(Self::Response) -> Response + Clone,
252 {
253 MapResponse::new(self, f)
254 }
255
256 /// Maps this service's error value to a different value. This does not
257 /// alter the behaviour of the [`poll_ready`] method.
258 ///
259 /// This method can be used to change the [`Error`] type of the service
260 /// into a different type. It is similar to the [`Result::map_err`] method.
261 ///
262 /// [`Error`]: crate::Service::Error
263 /// [`poll_ready`]: crate::Service::poll_ready
264 ///
265 /// # Example
266 /// ```
267 /// # use std::task::{Poll, Context};
268 /// # use tower::{Service, ServiceExt};
269 /// #
270 /// # struct DatabaseService;
271 /// # impl DatabaseService {
272 /// # fn new(address: &str) -> Self {
273 /// # DatabaseService
274 /// # }
275 /// # }
276 /// #
277 /// # struct Error {
278 /// # pub code: u32,
279 /// # pub message: String
280 /// # }
281 /// #
282 /// # impl Service<u32> for DatabaseService {
283 /// # type Response = String;
284 /// # type Error = Error;
285 /// # type Future = std::future::Ready<Result<String, Error>>;
286 /// #
287 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
288 /// # Poll::Ready(Ok(()))
289 /// # }
290 /// #
291 /// # fn call(&mut self, request: u32) -> Self::Future {
292 /// # std::future::ready(Ok(String::new()))
293 /// # }
294 /// # }
295 /// #
296 /// # fn main() {
297 /// # async {
298 /// // A service returning Result<_, Error>
299 /// let service = DatabaseService::new("127.0.0.1:8080");
300 ///
301 /// // Map the error to a new error
302 /// let mut new_service = service.map_err(|err| err.code);
303 ///
304 /// // Call the new service
305 /// let id = 13;
306 /// let code = new_service
307 /// .ready()
308 /// .await?
309 /// .call(id)
310 /// .await
311 /// .unwrap_err();
312 /// # Ok::<(), u32>(())
313 /// # };
314 /// # }
315 /// ```
316 fn map_err<F, Error>(self, f: F) -> MapErr<Self, F>
317 where
318 Self: Sized,
319 F: FnOnce(Self::Error) -> Error + Clone,
320 {
321 MapErr::new(self, f)
322 }
323
324 /// Maps this service's result type (`Result<Self::Response, Self::Error>`)
325 /// to a different value, regardless of whether the future succeeds or
326 /// fails.
327 ///
328 /// This is similar to the [`map_response`] and [`map_err`] combinators,
329 /// except that the *same* function is invoked when the service's future
330 /// completes, whether it completes successfully or fails. This function
331 /// takes the [`Result`] returned by the service's future, and returns a
332 /// [`Result`].
333 ///
334 /// Like the standard library's [`Result::and_then`], this method can be
335 /// used to implement control flow based on `Result` values. For example, it
336 /// may be used to implement error recovery, by turning some [`Err`]
337 /// responses from the service into [`Ok`] responses. Similarly, some
338 /// successful responses from the service could be rejected, by returning an
339 /// [`Err`] conditionally, depending on the value inside the [`Ok`]. Finally,
340 /// this method can also be used to implement behaviors that must run when a
341 /// service's future completes, regardless of whether it succeeded or failed.
342 ///
343 /// This method can be used to change the [`Response`] type of the service
344 /// into a different type. It can also be used to change the [`Error`] type
345 /// of the service. However, because the [`map_result`] function is not applied
346 /// to the errors returned by the service's [`poll_ready`] method, it must
347 /// be possible to convert the service's [`Error`] type into the error type
348 /// returned by the [`map_result`] function. This is trivial when the function
349 /// returns the same error type as the service, but in other cases, it can
350 /// be useful to use [`BoxError`] to erase differing error types.
351 ///
352 /// # Examples
353 ///
354 /// Recovering from certain errors:
355 ///
356 /// ```
357 /// # use std::task::{Poll, Context};
358 /// # use tower::{Service, ServiceExt};
359 /// #
360 /// # struct DatabaseService;
361 /// # impl DatabaseService {
362 /// # fn new(address: &str) -> Self {
363 /// # DatabaseService
364 /// # }
365 /// # }
366 /// #
367 /// # struct Record {
368 /// # pub name: String,
369 /// # pub age: u16
370 /// # }
371 /// # #[derive(Debug)]
372 /// # enum DbError {
373 /// # Parse(std::num::ParseIntError),
374 /// # NoRecordsFound,
375 /// # }
376 /// #
377 /// # impl Service<u32> for DatabaseService {
378 /// # type Response = Vec<Record>;
379 /// # type Error = DbError;
380 /// # type Future = std::future::Ready<Result<Vec<Record>, DbError>>;
381 /// #
382 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
383 /// # Poll::Ready(Ok(()))
384 /// # }
385 /// #
386 /// # fn call(&mut self, request: u32) -> Self::Future {
387 /// # std::future::ready(Ok(vec![Record { name: "Jack".into(), age: 32 }]))
388 /// # }
389 /// # }
390 /// #
391 /// # fn main() {
392 /// # async {
393 /// // A service returning Result<Vec<Record>, DbError>
394 /// let service = DatabaseService::new("127.0.0.1:8080");
395 ///
396 /// // If the database returns no records for the query, we just want an empty `Vec`.
397 /// let mut new_service = service.map_result(|result| match result {
398 /// // If the error indicates that no records matched the query, return an empty
399 /// // `Vec` instead.
400 /// Err(DbError::NoRecordsFound) => Ok(Vec::new()),
401 /// // Propagate all other responses (`Ok` and `Err`) unchanged
402 /// x => x,
403 /// });
404 ///
405 /// // Call the new service
406 /// let id = 13;
407 /// let name = new_service
408 /// .ready()
409 /// .await?
410 /// .call(id)
411 /// .await?;
412 /// # Ok::<(), DbError>(())
413 /// # };
414 /// # }
415 /// ```
416 ///
417 /// Rejecting some `Ok` responses:
418 ///
419 /// ```
420 /// # use std::task::{Poll, Context};
421 /// # use tower::{Service, ServiceExt};
422 /// #
423 /// # struct DatabaseService;
424 /// # impl DatabaseService {
425 /// # fn new(address: &str) -> Self {
426 /// # DatabaseService
427 /// # }
428 /// # }
429 /// #
430 /// # struct Record {
431 /// # pub name: String,
432 /// # pub age: u16
433 /// # }
434 /// # type DbError = String;
435 /// # type AppError = String;
436 /// #
437 /// # impl Service<u32> for DatabaseService {
438 /// # type Response = Record;
439 /// # type Error = DbError;
440 /// # type Future = std::future::Ready<Result<Record, DbError>>;
441 /// #
442 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
443 /// # Poll::Ready(Ok(()))
444 /// # }
445 /// #
446 /// # fn call(&mut self, request: u32) -> Self::Future {
447 /// # std::future::ready(Ok(Record { name: "Jack".into(), age: 32 }))
448 /// # }
449 /// # }
450 /// #
451 /// # fn main() {
452 /// # async {
453 /// use tower::BoxError;
454 ///
455 /// // A service returning Result<Record, DbError>
456 /// let service = DatabaseService::new("127.0.0.1:8080");
457 ///
458 /// // If the user is zero years old, return an error.
459 /// let mut new_service = service.map_result(|result| {
460 /// let record = result?;
461 ///
462 /// if record.age == 0 {
463 /// // Users must have been born to use our app!
464 /// let app_error = AppError::from("users cannot be 0 years old!");
465 ///
466 /// // Box the error to erase its type (as it can be an `AppError`
467 /// // *or* the inner service's `DbError`).
468 /// return Err(BoxError::from(app_error));
469 /// }
470 ///
471 /// // Otherwise, return the record.
472 /// Ok(record)
473 /// });
474 ///
475 /// // Call the new service
476 /// let id = 13;
477 /// let record = new_service
478 /// .ready()
479 /// .await?
480 /// .call(id)
481 /// .await?;
482 /// # Ok::<(), BoxError>(())
483 /// # };
484 /// # }
485 /// ```
486 ///
487 /// Performing an action that must be run for both successes and failures:
488 ///
489 /// ```
490 /// # use std::convert::TryFrom;
491 /// # use std::task::{Poll, Context};
492 /// # use tower::{Service, ServiceExt};
493 /// #
494 /// # struct DatabaseService;
495 /// # impl DatabaseService {
496 /// # fn new(address: &str) -> Self {
497 /// # DatabaseService
498 /// # }
499 /// # }
500 /// #
501 /// # impl Service<u32> for DatabaseService {
502 /// # type Response = String;
503 /// # type Error = u8;
504 /// # type Future = std::future::Ready<Result<String, u8>>;
505 /// #
506 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
507 /// # Poll::Ready(Ok(()))
508 /// # }
509 /// #
510 /// # fn call(&mut self, request: u32) -> Self::Future {
511 /// # std::future::ready(Ok(String::new()))
512 /// # }
513 /// # }
514 /// #
515 /// # fn main() {
516 /// # async {
517 /// // A service returning Result<Record, DbError>
518 /// let service = DatabaseService::new("127.0.0.1:8080");
519 ///
520 /// // Print a message whenever a query completes.
521 /// let mut new_service = service.map_result(|result| {
522 /// println!("query completed; success={}", result.is_ok());
523 /// result
524 /// });
525 ///
526 /// // Call the new service
527 /// let id = 13;
528 /// let response = new_service
529 /// .ready()
530 /// .await?
531 /// .call(id)
532 /// .await;
533 /// # response
534 /// # };
535 /// # }
536 /// ```
537 ///
538 /// [`map_response`]: ServiceExt::map_response
539 /// [`map_err`]: ServiceExt::map_err
540 /// [`map_result`]: ServiceExt::map_result
541 /// [`Error`]: crate::Service::Error
542 /// [`Response`]: crate::Service::Response
543 /// [`poll_ready`]: crate::Service::poll_ready
544 /// [`BoxError`]: crate::BoxError
545 fn map_result<F, Response, Error>(self, f: F) -> MapResult<Self, F>
546 where
547 Self: Sized,
548 Error: From<Self::Error>,
549 F: FnOnce(Result<Self::Response, Self::Error>) -> Result<Response, Error> + Clone,
550 {
551 MapResult::new(self, f)
552 }
553
554 /// Composes a function *in front of* the service.
555 ///
556 /// This adapter produces a new service that passes each value through the
557 /// given function `f` before sending it to `self`.
558 ///
559 /// # Example
560 /// ```
561 /// # use std::convert::TryFrom;
562 /// # use std::task::{Poll, Context};
563 /// # use tower::{Service, ServiceExt};
564 /// #
565 /// # struct DatabaseService;
566 /// # impl DatabaseService {
567 /// # fn new(address: &str) -> Self {
568 /// # DatabaseService
569 /// # }
570 /// # }
571 /// #
572 /// # impl Service<String> for DatabaseService {
573 /// # type Response = String;
574 /// # type Error = u8;
575 /// # type Future = std::future::Ready<Result<String, u8>>;
576 /// #
577 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
578 /// # Poll::Ready(Ok(()))
579 /// # }
580 /// #
581 /// # fn call(&mut self, request: String) -> Self::Future {
582 /// # std::future::ready(Ok(String::new()))
583 /// # }
584 /// # }
585 /// #
586 /// # fn main() {
587 /// # async {
588 /// // A service taking a String as a request
589 /// let service = DatabaseService::new("127.0.0.1:8080");
590 ///
591 /// // Map the request to a new request
592 /// let mut new_service = service.map_request(|id: u32| id.to_string());
593 ///
594 /// // Call the new service
595 /// let id = 13;
596 /// let response = new_service
597 /// .ready()
598 /// .await?
599 /// .call(id)
600 /// .await;
601 /// # response
602 /// # };
603 /// # }
604 /// ```
605 fn map_request<F, NewRequest>(self, f: F) -> MapRequest<Self, F>
606 where
607 Self: Sized,
608 F: FnMut(NewRequest) -> Request,
609 {
610 MapRequest::new(self, f)
611 }
612
613 /// Composes this service with a [`Filter`] that conditionally accepts or
614 /// rejects requests based on a [predicate].
615 ///
616 /// This adapter produces a new service that passes each value through the
617 /// given function `predicate` before sending it to `self`.
618 ///
619 /// # Example
620 /// ```
621 /// # use std::convert::TryFrom;
622 /// # use std::task::{Poll, Context};
623 /// # use tower::{Service, ServiceExt};
624 /// #
625 /// # struct DatabaseService;
626 /// # impl DatabaseService {
627 /// # fn new(address: &str) -> Self {
628 /// # DatabaseService
629 /// # }
630 /// # }
631 /// #
632 /// # #[derive(Debug)] enum DbError {
633 /// # Parse(std::num::ParseIntError)
634 /// # }
635 /// #
636 /// # impl std::fmt::Display for DbError {
637 /// # fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { std::fmt::Debug::fmt(self, f) }
638 /// # }
639 /// # impl std::error::Error for DbError {}
640 /// # impl Service<u32> for DatabaseService {
641 /// # type Response = String;
642 /// # type Error = DbError;
643 /// # type Future = std::future::Ready<Result<String, DbError>>;
644 /// #
645 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
646 /// # Poll::Ready(Ok(()))
647 /// # }
648 /// #
649 /// # fn call(&mut self, request: u32) -> Self::Future {
650 /// # std::future::ready(Ok(String::new()))
651 /// # }
652 /// # }
653 /// #
654 /// # fn main() {
655 /// # async {
656 /// // A service taking a u32 as a request and returning Result<_, DbError>
657 /// let service = DatabaseService::new("127.0.0.1:8080");
658 ///
659 /// // Fallibly map the request to a new request
660 /// let mut new_service = service
661 /// .filter(|id_str: &str| id_str.parse().map_err(DbError::Parse));
662 ///
663 /// // Call the new service
664 /// let id = "13";
665 /// let response = new_service
666 /// .ready()
667 /// .await?
668 /// .call(id)
669 /// .await;
670 /// # response
671 /// # };
672 /// # }
673 /// ```
674 ///
675 /// [`Filter`]: crate::filter::Filter
676 /// [predicate]: crate::filter::Predicate
677 #[cfg(feature = "filter")]
678 fn filter<F, NewRequest>(self, filter: F) -> crate::filter::Filter<Self, F>
679 where
680 Self: Sized,
681 F: crate::filter::Predicate<NewRequest>,
682 {
683 crate::filter::Filter::new(self, filter)
684 }
685
686 /// Composes this service with an [`AsyncFilter`] that conditionally accepts or
687 /// rejects requests based on an [async predicate].
688 ///
689 /// This adapter produces a new service that passes each value through the
690 /// given function `predicate` before sending it to `self`.
691 ///
692 /// # Example
693 /// ```
694 /// # use std::convert::TryFrom;
695 /// # use std::task::{Poll, Context};
696 /// # use tower::{Service, ServiceExt};
697 /// #
698 /// # #[derive(Clone)] struct DatabaseService;
699 /// # impl DatabaseService {
700 /// # fn new(address: &str) -> Self {
701 /// # DatabaseService
702 /// # }
703 /// # }
704 /// # #[derive(Debug)]
705 /// # enum DbError {
706 /// # Rejected
707 /// # }
708 /// # impl std::fmt::Display for DbError {
709 /// # fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { std::fmt::Debug::fmt(self, f) }
710 /// # }
711 /// # impl std::error::Error for DbError {}
712 /// #
713 /// # impl Service<u32> for DatabaseService {
714 /// # type Response = String;
715 /// # type Error = DbError;
716 /// # type Future = std::future::Ready<Result<String, DbError>>;
717 /// #
718 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
719 /// # Poll::Ready(Ok(()))
720 /// # }
721 /// #
722 /// # fn call(&mut self, request: u32) -> Self::Future {
723 /// # std::future::ready(Ok(String::new()))
724 /// # }
725 /// # }
726 /// #
727 /// # fn main() {
728 /// # async {
729 /// // A service taking a u32 as a request and returning Result<_, DbError>
730 /// let service = DatabaseService::new("127.0.0.1:8080");
731 ///
732 /// /// Returns `true` if we should query the database for an ID.
733 /// async fn should_query(id: u32) -> bool {
734 /// // ...
735 /// # true
736 /// }
737 ///
738 /// // Filter requests based on `should_query`.
739 /// let mut new_service = service
740 /// .filter_async(|id: u32| async move {
741 /// if should_query(id).await {
742 /// return Ok(id);
743 /// }
744 ///
745 /// Err(DbError::Rejected)
746 /// });
747 ///
748 /// // Call the new service
749 /// let id = 13;
750 /// # let id: u32 = id;
751 /// let response = new_service
752 /// .ready()
753 /// .await?
754 /// .call(id)
755 /// .await;
756 /// # response
757 /// # };
758 /// # }
759 /// ```
760 ///
761 /// [`AsyncFilter`]: crate::filter::AsyncFilter
762 /// [asynchronous predicate]: crate::filter::AsyncPredicate
763 #[cfg(feature = "filter")]
764 fn filter_async<F, NewRequest>(self, filter: F) -> crate::filter::AsyncFilter<Self, F>
765 where
766 Self: Sized,
767 F: crate::filter::AsyncPredicate<NewRequest>,
768 {
769 crate::filter::AsyncFilter::new(self, filter)
770 }
771
772 /// Composes an asynchronous function *after* this service.
773 ///
774 /// This takes a function or closure returning a future, and returns a new
775 /// `Service` that chains that function after this service's [`Future`]. The
776 /// new `Service`'s future will consist of this service's future, followed
777 /// by the future returned by calling the chained function with the future's
778 /// [`Output`] type. The chained function is called regardless of whether
779 /// this service's future completes with a successful response or with an
780 /// error.
781 ///
782 /// This method can be thought of as an equivalent to the [`futures`
783 /// crate]'s [`FutureExt::then`] combinator, but acting on `Service`s that
784 /// _return_ futures, rather than on an individual future. Similarly to that
785 /// combinator, [`ServiceExt::then`] can be used to implement asynchronous
786 /// error recovery, by calling some asynchronous function with errors
787 /// returned by this service. Alternatively, it may also be used to call a
788 /// fallible async function with the successful response of this service.
789 ///
790 /// This method can be used to change the [`Response`] type of the service
791 /// into a different type. It can also be used to change the [`Error`] type
792 /// of the service. However, because the `then` function is not applied
793 /// to the errors returned by the service's [`poll_ready`] method, it must
794 /// be possible to convert the service's [`Error`] type into the error type
795 /// returned by the `then` future. This is trivial when the function
796 /// returns the same error type as the service, but in other cases, it can
797 /// be useful to use [`BoxError`] to erase differing error types.
798 ///
799 /// # Examples
800 ///
801 /// ```
802 /// # use std::task::{Poll, Context};
803 /// # use tower::{Service, ServiceExt};
804 /// #
805 /// # struct DatabaseService;
806 /// # impl DatabaseService {
807 /// # fn new(address: &str) -> Self {
808 /// # DatabaseService
809 /// # }
810 /// # }
811 /// #
812 /// # type Record = ();
813 /// # type DbError = ();
814 /// #
815 /// # impl Service<u32> for DatabaseService {
816 /// # type Response = Record;
817 /// # type Error = DbError;
818 /// # type Future = std::future::Ready<Result<Record, DbError>>;
819 /// #
820 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
821 /// # Poll::Ready(Ok(()))
822 /// # }
823 /// #
824 /// # fn call(&mut self, request: u32) -> Self::Future {
825 /// # std::future::ready(Ok(()))
826 /// # }
827 /// # }
828 /// #
829 /// # fn main() {
830 /// // A service returning Result<Record, DbError>
831 /// let service = DatabaseService::new("127.0.0.1:8080");
832 ///
833 /// // An async function that attempts to recover from errors returned by the
834 /// // database.
835 /// async fn recover_from_error(error: DbError) -> Result<Record, DbError> {
836 /// // ...
837 /// # Ok(())
838 /// }
839 /// # async {
840 ///
841 /// // If the database service returns an error, attempt to recover by
842 /// // calling `recover_from_error`. Otherwise, return the successful response.
843 /// let mut new_service = service.then(|result| async move {
844 /// match result {
845 /// Ok(record) => Ok(record),
846 /// Err(e) => recover_from_error(e).await,
847 /// }
848 /// });
849 ///
850 /// // Call the new service
851 /// let id = 13;
852 /// let record = new_service
853 /// .ready()
854 /// .await?
855 /// .call(id)
856 /// .await?;
857 /// # Ok::<(), DbError>(())
858 /// # };
859 /// # }
860 /// ```
861 ///
862 /// [`Future`]: crate::Service::Future
863 /// [`Output`]: std::future::Future::Output
864 /// [`futures` crate]: https://docs.rs/futures
865 /// [`FutureExt::then`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html#method.then
866 /// [`Error`]: crate::Service::Error
867 /// [`Response`]: crate::Service::Response
868 /// [`poll_ready`]: crate::Service::poll_ready
869 /// [`BoxError`]: crate::BoxError
870 fn then<F, Response, Error, Fut>(self, f: F) -> Then<Self, F>
871 where
872 Self: Sized,
873 Error: From<Self::Error>,
874 F: FnOnce(Result<Self::Response, Self::Error>) -> Fut + Clone,
875 Fut: Future<Output = Result<Response, Error>>,
876 {
877 Then::new(self, f)
878 }
879
880 /// Composes a function that transforms futures produced by the service.
881 ///
882 /// This takes a function or closure returning a future computed from the future returned by
883 /// the service's [`call`] method, as opposed to the responses produced by the future.
884 ///
885 /// # Examples
886 ///
887 /// ```
888 /// # use std::task::{Poll, Context};
889 /// # use tower::{Service, ServiceExt, BoxError};
890 /// #
891 /// # struct DatabaseService;
892 /// # impl DatabaseService {
893 /// # fn new(address: &str) -> Self {
894 /// # DatabaseService
895 /// # }
896 /// # }
897 /// #
898 /// # type Record = ();
899 /// # type DbError = crate::BoxError;
900 /// #
901 /// # impl Service<u32> for DatabaseService {
902 /// # type Response = Record;
903 /// # type Error = DbError;
904 /// # type Future = std::future::Ready<Result<Record, DbError>>;
905 /// #
906 /// # fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
907 /// # Poll::Ready(Ok(()))
908 /// # }
909 /// #
910 /// # fn call(&mut self, request: u32) -> Self::Future {
911 /// # std::future::ready(Ok(()))
912 /// # }
913 /// # }
914 /// #
915 /// # fn main() {
916 /// use std::time::Duration;
917 /// use tokio::time::timeout;
918 ///
919 /// // A service returning Result<Record, DbError>
920 /// let service = DatabaseService::new("127.0.0.1:8080");
921 /// # async {
922 ///
923 /// let mut new_service = service.map_future(|future| async move {
924 /// let res = timeout(Duration::from_secs(1), future).await?;
925 /// Ok::<_, BoxError>(res)
926 /// });
927 ///
928 /// // Call the new service
929 /// let id = 13;
930 /// let record = new_service
931 /// .ready()
932 /// .await?
933 /// .call(id)
934 /// .await?;
935 /// # Ok::<(), BoxError>(())
936 /// # };
937 /// # }
938 /// ```
939 ///
940 /// Note that normally you wouldn't implement timeouts like this and instead use [`Timeout`].
941 ///
942 /// [`call`]: crate::Service::call
943 /// [`Timeout`]: crate::timeout::Timeout
944 fn map_future<F, Fut, Response, Error>(self, f: F) -> MapFuture<Self, F>
945 where
946 Self: Sized,
947 F: FnMut(Self::Future) -> Fut,
948 Error: From<Self::Error>,
949 Fut: Future<Output = Result<Response, Error>>,
950 {
951 MapFuture::new(self, f)
952 }
953
954 /// Returns a buffered version of this service.
955 ///
956 /// See [`Buffer::new()`] for the details.
957 #[cfg(feature = "buffer")]
958 fn buffered(self, bound: usize) -> Buffer<Request, Self::Future>
959 where
960 Self: Send + Sized + 'static,
961 Self::Future: Send,
962 Self::Error: Into<crate::BoxError> + Send + Sync,
963 Request: Send + Sized + 'static,
964 {
965 Buffer::new(self, bound)
966 }
967
968 /// Returns a retry version of this service.
969 ///
970 /// See [`Retry::new()`] for the details.
971 #[cfg(feature = "retry")]
972 fn retry<P>(self, policy: P) -> Retry<P, Self>
973 where
974 Self: Sized,
975 {
976 Retry::new(policy, self)
977 }
978
979 /// Convert the service into a [`Service`] + [`Send`] trait object.
980 ///
981 /// See [`BoxService`] for more details.
982 ///
983 /// If `Self` implements the [`Clone`] trait, the [`boxed_clone`] method
984 /// can be used instead, to produce a boxed service which will also
985 /// implement [`Clone`].
986 ///
987 /// # Example
988 ///
989 /// ```
990 /// use tower::{Service, ServiceExt, BoxError, service_fn, util::BoxService};
991 /// #
992 /// # struct Request;
993 /// # struct Response;
994 /// # impl Response {
995 /// # fn new() -> Self { Self }
996 /// # }
997 ///
998 /// let service = service_fn(|req: Request| async {
999 /// Ok::<_, BoxError>(Response::new())
1000 /// });
1001 ///
1002 /// let service: BoxService<Request, Response, BoxError> = service
1003 /// .map_request(|req| {
1004 /// println!("received request");
1005 /// req
1006 /// })
1007 /// .map_response(|res| {
1008 /// println!("response produced");
1009 /// res
1010 /// })
1011 /// .boxed();
1012 /// # let service = assert_service(service);
1013 /// # fn assert_service<S, R>(svc: S) -> S
1014 /// # where S: Service<R> { svc }
1015 /// ```
1016 ///
1017 /// [`Service`]: crate::Service
1018 /// [`boxed_clone`]: Self::boxed_clone
1019 fn boxed(self) -> BoxService<Request, Self::Response, Self::Error>
1020 where
1021 Self: Sized + Send + 'static,
1022 Self::Future: Send + 'static,
1023 {
1024 BoxService::new(self)
1025 }
1026
1027 /// Convert the service into a [`Service`] + [`Clone`] + [`Send`] trait object.
1028 ///
1029 /// This is similar to the [`boxed`] method, but it requires that `Self` implement
1030 /// [`Clone`], and the returned boxed service implements [`Clone`].
1031 /// See [`BoxCloneService`] for more details.
1032 ///
1033 /// # Example
1034 ///
1035 /// ```
1036 /// use tower::{Service, ServiceExt, BoxError, service_fn, util::BoxCloneService};
1037 /// #
1038 /// # struct Request;
1039 /// # struct Response;
1040 /// # impl Response {
1041 /// # fn new() -> Self { Self }
1042 /// # }
1043 ///
1044 /// let service = service_fn(|req: Request| async {
1045 /// Ok::<_, BoxError>(Response::new())
1046 /// });
1047 ///
1048 /// let service: BoxCloneService<Request, Response, BoxError> = service
1049 /// .map_request(|req| {
1050 /// println!("received request");
1051 /// req
1052 /// })
1053 /// .map_response(|res| {
1054 /// println!("response produced");
1055 /// res
1056 /// })
1057 /// .boxed_clone();
1058 ///
1059 /// // The boxed service can still be cloned.
1060 /// service.clone();
1061 /// # let service = assert_service(service);
1062 /// # fn assert_service<S, R>(svc: S) -> S
1063 /// # where S: Service<R> { svc }
1064 /// ```
1065 ///
1066 /// [`Service`]: crate::Service
1067 /// [`boxed`]: Self::boxed
1068 fn boxed_clone(self) -> BoxCloneService<Request, Self::Response, Self::Error>
1069 where
1070 Self: Clone + Sized + Send + 'static,
1071 Self::Future: Send + 'static,
1072 {
1073 BoxCloneService::new(self)
1074 }
1075}
1076
1077impl<T: ?Sized, Request> ServiceExt<Request> for T where T: tower_service::Service<Request> {}
1078
1079/// Convert an `Option<Layer>` into a [`Layer`].
1080///
1081/// The returned [`OptionLayer`] unifies the error types of the layered and
1082/// unlayered branches to [`BoxError`], so the optional layer is allowed to
1083/// change the error type.
1084///
1085/// ```
1086/// # use std::time::Duration;
1087/// # use tower::Service;
1088/// # use tower::builder::ServiceBuilder;
1089/// use tower::util::option_layer;
1090/// # use tower::timeout::TimeoutLayer;
1091/// # async fn wrap<S>(svc: S) where S: Service<(), Error = &'static str> + 'static + Send, S::Future: Send {
1092/// # let timeout = Some(Duration::new(10, 0));
1093/// // Layer to apply a timeout if configured
1094/// let maybe_timeout = option_layer(timeout.map(TimeoutLayer::new));
1095///
1096/// ServiceBuilder::new()
1097/// .layer(maybe_timeout)
1098/// .service(svc);
1099/// # }
1100/// ```
1101///
1102/// [`Layer`]: crate::layer::Layer
1103/// [`OptionLayer`]: crate::util::OptionLayer
1104/// [`BoxError`]: crate::BoxError
1105pub fn option_layer<L>(layer: Option<L>) -> OptionLayer<L> {
1106 OptionLayer::new(layer)
1107}