message-io 0.12.2

Event-driven message library to build network applications easy and fast
Documentation 
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

use super::common::{Message};

use message_io::network::{NetEvent, Transport, Endpoint};
use message_io::node::{self, NodeHandler, NodeListener};

use std::net::{SocketAddr};
use std::collections::{HashMap};

pub struct Participant {
    handler: NodeHandler<()>,
    listener: Option<NodeListener<()>>,
    name: String,
    discovery_endpoint: Endpoint,
    public_addr: SocketAddr,
    known_participants: HashMap<String, Endpoint>, // Used only for free resources later
}

impl Participant {
    pub fn new(name: &str) -> Option<Participant> {
        let (handler, listener) = node::split();

        // A listener for any other participant that want to establish connection.
        // 'addr' contains the port that the OS gives for us when we put a 0.
        let listen_addr = "127.0.0.1:0";
        let listen_addr = match handler.network().listen(Transport::Udp, listen_addr) {
            Ok((_, addr)) => addr,
            Err(_) => {
                println!("Can not listen on {}", listen_addr);
                return None
            }
        };

        let discovery_addr = "127.0.0.1:5000"; // Connection to the discovery server.
        match handler.network().connect(Transport::FramedTcp, discovery_addr) {
            Ok((endpoint, _)) => Some(Participant {
                handler,
                listener: Some(listener),
                name: name.to_string(),
                discovery_endpoint: endpoint,
                public_addr: listen_addr,
                known_participants: HashMap::new(),
            }),
            Err(_) => {
                println!("Can not connect to the discovery server at {}", discovery_addr);
                return None
            }
        }
    }

    pub fn run(mut self) {
        // Register this participant into the discovery server
        let message = Message::RegisterParticipant(self.name.clone(), self.public_addr);
        let output_data = bincode::serialize(&message).unwrap();
        self.handler.network().send(self.discovery_endpoint, &output_data);

        let listener = self.listener.take().unwrap();
        listener.for_each(move |event| match event.network() {
            NetEvent::Message(_, input_data) => {
                let message: Message = bincode::deserialize(&input_data).unwrap();
                match message {
                    Message::ParticipantList(participants) => {
                        println!("Participant list received ({} participants)", participants.len());
                        for (name, addr) in participants {
                            self.discovered_participant(
                                &name,
                                addr,
                                "I see you in the participant list",
                            );
                        }
                    }
                    Message::ParticipantNotificationAdded(name, addr) => {
                        println!("New participant '{}' in the network", name);
                        self.discovered_participant(&name, addr, "welcome to the network!");
                    }
                    Message::ParticipantNotificationRemoved(name) => {
                        println!("Removed participant '{}' from the network", name);

                        // Free network resource.
                        // It is only necessary because the connections among participants
                        // are done by UDP,
                        // UDP is not connection-oriented protocol, and the
                        // Connected/Disconnected events are not generated by UDP.
                        if let Some(endpoint) = self.known_participants.remove(&name) {
                            self.handler.network().remove(endpoint.resource_id());
                        }
                    }
                    Message::Gretings(name, gretings) => {
                        println!("'{}' says: {}", name, gretings);
                    }
                    _ => unreachable!(),
                }
            }
            NetEvent::Connected(_, _) => (),
            NetEvent::Disconnected(endpoint) => {
                if endpoint == self.discovery_endpoint {
                    println!("Discovery server disconnected, closing");
                    self.handler.stop();
                }
            }
        });
    }

    fn discovered_participant(&mut self, name: &str, addr: SocketAddr, message: &str) {
        if let Ok((endpoint, _)) = self.handler.network().connect(Transport::Udp, addr) {
            let gretings = format!("Hi '{}', {}", name, message);
            let message = Message::Gretings(self.name.clone(), gretings);
            let output_data = bincode::serialize(&message).unwrap();
            self.handler.network().send(endpoint, &output_data);
            self.known_participants.insert(name.to_string(), endpoint);
        }
    }
}