liminal-server 0.25.0

Standalone server for the liminal messaging bus
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

//! Carries the admitted session into the existing wire response builders.

use crate::ServerError;
use crate::auth_lys::session::LysSession;
use crate::server::connection::ConnectionConversation;
use crate::server::connection::channel_registry::ChannelAccessError;
use crate::server::connection::services::{
    ChannelOperation, ConnectionServices, ConnectionSubscription, PublishOutcome,
};
use liminal::protocol::{MessageEnvelope, SchemaId};

/// A borrowed adapter, never a second credential or a retained session cache.
pub struct AdmittedServices<'a> {
    connection: u64,
    services: &'a dyn ConnectionServices,
    session: Option<&'a dyn LysSession>,
}

impl<'a> AdmittedServices<'a> {
    /// Borrows exactly the identity proved by this connection's Connect.
    #[must_use]
    pub const fn new(
        connection: u64,
        services: &'a dyn ConnectionServices,
        session: Option<&'a dyn LysSession>,
    ) -> Self {
        Self {
            connection,
            services,
            session,
        }
    }
}

impl std::fmt::Debug for AdmittedServices<'_> {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("AdmittedServices")
            .field("connection", &self.connection)
            .finish_non_exhaustive()
    }
}

impl ConnectionServices for AdmittedServices<'_> {
    fn publish(
        &self,
        connection: u64,
        channel: &str,
        envelope: &MessageEnvelope,
        key: Option<&str>,
    ) -> Result<PublishOutcome, ServerError> {
        if connection != self.connection {
            return Err(
                crate::agents::error::AgentError::Invalid("publish connection changed").into(),
            );
        }
        self.services
            .publish_identified(connection, self.session, channel, envelope, key)
    }
    fn subscribe(
        &self,
        channel: &str,
        schemas: &[SchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        self.services
            .subscribe_identified(self.connection, self.session, channel, schemas, install)
    }
    fn unsubscribe(&self, subscription: ConnectionSubscription) -> Result<(), ServerError> {
        self.services.unsubscribe(subscription)
    }
    fn open_conversation(
        &self,
        id: u64,
        subject: &str,
    ) -> Result<ConnectionConversation, ServerError> {
        self.services.open_conversation(id, subject)
    }
    fn conversation_message(
        &self,
        conversation: &ConnectionConversation,
        envelope: &MessageEnvelope,
        op_id: Option<u64>,
    ) -> Result<(), ServerError> {
        self.services
            .conversation_message(conversation, envelope, op_id)
    }
    fn close_conversation(&self, conversation: ConnectionConversation) -> Result<(), ServerError> {
        self.services.close_conversation(conversation)
    }
    fn flush_durable_state(&self) -> Result<(), ServerError> {
        self.services.flush_durable_state()
    }
    fn supports_channel_operations(&self) -> bool {
        self.services.supports_channel_operations()
    }
    fn admit_channel(
        &self,
        operation: ChannelOperation,
        channel: &str,
    ) -> Result<(), ChannelAccessError> {
        self.services.admit_channel(operation, channel)
    }
}