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
use std::collections::HashMap;
use std::net::SocketAddr;
use futures::{Future, Stream};
use futures::sync::mpsc;
use {Error, Request, Service};
use dispatch::{PrimitiveDispatch, StreamingDispatch};
use protocol::Flatten;
use resolve::ResolveInfo;
pub type HashRing = Vec<(u64, String)>;
enum Method {
Resolve,
Routing,
}
impl Into<u64> for Method {
#[inline]
fn into(self) -> u64 {
match self {
Method::Resolve => 0,
Method::Routing => 5,
}
}
}
#[derive(Clone, Debug)]
pub struct Locator {
service: Service,
}
impl Locator {
pub fn new(service: Service) -> Self {
Self { service }
}
pub fn resolve(&self, name: &str) -> impl Future<Item = ResolveInfo<SocketAddr>, Error = Error> {
let (dispatch, future) = PrimitiveDispatch::pair();
self.service.call(Request::new(Method::Resolve.into(), &[name]).unwrap(), dispatch);
future.map(|ResolveInfo{addrs, version, methods}| {
let addrs = addrs.into_iter()
.map(|(ip, port)| SocketAddr::new(ip, port))
.collect();
ResolveInfo {
addrs: addrs,
version: version,
methods: methods,
}
})
}
pub fn routing(&self, uuid: &str) ->
impl Stream<Item = HashMap<String, HashRing>, Error = Error>
{
let (tx, rx) = mpsc::unbounded();
let dispatch = StreamingDispatch::new(tx);
self.service.call(Request::new(Method::Routing.into(), &[uuid]).unwrap(), dispatch);
rx.map_err(|()| Error::Canceled).then(Flatten::flatten)
}
}