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
#![feature(const_btree_new)]
use log::*;
use redis::{Commands, IntoConnectionInfo};
use spin::{Mutex, RwLock};
use std::collections::BTreeMap;
use std::thread;
use std::time::Duration;
use unmp::link::{ErrorKind, Link};
use unmp::net;
static LINKS: Mutex<BTreeMap<String, Link>> = Mutex::new(BTreeMap::new());
static DRIVERS: RwLock<BTreeMap<Link, Driver>> = RwLock::new(BTreeMap::new());
struct Driver {
tx: Mutex<redis::Connection>,
topic: String,
}
fn sendto(link: Link, buf: &[u8]) -> Result<(), ErrorKind> {
trace!("redis send: {:02X?}.", buf);
if let Some(driver) = DRIVERS.read().get(&link) {
match driver.tx.lock().publish::<_, _, u8>(&driver.topic, buf) {
Ok(_) => {
return Ok(());
}
Err(_) => {
warn!("redis send error.");
return Err(ErrorKind::TimedOut);
}
};
} else {
return Err(ErrorKind::NotConnected);
}
}
pub fn start(url: &str, topic: &str) -> Link {
let url = url.into_connection_info();
if let Err(_) = url {
return Link::INVALID;
}
let url = url.unwrap();
let identifier = format!("{}/{}", url.addr, topic);
if let Some(link) = LINKS.lock().get(&identifier) {
return *link;
}
let client = redis::Client::open(url).unwrap();
let tx = client.get_connection().unwrap();
tx.set_write_timeout(Some(Duration::new(0, 500_000)))
.expect("redis set write timeout");
let driver = Driver {
tx: Mutex::new(tx),
topic: String::from(topic),
};
let link = Link::new(sendto);
info!("redis new {}({}).", link, identifier);
LINKS.lock().insert(identifier.clone(), link);
DRIVERS.write().insert(link, driver);
let topic_tmp = String::from(topic);
thread::spawn(move || {
let mut rx = client.get_connection().unwrap();
let mut pubsub = rx.as_pubsub();
pubsub.subscribe(topic_tmp).unwrap();
loop {
match pubsub.get_message() {
Ok(msg) => {
let buf = msg.get_payload_bytes();
trace!("redis recv: {:02X?}.", buf);
thread::spawn(move || {
let buf = msg.get_payload_bytes();
net::when_recv(link, &buf);
});
}
Err(e) => {
warn!("redis recv error: {:?}", e.kind());
LINKS.lock().remove(&identifier);
DRIVERS.write().remove(&link);
info!("redis rm {}.", link);
link.destroy();
return;
}
}
}
});
return link;
}