harmos_sidecar/kit/
observer_host.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::future::Future;
18use std::sync::Arc;
19// Third Party Imports
20
21// Local Imports
22use super::{CallError, ObserverContext, PublicationError, Supervisor};
23use harmos_domain::{Message, StreamSpec, inventory::publications};
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33/* === Enums === */
34
35/* === Traits === */
36
37pub trait ObserverHost<R>: Send + Sync + Sized + 'static {
38    fn bind(resources: Arc<R>, supervisor: Supervisor) -> Self;
39}
40
41pub trait ObserverPublisher<R: Sync, V: Send>: Send + Sync + Sized {
42    type Error: std::error::Error + From<PublicationError> + Send + Sync + 'static;
43    const STREAM: Option<publications::Entry> = None;
44
45    fn publish(&self, context: &ObserverContext<'_, R, Self>, value: V) -> impl Future<Output = Result<(), Self::Error>> + Send;
46}
47
48pub trait ObserverStreamPublisher<R: Sync, S: StreamSpec>: Send + Sync + Sized
49where
50    S::Row: Sync,
51{
52    type Error: std::error::Error + From<PublicationError> + Send + Sync + 'static;
53
54    fn publish_on(&self, context: &ObserverContext<'_, R, Self>, row: &S::Row) -> impl Future<Output = Result<(), Self::Error>> + Send;
55}
56
57pub trait ObserverCaller<R: Sync, C>: Send + Sync + Sized {
58    type Request: Send;
59    type Response: Send;
60    type Error: std::error::Error + Send + Sync + 'static;
61
62    fn call(
63        &self, context: &ObserverContext<'_, R, Self>, request: Self::Request,
64    ) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send;
65}
66
67/* === Implementations === */
68
69impl<R> ObserverHost<R> for Supervisor {
70    fn bind(_: Arc<R>, supervisor: Supervisor) -> Self {
71        supervisor
72    }
73}
74
75impl<R: Sync, C> ObserverCaller<R, C> for Supervisor
76where
77    C: harmos_domain::operations::OperationDeclaration,
78{
79    type Request = C;
80    type Response = C::Response;
81    type Error = CallError;
82
83    async fn call(&self, _: &ObserverContext<'_, R, Self>, request: Self::Request) -> Result<Self::Response, Self::Error> {
84        let bytes = self.child_call::<C>(&request, false).await?;
85        C::Response::decode(bytes.as_slice()).map_err(|error| CallError::Malformed { method: C::ID, message: error.to_string() })
86    }
87}
88
89/* === Public Functions === */
90
91/* === Private Functions === */
92
93/* === Tests === */
harmos_sidecar/kit/observer_host.rs

harmos_sidecar/kit/
observer_host.rs
