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
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

// 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::collections::HashSet;
use std::rc::Rc;
use std::sync::mpsc::Sender;

use core::{EndpointId, Message};
use core::socket::{Protocol, Reply};
use core::endpoint::Pipe;
use core::context::{Context, Event};
use super::pipes::PipeCollection;
use super::{Timeout, PUB, SUB};
use super::policy::broadcast;
use io_error::*;

pub struct Pub {
    reply_tx: Sender<Reply>,
    pipes: PipeCollection,
    bc: HashSet<EndpointId>
}

/*****************************************************************************/
/*                                                                           */
/* Pub                                                                       */
/*                                                                           */
/*****************************************************************************/

impl From<Sender<Reply>> for Pub {
    fn from(tx: Sender<Reply>) -> Pub {
        Pub {
            reply_tx: tx,
            pipes: PipeCollection::new(),
            bc: HashSet::new()
        }
    }
}

/*****************************************************************************/
/*                                                                           */
/* Protocol                                                                  */
/*                                                                           */
/*****************************************************************************/

impl Protocol for Pub {
    fn id(&self)      -> u16 { PUB }
    fn peer_id(&self) -> u16 { SUB }

    fn add_pipe(&mut self, _: &mut Context, eid: EndpointId, pipe: Pipe) {
        self.pipes.insert(eid, pipe);
    }
    fn remove_pipe(&mut self, ctx: &mut Context, eid: EndpointId) -> Option<Pipe> {
        self.bc.remove(&eid);
        if self.bc.is_empty() {
            ctx.raise(Event::CanSend(false));
        }
        self.pipes.remove(&eid)
    }
    fn send(&mut self, ctx: &mut Context, msg: Message, timeout: Timeout) {
        let msg = Rc::new(msg);

        broadcast::send_to_all(&mut self.bc, &mut self.pipes, ctx, msg);
        ctx.raise(Event::CanSend(false));

        let _ = self.reply_tx.send(Reply::Send);
        if let Some(sched) = timeout {
            ctx.cancel(sched);
        }
    }
    fn on_send_ack(&mut self, _: &mut Context, _: EndpointId) {
    }
    fn on_send_timeout(&mut self, _: &mut Context) {
    }
    fn on_send_ready(&mut self, ctx: &mut Context, eid: EndpointId) {
        if self.bc.is_empty() {
            ctx.raise(Event::CanSend(true));
        }
        self.bc.insert(eid);
    }
    fn on_send_not_ready(&mut self, ctx: &mut Context, eid: EndpointId) {
        let was_empty = self.bc.is_empty();

        self.bc.remove(&eid);

        let is_empty = self.bc.is_empty();
        if was_empty ^ is_empty {
            ctx.raise(Event::CanSend(false));
        }
    }
    fn recv(&mut self, ctx: &mut Context, timeout: Timeout) {
        let error = other_io_error("Recv is not supported by pub protocol");
        let _ = self.reply_tx.send(Reply::Err(error));
        if let Some(sched) = timeout {
            ctx.cancel(sched);
        }
    }
    fn on_recv_ack(&mut self, _: &mut Context, _: EndpointId, _: Message) {
    }
    fn on_recv_timeout(&mut self, _: &mut Context) {
    }
    fn on_recv_ready(&mut self, _: &mut Context, _: EndpointId) {
    }
    fn on_recv_not_ready(&mut self, _: &mut Context, _: EndpointId) {
    }
    fn is_send_ready(&self) -> bool {
        !self.bc.is_empty()
    }
    fn is_recv_ready(&self) -> bool {
        false
    }
    fn close(&mut self, ctx: &mut Context) {
        self.pipes.close_all(ctx)
    }
}