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, Serving, 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            harmos_domain::model::operations::lifecycle::Descriptor::of_initialize::<P>().map_err(std::io::Error::other)?,
60            harmos_domain::model::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 {
80            declaration,
81            capacity,
82            _plugin: PhantomData,
83        })
84    }
85}
86
87impl<P> Clone for PluginNative<P> {
88    fn clone(&self) -> Self {
89        Self {
90            declaration: self.declaration.clone(),
91            capacity: self.capacity,
92            _plugin: PhantomData,
93        }
94    }
95}
96
97impl<P: Plugin<Context = Supervisor> + Declared> Attach for PluginNative<P> {
98    type Error = std::io::Error;
99
100    fn owner(&self) -> &str {
101        &self.declaration.id
102    }
103
104    fn publications(&self) -> Vec<StreamDeclaration> {
105        self.declaration.streams.clone()
106    }
107
108    fn operations(&self) -> harmos_domain::Result<Vec<crate::operations::OperationDefinition>> {
109        Ok(self.declaration.operations.clone())
110    }
111
112    fn operation_contracts(&self) -> harmos_domain::Result<Option<Vec<crate::operations::Contract>>> {
113        Ok(Some(self.declaration.contracts.provides.clone()))
114    }
115
116    fn required_contracts(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::inventory::contracts::Required>>> {
117        Ok(Some(self.declaration.contracts.requires.clone()))
118    }
119
120    fn sdk_lifecycle(&self) -> harmos_domain::Result<Option<[harmos_domain::model::operations::lifecycle::Descriptor; 2]>> {
121        Ok(Some(self.declaration.contracts.lifecycle.clone()))
122    }
123
124    fn observers(&self) -> Vec<crate::inventory::observers::Declaration> {
125        self.declaration.observers.clone()
126    }
127
128    fn observe_contract(&self) -> harmos_domain::Result<Option<crate::operations::Contract>> {
129        Ok(self.declaration.observe_contract())
130    }
131
132    fn listener_declarations(&self) -> harmos_domain::Result<Option<Vec<harmos_domain::listeners::Definition>>> {
133        Ok(Some(self.declaration.listeners.clone()))
134    }
135
136    async fn attach(&self) -> std::io::Result<Attached> {
137        Ok(local::binary(opening(self)?))
138    }
139}
140
141/* === Public Functions === */
142
143pub async fn run_plugin<P>(capacity: NonZeroUsize) -> std::io::Result<()>
144where
145    P: Plugin<Context = Supervisor> + Declared,
146{
147    std::hint::black_box(P::markers().map_err(std::io::Error::other)?);
148    if catalog::requested(std::env::args_os().skip(1)) {
149        let bytes = catalog::encode::<P>()?;
150        let mut stdout = tokio::io::stdout();
151        stdout.write_all(&bytes).await?;
152        return stdout.flush().await;
153    }
154    #[cfg(any(unix, windows))]
155    if let Some(control) = crate::attach::control::inherited()? {
156        return serve_plugin_channels::<P>(capacity, (Box::new(tokio::io::stdin()), Box::new(tokio::io::stdout())), control).await;
157    }
158    let _ = capacity;
159    Err(std::io::Error::new(std::io::ErrorKind::Unsupported, "stdio launch requires an attached binary control channel"))
160}
161
162pub async fn serve_plugin_channels<P>(
163    capacity: NonZeroUsize, data: crate::transport::Io, control: crate::transport::Io,
164) -> std::io::Result<()>
165where
166    P: Plugin<Context = Supervisor> + Declared,
167{
168    let native = PluginNative::<P>::new(capacity)?;
169    let opening = Arc::new(tokio::sync::Mutex::new(opening(&native)?));
170    super::connection::run_binary(opening, data, control, std::future::pending()).await
171}
172
173/* === Private Functions === */
174
175fn opening<P>(native: &PluginNative<P>) -> std::io::Result<Opening>
176where
177    P: Plugin<Context = Supervisor> + Declared,
178{
179    let definitions = native.declaration.operations.clone();
180    let contracts = native.declaration.contracts.provides.clone();
181    let registrations = catalog::registrations::<P>().map_err(std::io::Error::other)?;
182    let capacity = native.capacity;
183    let observers = native.declaration.observers.clone();
184    Ok(Opening {
185        serving: None,
186        prepare: Box::new(move |assigned_owner| {
187            let (owner, table) =
188                Owner::<P>::new_with_observers(assigned_owner, contracts.clone(), capacity, observers.clone()).map_err(Refusal::of)?;
189            harmos_operations::lifecycle::bind(owner.clone(), &table).map_err(Refusal::of)?;
190            let interrupted_owner = Arc::clone(&owner);
191            Ok(Serving {
192                owner: assigned_owner.to_owned(),
193                operations: table,
194                retirement: Box::new(move || {
195                    let owner = Arc::clone(&owner);
196                    Box::pin(async move { owner.retire().await.map_err(Refusal::of) })
197                }),
198                quiescence: Box::new(move || {
199                    let owner = Arc::clone(&interrupted_owner);
200                    Box::pin(async move { owner.quiesce().await.map_err(Refusal::of) })
201                }),
202            })
203        }),
204        catalog: PluginCatalog {
205            registrations,
206            operations: definitions,
207            observers: native.declaration.observers.clone(),
208        },
209    })
210}
211
212/* === Tests === */
harmos_sidecar/kit/plugin.rs

harmos_sidecar/kit/
plugin.rs
