swindon 0.7.8

An HTTP edge (frontend) server with smart websockets support 
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
113

use std::time::Instant;
use std::sync::mpsc::Receiver;
use std::collections::HashMap;

use super::{Event, Action};
use super::pool::Pool;
use super::try_iter::try_iter;
use metrics::Integer;

lazy_static! {
    pub static ref SESSION_POOLS: Integer = Integer::new();
}

fn pool_action(pool: &mut Pool, ts: Instant, action: Action) {
    use super::Action::*;
    match action {
        // handled earlier
        NewSessionPool {..} => unreachable!(),
        StopSessionPool => unreachable!(),
        // Connection management
        NewConnection { conn_id, channel } => {
            pool.add_connection(conn_id, channel);
        }
        Associate { session_id, conn_id, metadata } => {
            pool.associate(conn_id, session_id, ts, metadata);
        }
        UpdateActivity { session_id, timestamp } => {
            pool.update_activity(session_id, timestamp);
        }
        Disconnect { conn_id } => {
            pool.del_connection(conn_id);
        }
        // Subscriptions
        Subscribe { conn_id, topic } => {
            pool.subscribe(conn_id, topic);
        }
        Unsubscribe { conn_id, topic } => {
            pool.unsubscribe(conn_id, topic);
        }
        Publish { topic, data } => {
            pool.publish(topic, data);
        }
        // Lattices
        Attach { conn_id, namespace } => {
            pool.lattice_attach(conn_id, namespace);
        }
        Lattice { namespace, delta } => {
            pool.lattice_update(namespace, delta);
        }
        Detach { conn_id, namespace } => {
            pool.lattice_detach(conn_id, namespace);
        }
        AttachUsers { conn_id, list } => {
            pool.users_attach(conn_id, list);
        }
        UpdateUsers { session_id, list } => {
            pool.users_update(session_id, list);
        }
        DetachUsers { conn_id } => {
            pool.users_detach(conn_id);
        }
    }
}

pub fn run(rx: Receiver<Event>) {
    use super::Action::*;
    use std::sync::mpsc::RecvTimeoutError::*;

    let mut pools = HashMap::new();

    loop {
        let now = Instant::now();
        let timeout = pools.iter_mut()
            .map(|(_, pool): (_, &mut Pool)| pool.cleanup(now))
            .flat_map(|x| x)
            .min();
        let result = match timeout {
            Some(t) => rx.recv_timeout(t.duration_since(now)),
            None => rx.recv().map_err(|_| Disconnected),
        };
        let value = match result {
            Ok(x) => Some(x),
            Err(Timeout) => continue,
            Err(Disconnected) => {
                panic!("Process pools are not needed for anyone");
            }
        };
        for msg in value.into_iter().chain(try_iter(&rx)) {
            let Event { timestamp, action, pool } = msg;
            debug!("Received action {:?} {:?}", pool, action);
            match action {
                // Pool management
                NewSessionPool { config, channel } => {
                    pools.insert(pool.clone(),
                        Pool::new(pool, config, channel));
                    SESSION_POOLS.set(pools.len() as i64);
                }
                StopSessionPool => {
                    if let Some(pool) = pools.remove(&pool) {
                        pool.stop();
                        SESSION_POOLS.set(pools.len() as i64);
                    }
                }
                _ => {
                    // For all other actions we resolve pool first
                    pools.get_mut(&pool)
                    .map(|p| pool_action(p, timestamp, action))
                    .unwrap_or_else(|| debug!("Undefined pool {:?}", pool))
                }
            }
        }
    }
}