asteroid-mq 0.1.0-alpha.3

An embeddable message queue system
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

use std::{
    borrow::Cow,
    collections::{HashMap, HashSet},
    sync::Arc,
};

use super::EdgePayload;
pub(crate) mod bincode;
pub use bincode::*;
#[cfg(feature = "cbor")]
pub(crate) mod cbor;
#[cfg(feature = "cbor")]
pub use cbor::*;
pub(crate) mod json;
pub use asteroid_mq_model::CodecKind;
pub use json::*;

#[derive(Debug)]
pub struct CodecError {
    reason: Cow<'static, str>,
}

impl std::fmt::Display for CodecError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "codec error: {}", self.reason)
    }
}

impl CodecError {
    pub fn decode_error<E: std::fmt::Display>(e: E) -> Self {
        Self {
            reason: format!("decode error: {}", e).into(),
        }
    }
    pub fn unregistered_codec(codec: CodecKind) -> Self {
        Self {
            reason: format!("unregistered codec: {}", codec).into(),
        }
    }
}

pub trait Codec: Send + Sync + 'static {
    fn encode(&self, value: &EdgePayload) -> Vec<u8>;
    fn decode(&self, bytes: &[u8]) -> Result<EdgePayload, CodecError>;
}
pub struct CodecRegistry {
    registry: HashMap<CodecKind, Arc<dyn Codec>>,
}

impl std::fmt::Debug for CodecRegistry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_set().entries(self.registry.keys()).finish()
    }
}

impl CodecRegistry {
    pub fn new_empty() -> Self {
        Self {
            registry: HashMap::new(),
        }
    }
    pub fn new_preloaded() -> Self {
        let mut registry = Self::new_empty();
        #[cfg(feature = "cbor")]
        registry.register(CodecKind::CBOR, cbor::Cbor);
        registry.register(CodecKind::BINCODE, bincode::Bincode);
        registry.register(CodecKind::JSON, json::Json);
        registry
    }
    pub fn register<C: Codec + 'static>(&mut self, kind: CodecKind, codec: C) {
        self.registry.insert(kind, Arc::new(codec));
    }
    pub fn get(&self, kind: &CodecKind) -> Option<Arc<dyn Codec>> {
        self.registry.get(kind).cloned()
    }
    pub fn pick_preferred_codec(&self, supported: &HashSet<CodecKind>) -> Option<CodecKind> {
        supported
            .iter()
            .find(|kind| self.registry.contains_key(kind))
            .copied()
    }
    pub fn encode(&self, kind: CodecKind, value: &EdgePayload) -> Result<Vec<u8>, CodecError> {
        self.registry
            .get(&kind)
            .ok_or_else(|| CodecError::unregistered_codec(kind))
            .map(|codec| codec.encode(value))
    }
    pub fn decode(&self, kind: CodecKind, bytes: &[u8]) -> Result<EdgePayload, CodecError> {
        self.registry
            .get(&kind)
            .ok_or_else(|| CodecError::unregistered_codec(kind))
            .and_then(|codec| codec.decode(bytes))
    }
}