swindon 0.5.2

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

use std::sync::Arc;

use futures::Future;
use futures::stream::{Stream};
use futures::{AsyncSink, Sink};
use futures::sync::oneshot::{channel as oneshot, Sender};
use futures::sync::mpsc::{UnboundedReceiver as Receiver};
use tokio_core::reactor::Handle;

use http_pools::{REQUESTS, FAILED_503};
use runtime::Runtime;
use config::SessionPool;
use chat::Shutdown;
use chat::backend;
use chat::processor::{PoolMessage};


pub fn run(runtime: &Arc<Runtime>, settings: &Arc<SessionPool>,
           handle: &Handle, stream: Receiver<PoolMessage>)
    -> Sender<Shutdown>
{
    use chat::processor::PoolMessage::*;
    let mut handlers = Vec::new();
    let runtime = runtime.clone();
    for dest in &settings.inactivity_handlers {
        let path = if dest.path == "/" {
            Arc::new("/tangle/session_inactive".to_string())
        } else {
            Arc::new(dest.path.to_string() + "/tangle/session_inactive")
        };
        handlers.push((path, dest.upstream.clone()));
    }
    let (tx, rx) = oneshot();
    handle.spawn(stream.for_each(move |msg| {
            match msg {
                InactiveSession { session_id, .. } => {
                    info!("Sending inactivity: {:?}", session_id);
                    for &(ref path, ref upname) in &handlers {
                        let config = runtime.config.get();
                        let dest_settings = match
                            config.http_destinations.get(upname)
                        {
                            Some(x) => x,
                            None => {
                                error!("No such destination {:?} \
                                    for sending inactivity",
                                    upname);
                                continue;
                            }
                        };
                        let mut up = runtime.http_pools.upstream(&upname);
                        let codec = Box::new(backend::InactivityCodec::new(
                            path, &session_id, dest_settings));
                        match up.get_mut().get_mut() {
                            Some(pool) => {
                                match pool.start_send(codec) {
                                    Ok(AsyncSink::NotReady(_)) => {
                                        FAILED_503.incr(1);
                                        // TODO(tailhook) retry later
                                        warn!("Coudn't send inactivity");
                                    }
                                    Ok(AsyncSink::Ready) => {
                                        REQUESTS.incr(1);
                                        debug!("Sent /tangle/session_inactive");
                                    }
                                    Err(e) => {
                                        // TODO(tailhook) log, retry later
                                        error!("Error sending inactivity: {}",
                                            e);
                                    }
                                }
                            }
                            None => {
                                error!("No such destination {:?} \
                                    for sending inactivity",
                                    upname);
                            }
                        }
                    }
                }
            }
            Ok(())
        })
        .select(rx.then(move |_| Ok(())))
        .map(move |(_, _)| info!("Inactivity handler exited"))
        .map_err(move |(_, _)| info!("Inactivity handler exited")));
    return tx;
}