wsrouter 0.2.1

Easy WebSocket-only routing for servers and clients
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

use std::sync::Arc;
use futures_util::{SinkExt, StreamExt};
use tokio_tungstenite::connect_async;
use tokio_tungstenite::tungstenite::Message;
use crate::parser::Parsed;
use crate::routes::{Dispatcher, Params, Route};

pub struct Connector {
    url: String,
    routes: Vec<Route>
}







impl Connector {
    //create new connector with an url
    pub fn new(url: impl Into<String>) -> Self {
        let url = url.into();
        Self {
            url,
            routes: Vec::new()
        }
    }

    //add new incomeing routes
    pub fn route<F>(&mut self, name: impl Into<String>, callback: F)
    where
        F: Fn(&Params, &Dispatcher) + Send + Sync + 'static,
    {
        let name = name.into();
        self.routes.push(Route {
            name,
            callback: Box::new(callback),
        });
    }

    //connect to the server by consuming this connector and returning a dispather
    pub async fn connect(self) -> Dispatcher {

        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<String>();
        let routes = self.routes;
        let sender_clone = sender.clone();
        let url = Arc::new(self.url);
        //let (mut write, mut read) = ws_stream.split();

        //move everything in one spawn to manage reconnection
        let _life_cycle = tokio::spawn(async move {
            loop {
                let (ws_stream, _) = match connect_async(url.to_string()).await  {
                    Ok(stream) => stream,
                    Err(e) => {
                        eprintln!("connect error: {}", e);
                        continue;
                    }
                };
                let (mut write, mut read) = ws_stream.split();

                if let Some(found_route) = routes.iter().find(|route| route.name == "CONNECTED") {
                    let params =&Params::new();
                    (found_route.callback)(&params, &Dispatcher { sender: sender_clone.clone() });
                }
                let connection_result = loop {
                    tokio::select! {
                        Some(msg) = receiver.recv() => {
                            if let Err(e) = write.send(Message::text(msg)).await {
                                eprintln!("send error: {}", e);
                                break;
                            }
                        },
                        Some(msg) = read.next() => {
                            let msg = match msg {
                                Ok(msg) => msg,
                                Err(e) => {
                                    eprintln!("couldn't extract message: {}", e);
                                    break
                                }
                            };
                            let parsed = Parsed::parse(msg.to_string());
                            let params = parsed.params;
                            let command = parsed.command;

                            if let Some(found_route) = routes.iter().find(|route| route.name == command.to_string()) {
                                (found_route.callback)(&params, &Dispatcher { sender: sender_clone.clone() });
                                }
                            else {
                                println!("No route found for {}", command);
                                println!("Current routes: {:?}", routes.iter().map(|route| route.name.clone()).collect::<Vec<String>>());
                                }
                            }
                        else => break
                    }
                };
                eprintln!("Connection error: {:?}", connection_result);
                if let Some(found_route) = routes.iter().find(|route| route.name == "DISCONNECT") {
                    let params =&Params::new();
                    (found_route.callback)(&params, &Dispatcher { sender: sender_clone.clone() });
                }
                tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
            }
        });

        Dispatcher { sender }
    }
}