kompact 0.6.0

Kompact is a Rust implementation of the Kompics component model combined with the Actor model.
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

use crate::net::ConnectionState;
use futures::sync;
use spaniel::frames::Frame;
use std::collections::HashMap;
use std::collections::VecDeque;
use std::net::SocketAddr;

/// Wrapper around a hashmap of frame queues.
///
/// Used when waiting for connections to establish and drained when possible.
pub struct QueueManager {
    inner: HashMap<SocketAddr, VecDeque<Frame>>,
}

impl QueueManager {
    pub fn new() -> Self {
        QueueManager {
            inner: HashMap::new(),
        }
    }

    /// Appends the given frame onto the SocketAddr's queue
    pub fn enqueue_frame(&mut self, frame: Frame, dst: SocketAddr) {
        self.inner
            .entry(dst)
            .or_insert(VecDeque::new())
            .push_back(frame);
    }

    /// Extracts the next queue-up frame for the SocketAddr, if one exists
    ///
    /// If the SocketAddr exists but its queue is empty, the entry is removed.
    pub fn pop_frame(&mut self, dst: &SocketAddr) -> Option<Frame> {
        let res = self.inner.get_mut(dst).and_then(|q| q.pop_back());
        if self.inner.contains_key(dst) && res.is_none() {
            self.inner.remove(dst);
        }
        res
    }

    /// Attempts to drain all Frame entries stored for the provided SocketAddr into the Sender
    pub fn try_drain(
        &mut self,
        dst: SocketAddr,
        tx: &mut sync::mpsc::UnboundedSender<Frame>,
    ) -> Option<ConnectionState> {
        while let Some(frame) = self.pop_frame(&dst) {
            if let Err(err) = tx.unbounded_send(frame) {
                let next = Some(ConnectionState::Closed);

                // Consume error and retrieve failed Frame
                let frame = err.into_inner();
                self.enqueue_frame(frame, dst);

                // End looping, return next state
                return next;
            }
        }
        None
    }

    pub fn exists(&self, dst: &SocketAddr) -> bool {
        self.inner.contains_key(dst)
    }

    pub fn has_frame(&self, dst: &SocketAddr) -> bool {
        self.inner.get(dst).map_or(false, |q| !q.is_empty())
    }
}