liminal-sdk 0.19.1

//! Blocking reception and an independently held teardown signal.

use super::{DeliveredMessage, Shutdown, SubscriptionStream};
use crate::SdkError;
use crate::remote::mailbox::{BlockingTake, ReaderEnd};
use alloc::format;
use alloc::sync::Arc;

impl SubscriptionStream {
    /// Blocks on a delivery or the reader's terminal, without a timer.
    ///
    /// # Errors
    /// Returns the server's terminal, reader failure or local close in words.
    pub fn recv(&self) -> Result<DeliveredMessage, SdkError> {
        match self.mailbox.take_blocking(false)? {
            BlockingTake::Item(message) => Ok(message),
            BlockingTake::Ended(ReaderEnd::Failed(error) | ReaderEnd::PeerClosed(error)) => {
                Err(error)
            }
            BlockingTake::Abandoned | BlockingTake::Ended(ReaderEnd::Stopped) => {
                Err(SdkError::Connection {
                    description: "subscription receive failed: the subscription was closed".into(),
                })
            }
        }
    }

    /// A signal that releases a blocked consumer and socket reader.
    /// The owning stream still joins the reader when dropped.
    ///
    /// # Errors
    /// Returns the socket-clone failure, without credentials.
    pub fn close_handle(
        &self,
    ) -> Result<impl Fn() -> Result<(), SdkError> + Send + Sync + use<>, SdkError> {
        let socket = self
            .writer
            .try_clone()
            .map_err(|error| SdkError::Connection {
                description: format!("failed to clone subscription close signal: {error}"),
            })?;
        let mailbox = Arc::clone(&self.mailbox);
        Ok(move || {
            let abandoned = mailbox.abandon();
            let shutdown = socket
                .shutdown(Shutdown::Both)
                .or_else(|error| {
                    if error.kind() == std::io::ErrorKind::NotConnected {
                        Ok(())
                    } else {
                        Err(error)
                    }
                })
                .map_err(|error| SdkError::Connection {
                    description: format!("failed to close subscription socket: {error}"),
                });
            abandoned.and(shutdown)
        })
    }
}

/// One delivered message, whichever transport carried it: the WebSocket
/// sibling's fields are the TCP message's (AGT-006 R2).
impl From<crate::WebSocketDeliveredMessage> for DeliveredMessage {
    fn from(message: crate::WebSocketDeliveredMessage) -> Self {
        Self {
            delivery_seq: message.delivery_seq(),
            schema_id: message.schema_id(),
            message: message.message(),
            payload: message.into_payload(),
        }
    }
}