scaproust 0.3.2

Nanomsg scalability protocols implementation in rust. Various messaging patterns over pluggable transports
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

// Copyright (c) 2015-2017 Contributors as noted in the AUTHORS file.
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0>
// or the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
// This file may not be copied, modified, or distributed except according to those terms.

use std::rc::Rc;
use std::io::{Result, Error};

use mio::Ready;

use core::Message;
use transport::async::stub::*;
use transport::async::dead::*;
use transport::pipe::{Event, Context};

pub trait PipeState<S : AsyncPipeStub + 'static> {

    fn name(&self) -> &'static str;
    fn open(self: Box<Self>, ctx: &mut Context) -> Box<PipeState<S>> {
        error!("[{:?}] open while {}", ctx, self.name());
        Box::new(Dead)
    }
    fn close(self: Box<Self>, ctx: &mut Context) -> Box<PipeState<S>> {
        error!("[{:?}] close while {}", ctx, self.name());
        Box::new(Dead)
    }
    fn send(self: Box<Self>, ctx: &mut Context, _: Rc<Message>) -> Box<PipeState<S>> {
        error!("[{:?}] send while {}", ctx, self.name());
        Box::new(Dead)
    }
    fn recv(self: Box<Self>, ctx: &mut Context) -> Box<PipeState<S>> {
        error!("[{:?}] recv while {}", ctx, self.name());
        Box::new(Dead)
    }
    fn error(self: Box<Self>, ctx: &mut Context, err: Error) -> Box<PipeState<S>> {
        info!("[{:?}] error while {}: {:?}", ctx, self.name(), err);
        
        ctx.raise(Event::Error(err));

        Box::new(Dead)
    }
    fn ready(self: Box<Self>, ctx: &mut Context, _: Ready) -> Box<PipeState<S>> {
        error!("[{:?}] ready while {}", ctx, self.name());
        Box::new(Dead)
    }
    fn enter(&mut self, _: &mut Context) {
    }
    fn leave(&mut self, _: &mut Context) {
    }
}

pub fn transition<F, T, S>(mut old_state: Box<F>, ctx: &mut Context) -> Box<T> where
    F : PipeState<S>,
    F : Into<T>,
    T : PipeState<S>,
    S : AsyncPipeStub + 'static
{
    old_state.leave(ctx);
    let mut new_state = Into::into(*old_state);
    new_state.enter(ctx);
    Box::new(new_state)
}

pub fn transition_if_ok<F, T, S>(f: Box<F>, ctx: &mut Context, res: Result<()>) -> Box<PipeState<S>> where
    F : PipeState<S>,
    F : Into<T>,
    T : PipeState<S> + 'static,
    S : AsyncPipeStub + 'static
{
    match res {
        Ok(..) => transition::<F, T, S>(f, ctx),
        Err(e) => f.error(ctx, e)
    }
}

pub fn no_transition_if_ok<F, S>(f: Box<F>, ctx: &mut Context, res: Result<()>) -> Box<PipeState<S>> where
    F : PipeState<S> + 'static,
    S : AsyncPipeStub + 'static
{
    match res {
        Ok(..) => f,
        Err(e) => f.error(ctx, e)
    }
}