liminal_sdk/remote/websocket/
subscription_signals.rs

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

liminal_sdk/remote/websocket/
subscription_signals.rs
