Struct CallAll

pub struct CallAll<Svc, S>
where
    Svc: Service<S::Item>,
    S: Stream, { /* private fields */ }

This is a Stream of responses resulting from calling the wrapped Service for each request received on the wrapped Stream.

# use std::task::{Poll, Context};
# use std::cell::Cell;
# use std::error::Error;
# use std::rc::Rc;
#
use std::future::{ready, Ready};
use futures::StreamExt;
use futures::channel::mpsc;
use tower_service::Service;
use tower::util::ServiceExt;

// First, we need to have a Service to process our requests.
#[derive(Debug, Eq, PartialEq)]
struct FirstLetter;
impl Service<&'static str> for FirstLetter {
     type Response = &'static str;
     type Error = Box<dyn Error + Send + Sync>;
     type Future = Ready<Result<Self::Response, Self::Error>>;

     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
     }

     fn call(&mut self, req: &'static str) -> Self::Future {
         ready(Ok(&req[..1]))
     }
}

#[tokio::main]
async fn main() {
    // Next, we need a Stream of requests.
    let (mut reqs, rx) = mpsc::unbounded();
    // Note that we have to help Rust out here by telling it what error type to use.
    // Specifically, it has to be From<Service::Error> + From<Stream::Error>.
    let mut rsps = FirstLetter.call_all(rx);

    // Now, let's send a few requests and then check that we get the corresponding responses.
    reqs.unbounded_send("one").unwrap();
    reqs.unbounded_send("two").unwrap();
    reqs.unbounded_send("three").unwrap();
    drop(reqs);

    // We then loop over the response `Stream` that we get back from call_all.
    let mut i = 0usize;
    while let Some(rsp) = rsps.next().await {
        // Each response is a Result (we could also have used TryStream::try_next)
        match (i + 1, rsp.unwrap()) {
            (1, "o") |
            (2, "t") |
            (3, "t") => {}
            (n, i) => {
                unreachable!("{}. response was '{}'", n, i);
            }
        }
        i += 1;
    }

    // And at the end, we can get the Service back when there are no more requests.
    assert_eq!(rsps.into_inner(), FirstLetter);
}

Implementations

impl<Svc, S> CallAll<Svc, S> where Svc: Service<S::Item>, S: Stream,

fn new(service: Svc, stream: S) -> CallAll<Svc, S>

Create new CallAll combinator.

Each request yielded by stream is passed to svc, and the resulting responses are yielded in the same order by the implementation of Stream for CallAll.

fn into_inner(self) -> Svc

Extract the wrapped Service.

Panics

Panics if take_service was already called.

fn take_service(self: Pin<&mut Self>) -> Svc

Extract the wrapped Service.

This CallAll can no longer be used after this function has been called.

Panics

Panics if take_service was already called.

fn unordered(self) -> CallAllUnordered<Svc, S>

Return responses as they are ready, regardless of the initial order.

This function must be called before the stream is polled.

Panics

Panics if poll was called.

Trait Implementations

impl<'__pin, Svc, S> Unpin for CallAll<Svc, S> where PinnedFieldsOf<__Origin<'__pin, Svc, S>>: Unpin, Svc: Service<S::Item>, S: Stream,

impl<Svc, S> Debug for CallAll<Svc, S> where Svc: Service<S::Item> + Debug, S: Stream + Debug, Svc::Future: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

impl<Svc, S> Stream for CallAll<Svc, S> where Svc: Service<S::Item>, S: Stream,

type Item = Result<<Svc as Service<<S as Stream>::Item>>::Response, <Svc as Service<<S as Stream>::Item>>::Error>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>

Auto Trait Implementations

impl<Svc, S> !Freeze for CallAll<Svc, S>

impl<Svc, S> !RefUnwindSafe for CallAll<Svc, S>

impl<Svc, S> !UnwindSafe for CallAll<Svc, S>

impl<Svc, S> Send for CallAll<Svc, S> where CallAll<Svc, S, FuturesOrdered<<Svc as Service<<S as Stream>::Item>>::Future>>: Send,

impl<Svc, S> Sync for CallAll<Svc, S> where CallAll<Svc, S, FuturesOrdered<<Svc as Service<<S as Stream>::Item>>::Future>>: Sync,

impl<Svc, S> UnsafeUnpin for CallAll<Svc, S> where CallAll<Svc, S, FuturesOrdered<<Svc as Service<<S as Stream>::Item>>::Future>>: UnsafeUnpin,

Blanket Implementations

impl<K, S, E, D> Discover for CallAll<Svc, S> where D: TryStream<Ok = Change<K, S>, Error = E> + ?Sized, K: Eq,

type Key = K;
type Service = S;
type Error = E;
fn poll_discover(self: Pin<&mut D>, cx: &mut Context<'_>) -> Poll<Option<Result<<D as TryStream>::Ok, <D as TryStream>::Error>>>

impl<S> TryStreamExt for CallAll<Svc, S> where S: TryStream + ?Sized,

impl<S, T, E> TryStream for CallAll<Svc, S> where S: Stream<Item = Result<T, E>> + ?Sized,

type Ok = T;
type Error = E;
fn try_poll_next(self: Pin<&mut S>, cx: &mut Context<'_>) -> Poll<Option<Result<<S as TryStream>::Ok, <S as TryStream>::Error>>>

impl<T> Any for CallAll<Svc, S> where T: 'static + ?Sized,

fn type_id(&self) -> TypeId

impl<T> Borrow<T> for CallAll<Svc, S> where T: ?Sized,

fn borrow(&self) -> &T

impl<T> BorrowMut<T> for CallAll<Svc, S> where T: ?Sized,

fn borrow_mut(&mut self) -> &mut T

impl<T> From<T> for CallAll<Svc, S>

fn from(t: T) -> T

Returns the argument unchanged.

impl<T> Instrument for CallAll<Svc, S>

impl<T> StreamExt for CallAll<Svc, S> where T: Stream + ?Sized,

impl<T> WithSubscriber for CallAll<Svc, S>

impl<T, U> Into<U> for CallAll<Svc, S> where U: From<T>,

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of [From]<T> for U chooses to do.

impl<T, U> TryFrom<U> for CallAll<Svc, S> where U: Into<T>,

type Error = Infallible;
fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

impl<T, U> TryInto<U> for CallAll<Svc, S> where U: TryFrom<T>,

type Error = <U as TryFrom<T>>::Error;
fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>