lrcall/server/
incoming.rs

1use super::limits::channels_per_key::MaxChannelsPerKey;
2use super::limits::requests_per_channel::MaxRequestsPerChannel;
3use super::{Channel, RequestName, Serve};
4use futures::prelude::*;
5use std::fmt;
6use std::hash::Hash;
7
8/// An extension trait for [streams](futures::prelude::Stream) of [`Channels`](Channel).
9pub trait Incoming<C>
10where
11    Self: Sized + Stream<Item = C>,
12    C: Channel,
13{
14    /// Enforces channel per-key limits.
15    fn max_channels_per_key<K, KF>(self, n: u32, keymaker: KF) -> MaxChannelsPerKey<Self, K, KF>
16    where
17        K: fmt::Display + Eq + Hash + Clone + Unpin,
18        KF: Fn(&C) -> K,
19    {
20        MaxChannelsPerKey::new(self, n, keymaker)
21    }
22
23    /// Caps the number of concurrent requests per channel.
24    fn max_concurrent_requests_per_channel(self, n: usize) -> MaxRequestsPerChannel<Self> {
25        MaxRequestsPerChannel::new(self, n)
26    }
27
28    /// Returns a stream of channels in execution. Each channel in execution is a stream of
29    /// futures, where each future is an in-flight request being rsponded to.
30    fn execute<S>(self, serve: S) -> impl Stream<Item = impl Stream<Item = impl Future<Output = ()>>>
31    where
32        C::Req: RequestName,
33        S: Serve<Req = C::Req, Resp = C::Resp> + Clone,
34    {
35        self.map(move |channel| channel.execute(serve.clone()))
36    }
37}
38
39#[cfg(feature = "tokio1")]
40/// Spawns all channels-in-execution, delegating to the tokio runtime to manage their completion.
41/// Each channel is spawned, and each request from each channel is spawned.
42/// Note that this function is generic over any stream-of-streams-of-futures, but it is intended
43/// for spawning streams of channels.
44///
45/// # Example
46/// ```rust
47/// use lrcall::{
48///     context,
49///     client::{self, NewClient},
50///     server::{self, BaseChannel, Channel, incoming::{Incoming, spawn_incoming}, serve},
51///     transport,
52/// };
53/// use futures::prelude::*;
54///
55/// #[tokio::main]
56/// async fn main() {
57///     let (tx, rx) = transport::channel::unbounded();
58///     let NewClient { client, dispatch } = client::new(client::Config::default(), tx);
59///     tokio::spawn(dispatch);
60///
61///     let incoming = stream::once(async move {
62///         BaseChannel::new(server::Config::default(), rx)
63///     }).execute(serve(|_, i| async move { Ok(i + 1) }));
64///     tokio::spawn(spawn_incoming(incoming));
65///     assert_eq!(client.call(context::rpc_current(), 1).await.unwrap(), 2);
66/// }
67/// ```
68pub async fn spawn_incoming(incoming: impl Stream<Item = impl Stream<Item = impl Future<Output = ()> + Send + 'static> + Send + 'static>) {
69    use futures::pin_mut;
70    pin_mut!(incoming);
71    while let Some(channel) = incoming.next().await {
72        tokio::spawn(async move {
73            pin_mut!(channel);
74            while let Some(request) = channel.next().await {
75                tokio::spawn(request);
76            }
77        });
78    }
79}
80
81impl<S, C> Incoming<C> for S
82where
83    S: Sized + Stream<Item = C>,
84    C: Channel,
85{
86}
lrcall/server/incoming.rs

lrcall/server/
incoming.rs
