1//! [`Stream<Item = Request>`][stream] + [`Service<Request>`] => [`Stream<Item = Response>`][stream].
2//!
3//! [`Service<Request>`]: crate::Service
4//! [stream]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html
5
6use super::common;
7use futures_core::Stream;
8use futures_util::stream::FuturesOrdered;
9use pin_project_lite::pin_project;
10use std::{
11 future::Future,
12 pin::Pin,
13 task::{Context, Poll},
14};
15use tower_service::Service;
16
17pin_project! {
18 /// This is a [`Stream`] of responses resulting from calling the wrapped [`Service`] for each
19 /// request received on the wrapped [`Stream`].
20 ///
21 /// ```rust
22 /// # use std::task::{Poll, Context};
23 /// # use std::cell::Cell;
24 /// # use std::error::Error;
25 /// # use std::rc::Rc;
26 /// #
27 /// use futures::future::{ready, Ready};
28 /// use futures::StreamExt;
29 /// use futures::channel::mpsc;
30 /// use tower_service::Service;
31 /// use tower::util::ServiceExt;
32 ///
33 /// // First, we need to have a Service to process our requests.
34 /// #[derive(Debug, Eq, PartialEq)]
35 /// struct FirstLetter;
36 /// impl Service<&'static str> for FirstLetter {
37 /// type Response = &'static str;
38 /// type Error = Box<dyn Error + Send + Sync>;
39 /// type Future = Ready<Result<Self::Response, Self::Error>>;
40 ///
41 /// fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
42 /// Poll::Ready(Ok(()))
43 /// }
44 ///
45 /// fn call(&mut self, req: &'static str) -> Self::Future {
46 /// ready(Ok(&req[..1]))
47 /// }
48 /// }
49 ///
50 /// #[tokio::main]
51 /// async fn main() {
52 /// // Next, we need a Stream of requests.
53 // TODO(eliza): when `tokio-util` has a nice way to convert MPSCs to streams,
54 // tokio::sync::mpsc again?
55 /// let (mut reqs, rx) = mpsc::unbounded();
56 /// // Note that we have to help Rust out here by telling it what error type to use.
57 /// // Specifically, it has to be From<Service::Error> + From<Stream::Error>.
58 /// let mut rsps = FirstLetter.call_all(rx);
59 ///
60 /// // Now, let's send a few requests and then check that we get the corresponding responses.
61 /// reqs.unbounded_send("one").unwrap();
62 /// reqs.unbounded_send("two").unwrap();
63 /// reqs.unbounded_send("three").unwrap();
64 /// drop(reqs);
65 ///
66 /// // We then loop over the response `Stream` that we get back from call_all.
67 /// let mut i = 0usize;
68 /// while let Some(rsp) = rsps.next().await {
69 /// // Each response is a Result (we could also have used TryStream::try_next)
70 /// match (i + 1, rsp.unwrap()) {
71 /// (1, "o") |
72 /// (2, "t") |
73 /// (3, "t") => {}
74 /// (n, i) => {
75 /// unreachable!("{}. response was '{}'", n, i);
76 /// }
77 /// }
78 /// i += 1;
79 /// }
80 ///
81 /// // And at the end, we can get the Service back when there are no more requests.
82 /// assert_eq!(rsps.into_inner(), FirstLetter);
83 /// }
84 /// ```
85 ///
86 /// [`Stream`]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html
87 #[derive(Debug)]
88 pub struct CallAll<Svc, S>
89 where
90 Svc: Service<S::Item>,
91 S: Stream,
92 {
93 #[pin]
94 inner: common::CallAll<Svc, S, FuturesOrdered<Svc::Future>>,
95 }
96}
97
98impl<Svc, S> CallAll<Svc, S>
99where
100 Svc: Service<S::Item>,
101 S: Stream,
102{
103 /// Create new [`CallAll`] combinator.
104 ///
105 /// Each request yielded by `stream` is passed to `svc`, and the resulting responses are
106 /// yielded in the same order by the implementation of [`Stream`] for [`CallAll`].
107 ///
108 /// [`Stream`]: https://docs.rs/futures/latest/futures/stream/trait.Stream.html
109 pub fn new(service: Svc, stream: S) -> CallAll<Svc, S> {
110 CallAll {
111 inner: common::CallAll::new(service, stream, FuturesOrdered::new()),
112 }
113 }
114
115 /// Extract the wrapped [`Service`].
116 ///
117 /// # Panics
118 ///
119 /// Panics if [`take_service`] was already called.
120 ///
121 /// [`take_service`]: crate::util::CallAll::take_service
122 pub fn into_inner(self) -> Svc {
123 self.inner.into_inner()
124 }
125
126 /// Extract the wrapped [`Service`].
127 ///
128 /// This [`CallAll`] can no longer be used after this function has been called.
129 ///
130 /// # Panics
131 ///
132 /// Panics if [`take_service`] was already called.
133 ///
134 /// [`take_service`]: crate::util::CallAll::take_service
135 pub fn take_service(self: Pin<&mut Self>) -> Svc {
136 self.project().inner.take_service()
137 }
138
139 /// Return responses as they are ready, regardless of the initial order.
140 ///
141 /// This function must be called before the stream is polled.
142 ///
143 /// # Panics
144 ///
145 /// Panics if [`poll`] was called.
146 ///
147 /// [`poll`]: std::future::Future::poll
148 pub fn unordered(self) -> super::CallAllUnordered<Svc, S> {
149 self.inner.unordered()
150 }
151}
152
153impl<Svc, S> Stream for CallAll<Svc, S>
154where
155 Svc: Service<S::Item>,
156 S: Stream,
157{
158 type Item = Result<Svc::Response, Svc::Error>;
159
160 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
161 self.project().inner.poll_next(cx)
162 }
163}
164
165impl<F: Future> common::Drive<F> for FuturesOrdered<F> {
166 fn is_empty(&self) -> bool {
167 FuturesOrdered::is_empty(self)
168 }
169
170 fn push(&mut self, future: F) {
171 FuturesOrdered::push_back(self, future)
172 }
173
174 fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Option<F::Output>> {
175 Stream::poll_next(self:Pin::new(self), cx)
176 }
177}
178