liminal_server/agents/
admitted_services.rs

1//! Carries the admitted session into the existing wire response builders.
2
3use crate::ServerError;
4use crate::auth_lys::session::LysSession;
5use crate::server::connection::ConnectionConversation;
6use crate::server::connection::channel_registry::ChannelAccessError;
7use crate::server::connection::services::{
8    ChannelOperation, ConnectionServices, ConnectionSubscription, PublishOutcome,
9};
10use liminal::protocol::{MessageEnvelope, SchemaId};
11
12/// A borrowed adapter, never a second credential or a retained session cache.
13pub struct AdmittedServices<'a> {
14    connection: u64,
15    services: &'a dyn ConnectionServices,
16    session: Option<&'a dyn LysSession>,
17}
18
19impl<'a> AdmittedServices<'a> {
20    /// Borrows exactly the identity proved by this connection's Connect.
21    #[must_use]
22    pub const fn new(
23        connection: u64,
24        services: &'a dyn ConnectionServices,
25        session: Option<&'a dyn LysSession>,
26    ) -> Self {
27        Self {
28            connection,
29            services,
30            session,
31        }
32    }
33}
34
35impl std::fmt::Debug for AdmittedServices<'_> {
36    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        out.debug_struct("AdmittedServices")
38            .field("connection", &self.connection)
39            .finish_non_exhaustive()
40    }
41}
42
43impl ConnectionServices for AdmittedServices<'_> {
44    fn publish(
45        &self,
46        connection: u64,
47        channel: &str,
48        envelope: &MessageEnvelope,
49        key: Option<&str>,
50    ) -> Result<PublishOutcome, ServerError> {
51        if connection != self.connection {
52            return Err(
53                crate::agents::error::AgentError::Invalid("publish connection changed").into(),
54            );
55        }
56        self.services
57            .publish_identified(connection, self.session, channel, envelope, key)
58    }
59    fn subscribe(
60        &self,
61        channel: &str,
62        schemas: &[SchemaId],
63        install: Option<liminal::channel::InboxInstall>,
64    ) -> Result<ConnectionSubscription, ServerError> {
65        self.services
66            .subscribe_identified(self.connection, self.session, channel, schemas, install)
67    }
68    fn unsubscribe(&self, subscription: ConnectionSubscription) -> Result<(), ServerError> {
69        self.services.unsubscribe(subscription)
70    }
71    fn open_conversation(
72        &self,
73        id: u64,
74        subject: &str,
75    ) -> Result<ConnectionConversation, ServerError> {
76        self.services.open_conversation(id, subject)
77    }
78    fn conversation_message(
79        &self,
80        conversation: &ConnectionConversation,
81        envelope: &MessageEnvelope,
82        op_id: Option<u64>,
83    ) -> Result<(), ServerError> {
84        self.services
85            .conversation_message(conversation, envelope, op_id)
86    }
87    fn close_conversation(&self, conversation: ConnectionConversation) -> Result<(), ServerError> {
88        self.services.close_conversation(conversation)
89    }
90    fn flush_durable_state(&self) -> Result<(), ServerError> {
91        self.services.flush_durable_state()
92    }
93    fn supports_channel_operations(&self) -> bool {
94        self.services.supports_channel_operations()
95    }
96    fn admit_channel(
97        &self,
98        operation: ChannelOperation,
99        channel: &str,
100    ) -> Result<(), ChannelAccessError> {
101        self.services.admit_channel(operation, channel)
102    }
103}
liminal_server/agents/admitted_services.rs

liminal_server/agents/
admitted_services.rs
