1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

//! Futures and streams returned from router
//!
use std::sync::Arc;

use abstract_ns::{IpList, Address, Error};
use futures::sync::oneshot;
use futures::{Future, Async, Stream};
use void::Void;

use async_slot as slot;
use config::Config;

/// A future returned from `Router::resolve_host`
#[derive(Debug)]
pub struct ResolveHostFuture(
    pub(crate) oneshot::Receiver<Result<IpList, Error>>);

/// A future returned from `Router::resolve`
#[derive(Debug)]
pub struct ResolveFuture(pub(crate) oneshot::Receiver<Result<Address, Error>>);

/// A stream returned from `Router::host_subscribe`
#[derive(Debug)]
pub struct HostStream(pub(crate) slot::Receiver<IpList>);

/// A stream returned from `Router::subscribe`
#[derive(Debug)]
pub struct AddrStream(pub(crate) slot::Receiver<Address>);

/// A sink that updates router created using `Router::updating_config`
#[derive(Debug)]
pub struct UpdateSink(pub(crate) slot::Sender<Arc<Config>>);


impl UpdateSink {
    /// Update a config
    ///
    /// Returns `true` if send worked (meaning router is still alive).
    pub fn update(&self, config: &Arc<Config>) -> bool {
        self.0.swap(config.clone()).is_ok()
    }
}

impl Future for ResolveHostFuture {
    type Item = IpList;
    type Error = Error;
    #[inline(always)]
    fn poll(&mut self) -> Result<Async<IpList>, Error> {
        match self.0.poll().map_err(|e| Error::TemporaryError(e.into()))? {
            Async::NotReady => Ok(Async::NotReady),
            Async::Ready(Ok(r))  => Ok(Async::Ready(r)),
            Async::Ready(Err(e))  => Err(e),
        }
    }
}

impl Future for ResolveFuture {
    type Item = Address;
    type Error = Error;
    #[inline(always)]
    fn poll(&mut self) -> Result<Async<Address>, Error> {
        match self.0.poll().map_err(|e| Error::TemporaryError(e.into()))? {
            Async::NotReady => Ok(Async::NotReady),
            Async::Ready(Ok(r))  => Ok(Async::Ready(r)),
            Async::Ready(Err(e))  => Err(e),
        }
    }
}

impl Stream for HostStream {
    type Item = IpList;
    type Error = Void;
    #[inline(always)]
    fn poll(&mut self) -> Result<Async<Option<IpList>>, Void> {
        match self.0.poll() {
            Ok(r) => Ok(r),
            Err(_) => Ok(Async::Ready(None)),
        }
    }
}

impl Stream for AddrStream {
    type Item = Address;
    type Error = Void;
    #[inline(always)]
    fn poll(&mut self) -> Result<Async<Option<Address>>, Void> {
        match self.0.poll() {
            Ok(r) => Ok(r),
            Err(_) => Ok(Async::Ready(None)),
        }
    }
}