rain_server 0.2.2-pre

Distributed computational framework for large-scale task-based pipelines. Server and governor binaries.
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

use bytes::BytesMut;
use futures::{Future, Stream};
use tokio_io::codec::length_delimited::{Builder, Framed};
use tokio_io::AsyncRead;
use tokio_io::AsyncWrite;

use rain_core::errors::{Error, Result};

pub type SendType = Vec<u8>;

#[derive(Debug)]
pub struct Sender {
    channel_sender: ::futures::unsync::mpsc::UnboundedSender<SendType>,
}

impl Sender {
    pub fn send(&self, data: SendType) {
        self.channel_sender.unbounded_send(data).unwrap()
    }
}

pub fn create_protocol_stream<S>(stream: S) -> Framed<S, SendType>
where
    S: AsyncRead + AsyncWrite + 'static,
{
    Builder::new()
        .little_endian()
        .max_frame_length(128 * 1024 * 1024 /*128 MB*/)
        .new_framed(stream)
}

pub struct Connection<S>
where
    S: AsyncRead + AsyncWrite + 'static,
{
    stream: Framed<S, SendType>,
    channel_receiver: ::futures::unsync::mpsc::UnboundedReceiver<SendType>,
    channel_sender: ::futures::unsync::mpsc::UnboundedSender<SendType>,
}

impl<S> Connection<S>
where
    S: AsyncRead + AsyncWrite + 'static,
{
    pub fn from(stream: Framed<S, SendType>) -> Self {
        let (channel_sender, channel_receiver) = ::futures::unsync::mpsc::unbounded();
        Connection {
            stream,
            channel_receiver,
            channel_sender,
        }
    }

    pub fn sender(&self) -> Sender {
        Sender {
            channel_sender: self.channel_sender.clone(),
        }
    }

    /*
    pub fn start<OnMessage, OnError>(self,
                 handle: &Handle,
                 on_message: OnMessage,
                 on_error: OnError) where OnMessage: Fn(BytesMut) -> Result<()> + 'static,
                                          OnError: FnOnce(Error) + 'static,
 {
        let Connection {
            stream,
            channel_receiver: receiver,
            ..
        } = self;
        let (write, read) = stream.split();
        let send_future = receiver.map_err(
            |_| panic!("Send channel failed!")).forward(write).map(|_| ());
        let read_future = read.map_err(|e| e.into()).for_each(on_message);
        let future = read_future.select(send_future).map(
            |_| { panic!("Executor connection closed") }).map_err(|(e, f)| on_error(e));
        handle.spawn(future);
    }*/

    pub fn start_future<OnMessage>(
        self,
        on_message: OnMessage,
    ) -> Box<Future<Item = (), Error = Error>>
    where
        OnMessage: Fn(BytesMut) -> Result<()> + 'static,
    {
        let Connection {
            stream,
            channel_receiver: receiver,
            ..
        } = self;
        let (write, read) = stream.split();
        let send_future = receiver
            .map_err(|_| panic!("Send channel failed!"))
            .forward(write)
            .map(|_| ());
        let read_future = read.map_err(|e| e.into()).for_each(on_message);
        Box::new(
            read_future
                .select(send_future)
                .map(|_| ())
                .map_err(|(e, _)| e),
        )
    }
}