rotor-stream 0.6.2

The stream abstration on top of rotor. This is the easiest way to build a protocol on top of TCP (Unix sockets) with rotor
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
113
114
115
116
117
118
119
120
121
122
123

use std::any::Any;

use rotor::{Machine, Response, EventSet, PollOpt, Evented};
use rotor::{Scope, GenericScope, Void};
use rotor::mio::{TryAccept};

use {StreamSocket, Accept};


/// Trait which must be implemented for a state machine to accept connection
///
/// This basically provides alternative constructor for the state machine.
pub trait Accepted: Machine {
    type Seed: Clone;
    type Socket: StreamSocket;
    /// The constructor of the state machine from the accepted connection
    fn accepted(sock: Self::Socket, seed: <Self as Accepted>::Seed,
        scope: &mut Scope<Self::Context>)
        -> Response<Self, Void>;
}


impl<M, A> Accept<M, A>
    where A: TryAccept<Output=M::Socket> + Evented + Any,
          M: Accepted,
{
    pub fn new<S: GenericScope>(sock: A,
        seed: <M as Accepted>::Seed, scope: &mut S)
        -> Response<Self, Void>
    {
        match scope.register(&sock, EventSet::readable(), PollOpt::edge()) {
            Ok(()) => {}
            Err(e) => return Response::error(Box::new(e)),
        }
        Response::ok(Accept::Server(sock, seed))
    }
}

impl<M, A> Machine for Accept<M, A>
    where A: TryAccept<Output=M::Socket> + Evented + Any,
          M: Accepted,
{
    type Context = M::Context;
    type Seed = (A::Output, <M as Accepted>::Seed);
    fn create((sock, seed): Self::Seed, scope: &mut Scope<Self::Context>)
        -> Response<Self, Void>
    {
        M::accepted(sock, seed, scope).wrap(Accept::Connection)
    }

    fn ready(self, events: EventSet, scope: &mut Scope<Self::Context>)
        -> Response<Self, Self::Seed>
    {
        match self {
            Accept::Server(a, s) => {
                match a.accept() {
                    Ok(Some(sock)) => {
                        let seed = (sock, s.clone());
                        Response::spawn(Accept::Server(a, s), seed)
                    }
                    Ok(None) =>  {
                        Response::ok(Accept::Server(a, s))
                    }
                    Err(_) => {
                        // TODO(tailhook) maybe log the error
                        Response::ok(Accept::Server(a, s))
                    }
                }
            }
            Accept::Connection(m) => {
                m.ready(events, scope)
                    .map(Accept::Connection, |_| unreachable!())
            }
        }
    }

    fn spawned(self, _scope: &mut Scope<Self::Context>)
        -> Response<Self, Self::Seed>
    {
        match self {
            Accept::Server(a, s) => {
                match a.accept() {
                    Ok(Some(sock)) => {
                        let seed = (sock, s.clone());
                        Response::spawn(Accept::Server(a, s), seed)
                    }
                    Ok(None) =>  {
                        Response::ok(Accept::Server(a, s))
                    }
                    Err(_) => {
                        // TODO(tailhook) maybe log the error
                        Response::ok(Accept::Server(a, s))
                    }
                }
            }
            Accept::Connection(_) => {
                unreachable!();
            }
        }
    }

    fn timeout(self, scope: &mut Scope<Self::Context>)
        -> Response<Self, Self::Seed>
    {
        match self {
            Accept::Server(..) => unreachable!(),
            Accept::Connection(m) => {
                m.timeout(scope).map(Accept::Connection, |_| unreachable!())
            }
        }
    }

    fn wakeup(self, scope: &mut Scope<Self::Context>)
        -> Response<Self, Self::Seed>
    {
        match self {
            me @ Accept::Server(..) => Response::ok(me),
            Accept::Connection(m) => {
                m.wakeup(scope).map(Accept::Connection, |_| unreachable!())
            }
        }
    }
}