aves 0.5.0

Typed async application runtime with domain messaging, RPC, and durable workflow automation
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

use std::future::Future;
use std::pin::Pin;

use tokio::sync::oneshot;

use crate::bus::{BusRpcContext, OpaqueRequestEnvelope};

use super::context::ConstructionFacade;

/// Erased asynchronous callback future used by a domain descriptor.
pub type DomainFuture<'a> = Pin<Box<dyn Future<Output = ()> + Send + 'a>>;

/// Erased lifecycle callbacks for one installed domain marker.
pub struct DomainCallbacks<Bus: ?Sized> {
    /// Constructs the marker's filtered services and invokes its `new` method.
    pub new: for<'facade, 'bus> fn(&'facade mut ConstructionFacade<'bus, Bus>) -> DomainFuture<'facade>,
    /// Invokes the marker's static `start` method.
    pub start: fn(BusRpcContext) -> DomainFuture<'static>,
    /// Invokes the marker's static `stop` method.
    pub stop: fn(BusRpcContext) -> DomainFuture<'static>,
}

/// Narrow integration contract implemented by an empty domain marker.
///
/// The implementation supplies erased callbacks which delegate to the
/// marker's existing runtime/domain implementation.  It is deliberately a
/// construction contract rather than another lifecycle API.
pub trait DomainDescriptorFactory<Bus: ?Sized>: Send + Sync + 'static {
    /// Returns the marker's type-erased lifecycle callbacks.
    fn callbacks() -> DomainCallbacks<Bus>;
}

/// Type-erased runtime descriptor for one installed domain.
pub struct DomainDescriptor<Bus: ?Sized> {
    name: &'static str,
    callbacks: DomainCallbacks<Bus>,
}

impl<Bus: ?Sized> DomainDescriptor<Bus> {
    /// Creates a descriptor from marker-owned lifecycle callbacks.
    pub const fn from_callbacks(name: &'static str, callbacks: DomainCallbacks<Bus>) -> Self {
        Self { name, callbacks }
    }

    /// Creates a descriptor for an installed empty domain marker.
    pub fn for_domain<Domain>(name: &'static str) -> Self
    where
        Domain: DomainDescriptorFactory<Bus>,
    {
        Self::from_callbacks(name, Domain::callbacks())
    }

    /// Returns the marker name recorded by the installation macro.
    pub const fn name(&self) -> &'static str {
        self.name
    }

    pub(crate) fn construct<'facade, 'bus>(
        &self,
        facade: &'facade mut ConstructionFacade<'bus, Bus>,
    ) -> DomainFuture<'facade> {
        (self.callbacks.new)(facade)
    }

    pub(crate) fn start(&self, context: BusRpcContext) -> DomainFuture<'static> {
        (self.callbacks.start)(context)
    }

    pub(crate) fn stop(&self, context: BusRpcContext) -> DomainFuture<'static> {
        (self.callbacks.stop)(context)
    }
}

/// Sends one lifecycle request through the normal bus and awaits its typed reply.
pub async fn dispatch_lifecycle_request<Marker, Request, Response, DomainError>(
    context: BusRpcContext,
    request: impl FnOnce(oneshot::Sender<Result<Response, DomainError>>) -> Request,
) -> Option<Result<Response, DomainError>>
where
    Marker: 'static,
    Request: Send + 'static,
    Response: Send + 'static,
    DomainError: Send + 'static,
{
    let (reply_to, reply) = oneshot::channel();
    if context
        .bus_sender()
        .expect("authenticated lifecycle context")
        .send(OpaqueRequestEnvelope::new::<Marker, _>(
            context.jwt().expect("authenticated lifecycle context"),
            request(reply_to),
        ))
        .await
        .is_err()
    {
        return None;
    }
    reply.await.ok()
}