oa_gateway/adapters/
mod.rs

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
//! Builds each enabled adapter and spawns its `run` task.
//!
//! The host talks to adapters only through [`oa_gateway_core::Engine`]. This
//! module owns construction and task lifetime; protocol work stays in the
//! adapter crates.

mod dds;
mod loopback;
mod owp;
mod stomp;

use std::sync::Arc;

use oa_gateway_core::Engine;
use oa_gateway_uci::{Schema, ValidateMode};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

use crate::addr::resolve_addr;
use crate::config::Config;
use crate::tls::HostTls;

/// Resolves every listen/connect address, then spawns the adapters that are on.
///
/// Addresses are resolved first so a bad one fails cleanly instead of leaving
/// the earlier adapters already running. An adapter that later fails is
/// logged inside its task; the handle still completes so shutdown can join
/// it.
///
/// `schema` and `validate` are handed to OWP and DDS. STOMP and loopback
/// do not convert or check payloads.
///
/// # Errors
///
/// Returns an error if an enabled adapter's address cannot be resolved, if
/// `owp.on_panic`, `dds.on_panic`, or `stomp.on_panic` is not `abort` or
/// `reconnect`, or if every adapter is disabled.
pub(crate) async fn start(
    config: &Config,
    engine: Arc<Engine>,
    schema: Option<Arc<Schema>>,
    validate: ValidateMode,
    tls: HostTls,
    shutdown: CancellationToken,
) -> Result<Vec<JoinHandle<()>>, String> {
    let owp_bind = if config.owp.enabled {
        Some(resolve_addr("owp.bind", &config.owp.bind).await?)
    } else {
        None
    };
    let stomp_broker = if config.stomp.enabled {
        Some(resolve_addr("stomp.broker", &config.stomp.broker).await?)
    } else {
        None
    };

    let mut handles = Vec::new();

    if config.loopback.enabled {
        handles.push(loopback::start(
            &config.loopback,
            Arc::clone(&engine),
            shutdown.clone(),
        ));
    }

    if let Some(bind) = owp_bind {
        handles.push(owp::start(
            &config.owp,
            bind,
            schema.as_ref(),
            validate,
            tls.owp,
            Arc::clone(&engine),
            shutdown.clone(),
        )?);
    }

    if let Some(broker) = stomp_broker {
        handles.push(stomp::start(
            &config.stomp,
            broker,
            tls.stomp,
            Arc::clone(&engine),
            shutdown.clone(),
        )?);
    }

    if config.dds.enabled {
        handles.push(dds::start(
            &config.dds,
            schema.as_ref(),
            validate,
            engine,
            shutdown,
        )?);
    }

    if handles.is_empty() {
        return Err(
            "no adapters enabled. Add a [loopback], [owp], [stomp], or [dds] section.".into(),
        );
    }

    Ok(handles)
}