rotor_stream/
accept.rs

1use std::any::Any;
2
3use rotor::{Machine, Response, EventSet, PollOpt, Evented};
4use rotor::{Scope, GenericScope, Void};
5use rotor::mio::{TryAccept};
6
7use {StreamSocket, Accept};
8
9
10/// Trait which must be implemented for a state machine to accept connection
11///
12/// This basically provides alternative constructor for the state machine.
13pub trait Accepted: Machine {
14    type Seed: Clone;
15    type Socket: StreamSocket;
16    /// The constructor of the state machine from the accepted connection
17    fn accepted(sock: Self::Socket, seed: <Self as Accepted>::Seed,
18        scope: &mut Scope<Self::Context>)
19        -> Response<Self, Void>;
20}
21
22
23impl<M, A> Accept<M, A>
24    where A: TryAccept<Output=M::Socket> + Evented + Any,
25          M: Accepted,
26{
27    pub fn new<S: GenericScope>(sock: A,
28        seed: <M as Accepted>::Seed, scope: &mut S)
29        -> Response<Self, Void>
30    {
31        match scope.register(&sock, EventSet::readable(), PollOpt::edge()) {
32            Ok(()) => {}
33            Err(e) => return Response::error(Box::new(e)),
34        }
35        Response::ok(Accept::Server(sock, seed))
36    }
37}
38
39impl<M, A> Machine for Accept<M, A>
40    where A: TryAccept<Output=M::Socket> + Evented + Any,
41          M: Accepted,
42{
43    type Context = M::Context;
44    type Seed = (A::Output, <M as Accepted>::Seed);
45    fn create((sock, seed): Self::Seed, scope: &mut Scope<Self::Context>)
46        -> Response<Self, Void>
47    {
48        M::accepted(sock, seed, scope).wrap(Accept::Connection)
49    }
50
51    fn ready(self, events: EventSet, scope: &mut Scope<Self::Context>)
52        -> Response<Self, Self::Seed>
53    {
54        match self {
55            Accept::Server(a, s) => {
56                match a.accept() {
57                    Ok(Some(sock)) => {
58                        let seed = (sock, s.clone());
59                        Response::spawn(Accept::Server(a, s), seed)
60                    }
61                    Ok(None) =>  {
62                        Response::ok(Accept::Server(a, s))
63                    }
64                    Err(_) => {
65                        // TODO(tailhook) maybe log the error
66                        Response::ok(Accept::Server(a, s))
67                    }
68                }
69            }
70            Accept::Connection(m) => {
71                m.ready(events, scope)
72                    .map(Accept::Connection, |_| unreachable!())
73            }
74        }
75    }
76
77    fn spawned(self, _scope: &mut Scope<Self::Context>)
78        -> Response<Self, Self::Seed>
79    {
80        match self {
81            Accept::Server(a, s) => {
82                match a.accept() {
83                    Ok(Some(sock)) => {
84                        let seed = (sock, s.clone());
85                        Response::spawn(Accept::Server(a, s), seed)
86                    }
87                    Ok(None) =>  {
88                        Response::ok(Accept::Server(a, s))
89                    }
90                    Err(_) => {
91                        // TODO(tailhook) maybe log the error
92                        Response::ok(Accept::Server(a, s))
93                    }
94                }
95            }
96            Accept::Connection(_) => {
97                unreachable!();
98            }
99        }
100    }
101
102    fn timeout(self, scope: &mut Scope<Self::Context>)
103        -> Response<Self, Self::Seed>
104    {
105        match self {
106            Accept::Server(..) => unreachable!(),
107            Accept::Connection(m) => {
108                m.timeout(scope).map(Accept::Connection, |_| unreachable!())
109            }
110        }
111    }
112
113    fn wakeup(self, scope: &mut Scope<Self::Context>)
114        -> Response<Self, Self::Seed>
115    {
116        match self {
117            me @ Accept::Server(..) => Response::ok(me),
118            Accept::Connection(m) => {
119                m.wakeup(scope).map(Accept::Connection, |_| unreachable!())
120            }
121        }
122    }
123}
rotor_stream/accept.rs

rotor_stream/
accept.rs
