liminal_sdk/remote/tcp/subscription/
subscription_signals.rs

1//! Blocking reception and an independently held teardown signal.
2
3use super::{DeliveredMessage, Shutdown, SubscriptionStream};
4use crate::SdkError;
5use crate::remote::mailbox::{BlockingTake, ReaderEnd};
6use alloc::format;
7use alloc::sync::Arc;
8
9impl SubscriptionStream {
10    /// Blocks on a delivery or the reader's terminal, without a timer.
11    ///
12    /// # Errors
13    /// Returns the server's terminal, reader failure or local close in words.
14    pub fn recv(&self) -> Result<DeliveredMessage, SdkError> {
15        match self.mailbox.take_blocking(false)? {
16            BlockingTake::Item(message) => Ok(message),
17            BlockingTake::Ended(ReaderEnd::Failed(error) | ReaderEnd::PeerClosed(error)) => {
18                Err(error)
19            }
20            BlockingTake::Abandoned | BlockingTake::Ended(ReaderEnd::Stopped) => {
21                Err(SdkError::Connection {
22                    description: "subscription receive failed: the subscription was closed".into(),
23                })
24            }
25        }
26    }
27
28    /// A signal that releases a blocked consumer and socket reader.
29    /// The owning stream still joins the reader when dropped.
30    ///
31    /// # Errors
32    /// Returns the socket-clone failure, without credentials.
33    pub fn close_handle(
34        &self,
35    ) -> Result<impl Fn() -> Result<(), SdkError> + Send + Sync + use<>, SdkError> {
36        let socket = self
37            .writer
38            .try_clone()
39            .map_err(|error| SdkError::Connection {
40                description: format!("failed to clone subscription close signal: {error}"),
41            })?;
42        let mailbox = Arc::clone(&self.mailbox);
43        Ok(move || {
44            let abandoned = mailbox.abandon();
45            let shutdown = socket
46                .shutdown(Shutdown::Both)
47                .or_else(|error| {
48                    if error.kind() == std::io::ErrorKind::NotConnected {
49                        Ok(())
50                    } else {
51                        Err(error)
52                    }
53                })
54                .map_err(|error| SdkError::Connection {
55                    description: format!("failed to close subscription socket: {error}"),
56                });
57            abandoned.and(shutdown)
58        })
59    }
60}
61
62/// One delivered message, whichever transport carried it: the WebSocket
63/// sibling's fields are the TCP message's (AGT-006 R2).
64impl From<crate::WebSocketDeliveredMessage> for DeliveredMessage {
65    fn from(message: crate::WebSocketDeliveredMessage) -> Self {
66        Self {
67            delivery_seq: message.delivery_seq(),
68            schema_id: message.schema_id(),
69            message: message.message(),
70            payload: message.into_payload(),
71        }
72    }
73}
liminal_sdk/remote/tcp/subscription/subscription_signals.rs

liminal_sdk/remote/tcp/subscription/
subscription_signals.rs
