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

//! 服务端

use log::*;
use spin::{Mutex, RwLock};
use std::collections::BTreeMap;
use std::thread;
use unmp::link::{ErrorKind, Link};
use unmp::net;
use websocket::OwnedMessage;

static DRIVERS: RwLock<BTreeMap<Link, Driver>> = RwLock::new(BTreeMap::new());

struct Driver {
    socket: Mutex<websocket::sender::Writer<std::net::TcpStream>>,
}
fn sendto(link: Link, buf: &[u8]) -> Result<(), ErrorKind> {
    trace!("websocket_server send: {:02X?}.", buf);
    if let Some(driver) = DRIVERS.write().get(&link) {
        match driver
            .socket
            .lock()
            .send_message(&OwnedMessage::Binary(buf.into()))
        {
            Ok(_) => {
                return Ok(());
            }
            Err(_) => {
                warn!("websocket_server send error.");
                return Err(ErrorKind::TimedOut);
            }
        };
    } else {
        return Err(ErrorKind::NotConnected);
    }
}

/// 创建一个链路实例，并监听指定端口
pub fn start(ip: &str, port: u16) {
    let server = websocket::sync::Server::bind((ip, port)).unwrap();

    thread::spawn(|| {
        for request in server.filter_map(Result::ok) {
            thread::spawn(|| {
                if !request.protocols().contains(&"unmp".to_string()) {
                    request.reject().unwrap();
                    return;
                }
                let client = request.use_protocol("unmp").accept().unwrap();
                let identifier = format!("{}", client.peer_addr().unwrap());

                let (mut receiver, sender) = client.split().unwrap();
                let driver = Driver {
                    socket: Mutex::new(sender),
                };
                let link = Link::new(sendto);
                info!("websocket_server new {}({}).", link, identifier);
                DRIVERS.write().insert(link, driver);

                for message in receiver.incoming_messages() {
                    match message {
                        Ok(m) => match m {
                            OwnedMessage::Binary(buf) => {
                                trace!("websocket_server recv: {:02X?}.", buf);

                                // 接收数据帧
                                thread::spawn(move || {
                                    net::when_recv(link, &buf);
                                });
                            }
                            _ => {}
                        },
                        Err(e) => {
                            warn!("websocket_server recv error: {:?}", e);
                            DRIVERS.write().remove(&link);
                            info!("websocket_server rm {}.", link);
                            link.destroy();
                            return;
                        }
                    };
                }
            });
        }
    });
}