proxima-lib 0.1.0

A sans-IO core where everything is a Pipe — one async In -> Result<Out, Err> trait worn four ways (transform, source, sink, observe), composed into proxies, gateways, caches, codecs and telemetry pipelines. Config-driven, runtime-agnostic: the same pipe runs on the from-scratch Prime per-core runtime, on tokio, or on no_std bare metal.
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

//! `PipeFactory` + [`ClientProtocol`] for the `mqtt` protocol terminal — a
//! `proxima::Client` transport that speaks MQTT v3.1.1.
//!
//! Reached via the `type` discriminator (`{"type":"mqtt", "dsn":
//! "mqtt://user:pass@broker:1883"}` or the field form) or the `.mqtt(dsn)`
//! builder sugar (`ClientProtocolExt`), which lowers to
//! `.protocol(MqttClientProtocol::dsn(dsn))`. Composes the sans-IO MQTT
//! client ([`MqttClientUpstream`](proxima_mqtt::MqttClientUpstream)) over
//! the prime TCP transport ([`PrimeTcpUpstream`](crate::PrimeTcpUpstream)),
//! exactly like the prime `kafka`/`pgwire`/`redis` factories.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use serde_json::Value;

use proxima_mqtt::{MqttClientConfig, MqttClientUpstream};
use proxima_primitives::pipe::handler::{PipeHandle, into_handle};
use proxima_primitives::pipe::pipe_factory::PipeFactory;

use crate::PrimeTcpUpstream;
use crate::client::handle::ClientProtocol;
use crate::error::ProximaError;

/// A [`PipeFactory`] for the `mqtt` key. Builds a client `Pipe` from a
/// [`MqttClientConfig`] parsed out of the spec.
#[derive(Debug, Default)]
pub struct MqttPipeFactory;

impl MqttPipeFactory {
    #[must_use]
    pub fn new() -> Self {
        Self
    }
}

impl PipeFactory for MqttPipeFactory {
    fn name(&self) -> &str {
        "mqtt"
    }

    fn build(
        &self,
        spec: &Value,
        _inner: Option<PipeHandle>,
    ) -> Pin<Box<dyn Future<Output = Result<PipeHandle, ProximaError>> + Send + '_>> {
        let spec = spec.clone();
        Box::pin(async move {
            let config = config_from_spec(&spec)?;
            let upstream = PrimeTcpUpstream::with_host(config.host.clone(), config.port);
            Ok(into_handle(MqttClientUpstream::new(upstream, config)))
        })
    }
}

/// Parse a [`MqttClientConfig`] from the spec: prefer a `dsn` string, else
/// deserialize the field form (serde ignores the `type` discriminator).
fn config_from_spec(spec: &Value) -> Result<MqttClientConfig, ProximaError> {
    if let Some(dsn) = spec.get("dsn").and_then(Value::as_str) {
        return MqttClientConfig::from_dsn(dsn)
            .map_err(|err| ProximaError::Config(format!("mqtt dsn: {err}")));
    }
    serde_json::from_value(spec.clone())
        .map_err(|err| ProximaError::Config(format!("mqtt config: {err}")))
}

/// The out-of-crate [`ClientProtocol`] a `.mqtt(dsn)` builder call merges.
pub struct MqttClientProtocol {
    dsn: String,
}

impl MqttClientProtocol {
    /// Point at an MQTT broker by DSN (`mqtt://[user:pass@]broker[:port]`).
    #[must_use]
    pub fn dsn(dsn: impl Into<String>) -> Self {
        Self { dsn: dsn.into() }
    }
}

impl ClientProtocol for MqttClientProtocol {
    fn spec(&self) -> Value {
        serde_json::json!({"type": "mqtt", "dsn": self.dsn})
    }

    fn factory(&self) -> Arc<dyn PipeFactory> {
        Arc::new(MqttPipeFactory::new())
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    #[test]
    fn config_from_dsn_spec() {
        let spec = serde_json::json!({ "type": "mqtt", "dsn": "mqtt://broker:1884" });
        let config = config_from_spec(&spec).expect("config");
        assert_eq!((config.host.as_str(), config.port), ("broker", 1884));
    }

    #[test]
    fn factory_name_is_the_spec_key() {
        assert_eq!(MqttPipeFactory::new().name(), "mqtt");
    }

    #[test]
    fn client_protocol_lowers_to_the_type_and_dsn_spec() {
        let protocol = MqttClientProtocol::dsn("mqtt://broker:1883");
        let spec = protocol.spec();
        assert_eq!(spec["type"], "mqtt");
        assert_eq!(spec["dsn"], "mqtt://broker:1883");
        assert_eq!(protocol.factory().name(), "mqtt");
    }
}