harmos-guest 0.1.13

Harmos guest kit for WebAssembly component plugins.
Documentation 
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

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::marker::PhantomData;
// Third Party Imports

// Local Imports
use super::Context;
use harmos_domain::{Message, StreamSpec, plugin::Declared};
use harmos_sdk::operations::Host;

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

pub struct Publisher<P> {
    context: Context,
    _plugin: PhantomData<fn() -> P>,
}

/* === Enums === */

#[derive(Debug, thiserror::Error)]
pub enum PublishError {
    #[error("stream is not in the authored inventory")]
    UndeclaredStream,

    #[error(transparent)]
    Call(#[from] harmos_domain::Refusal),
}

/* === Traits === */

/* === Implementations === */

// A refused publication forwards the host's failure unchanged.
impl From<PublishError> for harmos_domain::Refusal {
    fn from(error: PublishError) -> Self {
        match error {
            PublishError::Call(refusal) => refusal,
            error => Self::of(error),
        }
    }
}

impl<P> Clone for Publisher<P> {
    fn clone(&self) -> Self {
        Self {
            context: self.context.clone(),
            _plugin: PhantomData,
        }
    }
}

impl Context {
    pub fn publisher<P: Declared>(&self) -> Publisher<P> {
        Publisher {
            context: self.clone(),
            _plugin: PhantomData,
        }
    }
}

impl<P: Declared> Publisher<P> {
    pub async fn publish_on<S: StreamSpec>(&self, row: &S::Row) -> Result<(), PublishError>
    where
        S::Row: Message,
    {
        if !P::STREAMS.iter().any(|entry| entry.id == S::ID && entry.schema == S::VERSION) {
            return Err(PublishError::UndeclaredStream);
        }
        self.context.calling_context()?;
        let request = harmos_operations::streams::Publish {
            stream: S::ID.to_owned(),
            schema: S::VERSION,
            sequence: None,
            published_unix_micros: None,
            context: None,
            payload: row.encode(),
            following: Vec::new(),
        };
        let published = Host::call::<harmos_operations::streams::Publish>(&self.context, request).await?;
        drop(published);
        Ok(())
    }
}

/* === Public Functions === */

/* === Private Functions === */

/* === Tests === */