pub struct BindMany<S> { /* private fields */ }
Expand description
This stream replaces tokio_core::net::Incoming
and listens many sockets
It receives a stream of lists of addresses as an input. When a new value received on a stream it adapts:
- Removes sockets not in set we’re already received (already established connections aren’t interfered in any way)
- Adds sockets to set which wasn’t listened before
Instead of failing on bind error it logs the error and retries in a
second (you can change the delay using BindMany::retry_interval
)
It’s good idea to pass a stream with a Void error, because on receiving
error BindMany
will log a message (that doesn’t contain an error) and
will shutdown. It’s better to log specific error and send end-of-stream
instead, but that is user’s responsibility.
Note: we track identity of the sockets by SocketAddr
used to bind it,
this means 0.0.0.0
and 127.0.0.1
for example can be bound/unbound
independently despite the fact that 0.0.0.0
can accept connections for
127.0.0.1
.
Example
Simple example:
lp.run(
BindMany::new(address_stream)
.sleep_on_error(TIME_TO_WAIT_ON_ERROR, &h2)
.map(move |(mut socket, _addr)| {
// Your future is here:
Proto::new(socket)
// Errors should not pass silently
// common idea is to log them
.map_err(|e| error!("Protocol error: {}", e))
})
.listen(MAX_SIMULTANEOUS_CONNECTIONS)
).unwrap(); // stream doesn't end in this case
Example using name resolution via abstract-ns + ns-env-config:
extern crate ns_env_config;
let mut lp = Core::new().unwrap();
let ns = ns_env_config::init(&lp.handle()).unwrap();
lp.run(
BindMany::new(ns.resolve_auto("localhost", 8080)
.map(|addr| addr.addresses_at(0)))
.sleep_on_error(TIME_TO_WAIT_ON_ERROR, &h2)
.map(move |(mut socket, _addr)| {
// Your future is here:
Proto::new(socket)
// Errors should not pass silently
// common idea is to log them
.map_err(|e| eprintln!("Protocol error: {}", e))
})
.listen(MAX_SIMULTANEOUS_CONNECTIONS)
).unwrap(); // stream doesn't end in this case
Implementations
sourceimpl<S> BindMany<S>
impl<S> BindMany<S>
sourcepub fn retry_interval(&mut self, interval: Duration) -> &mut Self
pub fn retry_interval(&mut self, interval: Duration) -> &mut Self
Sets the retry interval
Each time binding socket fails (including he first one on start) istead of immediately failing we log the error and sleep this interval to retry (by default 1 second).
This behavior is important because if your configuration changes number of listening sockets, and one of them is either invalid or just DNS is temporarily down, you still need to serve other addresses.
This also helps if you have failover IP which can only be listened at when IP attached to the host, but server must be ready to listen it anyway (this one might be better achieved by non-local bind though).
Trait Implementations
sourceimpl<S> Stream for BindMany<S>where
S: Stream,
S::Item: IntoIterator<Item = SocketAddr>,
impl<S> Stream for BindMany<S>where
S: Stream,
S::Item: IntoIterator<Item = SocketAddr>,
sourcefn poll(&mut self) -> Result<Async<Option<Self::Item>>, Error>
fn poll(&mut self) -> Result<Async<Option<Self::Item>>, Error>
None
if
the stream is finished. Read moresourcefn wait(self) -> Wait<Self>where
Self: Sized,
fn wait(self) -> Wait<Self>where
Self: Sized,
sourcefn into_future(self) -> StreamFuture<Self>where
Self: Sized,
fn into_future(self) -> StreamFuture<Self>where
Self: Sized,
Future
. Read moresourcefn filter<F>(self, f: F) -> Filter<Self, F>where
F: FnMut(&Self::Item) -> bool,
Self: Sized,
fn filter<F>(self, f: F) -> Filter<Self, F>where
F: FnMut(&Self::Item) -> bool,
Self: Sized,
sourcefn filter_map<F, B>(self, f: F) -> FilterMap<Self, F>where
F: FnMut(Self::Item) -> Option<B>,
Self: Sized,
fn filter_map<F, B>(self, f: F) -> FilterMap<Self, F>where
F: FnMut(Self::Item) -> Option<B>,
Self: Sized,
sourcefn then<F, U>(self, f: F) -> Then<Self, F, U>where
F: FnMut(Result<Self::Item, Self::Error>) -> U,
U: IntoFuture,
Self: Sized,
fn then<F, U>(self, f: F) -> Then<Self, F, U>where
F: FnMut(Result<Self::Item, Self::Error>) -> U,
U: IntoFuture,
Self: Sized,
f
. Read moresourcefn and_then<F, U>(self, f: F) -> AndThen<Self, F, U>where
F: FnMut(Self::Item) -> U,
U: IntoFuture<Error = Self::Error>,
Self: Sized,
fn and_then<F, U>(self, f: F) -> AndThen<Self, F, U>where
F: FnMut(Self::Item) -> U,
U: IntoFuture<Error = Self::Error>,
Self: Sized,
f
. Read moresourcefn or_else<F, U>(self, f: F) -> OrElse<Self, F, U>where
F: FnMut(Self::Error) -> U,
U: IntoFuture<Item = Self::Item>,
Self: Sized,
fn or_else<F, U>(self, f: F) -> OrElse<Self, F, U>where
F: FnMut(Self::Error) -> U,
U: IntoFuture<Item = Self::Item>,
Self: Sized,
f
. Read moresourcefn collect(self) -> Collect<Self>where
Self: Sized,
fn collect(self) -> Collect<Self>where
Self: Sized,
sourcefn fold<F, T, Fut>(self, init: T, f: F) -> Fold<Self, F, Fut, T>where
F: FnMut(T, Self::Item) -> Fut,
Fut: IntoFuture<Item = T>,
Self::Error: From<<Fut as IntoFuture>::Error>,
Self: Sized,
fn fold<F, T, Fut>(self, init: T, f: F) -> Fold<Self, F, Fut, T>where
F: FnMut(T, Self::Item) -> Fut,
Fut: IntoFuture<Item = T>,
Self::Error: From<<Fut as IntoFuture>::Error>,
Self: Sized,
sourcefn skip_while<P, R>(self, pred: P) -> SkipWhile<Self, P, R>where
P: FnMut(&Self::Item) -> R,
R: IntoFuture<Item = bool, Error = Self::Error>,
Self: Sized,
fn skip_while<P, R>(self, pred: P) -> SkipWhile<Self, P, R>where
P: FnMut(&Self::Item) -> R,
R: IntoFuture<Item = bool, Error = Self::Error>,
Self: Sized,
true
. Read moresourcefn take_while<P, R>(self, pred: P) -> TakeWhile<Self, P, R>where
P: FnMut(&Self::Item) -> R,
R: IntoFuture<Item = bool, Error = Self::Error>,
Self: Sized,
fn take_while<P, R>(self, pred: P) -> TakeWhile<Self, P, R>where
P: FnMut(&Self::Item) -> R,
R: IntoFuture<Item = bool, Error = Self::Error>,
Self: Sized,
true
. Read moresourcefn for_each<F, U>(self, f: F) -> ForEach<Self, F, U>where
F: FnMut(Self::Item) -> U,
U: IntoFuture<Item = (), Error = Self::Error>,
Self: Sized,
fn for_each<F, U>(self, f: F) -> ForEach<Self, F, U>where
F: FnMut(Self::Item) -> U,
U: IntoFuture<Item = (), Error = Self::Error>,
Self: Sized,
sourcefn from_err<E>(self) -> FromErr<Self, E>where
E: From<Self::Error>,
Self: Sized,
fn from_err<E>(self) -> FromErr<Self, E>where
E: From<Self::Error>,
Self: Sized,
From
for
this stream’s Error
, returning a new stream. Read moresourcefn take(self, amt: u64) -> Take<Self>where
Self: Sized,
fn take(self, amt: u64) -> Take<Self>where
Self: Sized,
amt
items of the underlying stream. Read moresourcefn skip(self, amt: u64) -> Skip<Self>where
Self: Sized,
fn skip(self, amt: u64) -> Skip<Self>where
Self: Sized,
amt
items of the underlying stream. Read moresourcefn fuse(self) -> Fuse<Self>where
Self: Sized,
fn fuse(self) -> Fuse<Self>where
Self: Sized,
poll
will never again be called once it has
finished. Read moresourcefn by_ref(&mut self) -> &mut Selfwhere
Self: Sized,
fn by_ref(&mut self) -> &mut Selfwhere
Self: Sized,
sourcefn merge<S>(self, other: S) -> Merge<Self, S>where
S: Stream<Error = Self::Error>,
Self: Sized,
fn merge<S>(self, other: S) -> Merge<Self, S>where
S: Stream<Error = Self::Error>,
Self: Sized,
select
nowsourcefn zip<S>(self, other: S) -> Zip<Self, S>where
S: Stream<Error = Self::Error>,
Self: Sized,
fn zip<S>(self, other: S) -> Zip<Self, S>where
S: Stream<Error = Self::Error>,
Self: Sized,
sourcefn chain<S>(self, other: S) -> Chain<Self, S>where
S: Stream<Item = Self::Item, Error = Self::Error>,
Self: Sized,
fn chain<S>(self, other: S) -> Chain<Self, S>where
S: Stream<Item = Self::Item, Error = Self::Error>,
Self: Sized,
sourcefn peekable(self) -> Peekable<Self>where
Self: Sized,
fn peekable(self) -> Peekable<Self>where
Self: Sized,
peek
method. Read more