harmos_sidecar/kit/
plugin.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::marker::PhantomData;
18use std::num::NonZeroUsize;
19use std::sync::Arc;
20// Third Party Imports
21use tokio::io::AsyncWriteExt;
22// Local Imports
23use super::{Opening, Supervisor, catalog, local};
24use crate::inventory::{Declaration, Facility};
25use crate::{Attach, Attached, PluginCatalog, Refusal, StreamDeclaration};
26use harmos_sdk::plugin::{Declared, Owner, Plugin};
27
28/* === Constants === */
29
30/* === Macros === */
31
32/* === Type Aliases === */
33
34/* === Structs === */
35
36pub struct PluginNative<P> {
37    declaration: Declaration,
38    capacity: NonZeroUsize,
39    _plugin: PhantomData<fn() -> P>,
40}
41
42/* === Enums === */
43
44/* === Traits === */
45
46/* === Implementations === */
47
48impl<P: Plugin<Context = Supervisor> + Declared> PluginNative<P> {
49    pub fn new(capacity: NonZeroUsize) -> std::io::Result<Self> {
50        let markers = std::hint::black_box(P::markers().map_err(std::io::Error::other)?);
51        let declaration = Declaration::inspect(&markers.concat(), Facility::Sidecar).map_err(std::io::Error::other)?;
52        harmos_domain::inventory::publications::validate_match(P::STREAMS, &declaration.streams).map_err(std::io::Error::other)?;
53        let manifest = &declaration.contracts;
54        let provides = crate::operations::Contracts::from_tree(P::PROVIDES).map_err(std::io::Error::other)?;
55        let requires = harmos_domain::inventory::contracts::Required::from_tree(P::REQUIRES).map_err(std::io::Error::other)?;
56        let listeners = harmos_domain::listeners::Declarations::from_tree(P::LISTENERS).map_err(std::io::Error::other)?;
57        harmos_domain::inventory::listeners::validate(&listeners, &provides).map_err(std::io::Error::other)?;
58        let pair = [
59            crate::operations::lifecycle::Descriptor::of_initialize::<P>().map_err(std::io::Error::other)?,
60            crate::operations::lifecycle::Descriptor::of_shutdown::<P>().map_err(std::io::Error::other)?,
61        ];
62        if manifest.owner != P::ID || manifest.provides != provides || manifest.requires != requires || manifest.lifecycle != pair {
63            return Err(std::io::Error::other("plugin SDK manifest differs from its authored contracts"));
64        }
65        if declaration.listeners != listeners {
66            return Err(std::io::Error::other("listener manifest differs from authored selection"));
67        }
68        let mut observers = Vec::new();
69        harmos_domain::catalog::try_for_each(P::OBSERVERS, |entry| {
70            observers.push(entry.declaration());
71            Ok::<(), std::io::Error>(())
72        })?;
73        if declaration.observers != observers {
74            return Err(std::io::Error::other("observer manifest differs from authored selection"));
75        }
76        if declaration.sections != P::CONTRIBUTIONS {
77            return Err(std::io::Error::other("contribution sections differ from authored selection"));
78        }
79        Ok(Self { declaration, capacity, _plugin: PhantomData })
80    }
81}
82
83impl<P> Clone for PluginNative<P> {
84    fn clone(&self) -> Self {
85        Self { declaration: self.declaration.clone(), capacity: self.capacity, _plugin: PhantomData }
86    }
87}
88
89impl<P: Plugin<Context = Supervisor> + Declared> Attach for PluginNative<P> {
90    type Error = std::io::Error;
91
92    fn owner(&self) -> &str {
93        &self.declaration.id
94    }
95
96    fn publications(&self) -> Vec<StreamDeclaration> {
97        self.declaration.streams.clone()
98    }
99
100    fn operations(&self) -> harmos_domain::Result<Vec<crate::operations::Definition>> {
101        Ok(self.declaration.operations.clone())
102    }
103
104    fn operation_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::operations::Contract>>> {
105        Ok(Some(self.declaration.contracts.provides.clone()))
106    }
107
108    fn required_contracts(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::inventory::contracts::Required>>> {
109        Ok(Some(self.declaration.contracts.requires.clone()))
110    }
111
112    fn sdk_lifecycle(&self) -> harmos_domain::Result<Option<[crate::operations::lifecycle::Descriptor; 2]>> {
113        Ok(Some(self.declaration.contracts.lifecycle.clone()))
114    }
115
116    fn observers(&self) -> Vec<crate::inventory::observers::Declaration> {
117        self.declaration.observers.clone()
118    }
119
120    fn observe_contract(&self) -> harmos_domain::Result<Option<crate::operations::Contract>> {
121        Ok(self.declaration.observe_contract())
122    }
123
124    fn listener_declarations(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::listeners::Definition>>> {
125        Ok(Some(self.declaration.listeners.clone()))
126    }
127
128    async fn attach(&self) -> std::io::Result<Attached> {
129        Ok(local::binary(opening(self)?))
130    }
131}
132
133/* === Public Functions === */
134
135pub async fn run_plugin<P>(capacity: NonZeroUsize) -> std::io::Result<()>
136where
137    P: Plugin<Context = Supervisor> + Declared,
138{
139    std::hint::black_box(P::markers().map_err(std::io::Error::other)?);
140    if catalog::requested(std::env::args_os().skip(1)) {
141        let bytes = catalog::encode::<P>()?;
142        let mut stdout = tokio::io::stdout();
143        stdout.write_all(&bytes).await?;
144        return stdout.flush().await;
145    }
146    #[cfg(any(unix, windows))]
147    if let Some(control) = crate::attach::control::inherited()? {
148        return serve_plugin_channels::<P>(capacity, (Box::new(tokio::io::stdin()), Box::new(tokio::io::stdout())), control).await;
149    }
150    let _ = capacity;
151    Err(std::io::Error::new(std::io::ErrorKind::Unsupported, "stdio launch requires an attached binary control channel"))
152}
153
154pub async fn serve_plugin_channels<P>(
155    capacity: NonZeroUsize, data: crate::transport::Io, control: crate::transport::Io,
156) -> std::io::Result<()>
157where
158    P: Plugin<Context = Supervisor> + Declared,
159{
160    let native = PluginNative::<P>::new(capacity)?;
161    let opening = Arc::new(tokio::sync::Mutex::new(opening(&native)?));
162    super::connection::run_binary(opening, data, control, std::future::pending()).await
163}
164
165/* === Private Functions === */
166
167fn opening<P>(native: &PluginNative<P>) -> std::io::Result<Opening>
168where
169    P: Plugin<Context = Supervisor> + Declared,
170{
171    let definitions = native.declaration.operations.clone();
172    let contracts = native.declaration.contracts.provides.clone();
173    let registrations = catalog::registrations::<P>().map_err(std::io::Error::other)?;
174    let (owner, table) =
175        Owner::<P>::new_with_observers(contracts, native.capacity, native.declaration.observers.clone()).map_err(std::io::Error::other)?;
176    let interrupted_owner = Arc::clone(&owner);
177    Ok(Opening {
178        operations: table,
179        retirement: Box::new(move || {
180            let owner = Arc::clone(&owner);
181            Box::pin(async move { owner.retire().await.map_err(Refusal::of) })
182        }),
183        quiescence: Box::new(move || {
184            let owner = Arc::clone(&interrupted_owner);
185            Box::pin(async move { owner.quiesce().await.map_err(Refusal::of) })
186        }),
187        catalog: PluginCatalog { registrations, operations: definitions, observers: native.declaration.observers.clone() },
188    })
189}
190
191/* === Tests === */
harmos_sidecar/kit/plugin.rs

harmos_sidecar/kit/
plugin.rs
