aurelia 0.2.0

Embeddable service mesh for Rust distributed applications.
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

// This file is part of the Aurelia workspace.
// SPDX-FileCopyrightText: 2026 Zivatar Limited
// SPDX-License-Identifier: Apache-2.0

use actix::prelude::{Recipient, SendError};

use crate::ids::ErrorId;
use crate::peering::codec::MessageCodec;
use crate::peering::taberna::Taberna;
use crate::peering::BlobReceiver;

/// Typed Actix delivery envelope for a taberna request.
///
/// The envelope preserves both the decoded application message and any blob receiver carried by
/// the inbound request. Actix mailbox admission is the Aurelia acceptance boundary, so handlers
/// return `()` and application-level failures must travel in application messages.
pub struct ActixTabernaDelivery<M>
where
    M: Send + 'static,
{
    /// Decoded application message produced by the configured [`MessageCodec`].
    pub message: M,
    /// Optional blob receiver attached to the inbound request.
    pub blob_receiver: Option<BlobReceiver>,
}

impl<M> actix::Message for ActixTabernaDelivery<M>
where
    M: Send + 'static,
{
    type Result = ();
}

/// Registration handle for an Actix-backed taberna bridge.
///
/// Dropping the handle aborts the Actix-side bridge task, which drops the underlying
/// [`Taberna`] and schedules deregistration from its parent domus.
pub struct ActixTaberna {
    bridge: tokio::task::JoinHandle<()>,
}

impl ActixTaberna {
    pub(crate) fn new<Codec>(
        taberna: Taberna<Codec>,
        recipient: Recipient<ActixTabernaDelivery<Codec::AppMessage>>,
    ) -> Self
    where
        Codec: MessageCodec + 'static,
        Codec::AppMessage: Send + Sync + 'static,
    {
        let bridge = actix::spawn(run_bridge(taberna, recipient));
        Self { bridge }
    }
}

impl Drop for ActixTaberna {
    fn drop(&mut self) {
        self.bridge.abort();
    }
}

async fn run_bridge<Codec>(
    taberna: Taberna<Codec>,
    recipient: Recipient<ActixTabernaDelivery<Codec::AppMessage>>,
) where
    Codec: MessageCodec + 'static,
    Codec::AppMessage: Send + Sync + 'static,
{
    loop {
        let request = match taberna.next(None).await {
            Ok(request) => request,
            Err(err) if err.kind == ErrorId::ReceiveTimeout => continue,
            Err(_) => break,
        };

        let parts = request.into_parts();
        let delivery = ActixTabernaDelivery {
            message: parts.message,
            blob_receiver: parts.blob_receiver,
        };

        match recipient.try_send(delivery) {
            Ok(()) => {
                parts.completion.accept();
            }
            Err(SendError::Full(_delivery)) => {
                parts.completion.busy();
            }
            Err(SendError::Closed(_delivery)) => {
                parts.completion.taberna_shutdown();
                break;
            }
        }
    }
}

#[cfg(test)]
#[path = "tests/actix_adapter.rs"]
mod tests;