hotfix/session/
session_handle.rs

1use crate::session::admin_request::AdminRequest;
2use crate::session::error::{SendError, SendOutcome, SetNextTargetSeqNumError};
3use crate::session::session_ref::OutboundRequest;
4use crate::session::{InternalSessionRef, SessionInfo};
5use std::num::NonZeroU64;
6use tokio::sync::{mpsc, oneshot};
7
8/// A public handle to the session that can be used to interact with the session.
9///
10/// This wraps a subset of the channels of [`InternalSessionRef`].
11/// Whilst [`InternalSessionRef`] is intended for internal use within the engine,
12/// such as inbound message processing and disconnects, [`SessionHandle`] is public
13/// and only exposes APIs intended for consumers of the engine.
14#[derive(Clone, Debug)]
15pub struct SessionHandle<Outbound> {
16    outbound_message_sender: mpsc::Sender<OutboundRequest<Outbound>>,
17    admin_request_sender: mpsc::Sender<AdminRequest>,
18}
19
20impl<Outbound> SessionHandle<Outbound> {
21    pub async fn get_session_info(&self) -> Result<SessionInfo, SendError> {
22        let (sender, receiver) = oneshot::channel::<SessionInfo>();
23        self.admin_request_sender
24            .send(AdminRequest::RequestSessionInfo(sender))
25            .await?;
26        Ok(receiver.await?)
27    }
28
29    /// Sends a message and waits for confirmation that it was persisted.
30    ///
31    /// Returns `SendOutcome::Sent` with the sequence number if the message was
32    /// successfully persisted and sent, or `SendOutcome::Dropped` if the application
33    /// callback chose to drop the message.
34    pub async fn send(&self, msg: Outbound) -> Result<SendOutcome, SendError> {
35        let (tx, rx) = oneshot::channel();
36        let request = OutboundRequest {
37            message: msg,
38            confirm: Some(tx),
39        };
40        self.outbound_message_sender.send(request).await?;
41        rx.await?
42    }
43
44    /// Sends a message without waiting for confirmation.
45    ///
46    /// This is a fire-and-forget operation. The message will be queued for sending
47    /// but no confirmation is provided about whether it was actually sent.
48    pub async fn send_forget(&self, msg: Outbound) -> Result<(), SendError> {
49        let request = OutboundRequest {
50            message: msg,
51            confirm: None,
52        };
53        self.outbound_message_sender.send(request).await?;
54        Ok(())
55    }
56
57    pub async fn shutdown(&self, reconnect: bool) -> Result<(), SendError> {
58        self.admin_request_sender
59            .send(AdminRequest::InitiateGracefulShutdown { reconnect })
60            .await?;
61        Ok(())
62    }
63
64    pub async fn request_reset_on_next_logon(&self) -> Result<(), SendError> {
65        self.admin_request_sender
66            .send(AdminRequest::ResetSequenceNumbersOnNextLogon)
67            .await?;
68        Ok(())
69    }
70
71    /// Sets the next expected target sequence number.
72    ///
73    /// Permitted only while the session is `Disconnected`. Use this to realign
74    /// after a counterparty-initiated sequence reset without forcing a bilateral
75    /// reset — the peer's subsequent `ResendRequest` is handled by the existing
76    /// resend/gap-fill logic.
77    pub async fn set_next_target_seq_num(
78        &self,
79        seq_num: NonZeroU64,
80    ) -> Result<(), SetNextTargetSeqNumError> {
81        let (responder, receiver) = oneshot::channel();
82        self.admin_request_sender
83            .send(AdminRequest::SetNextTargetSeqNum { seq_num, responder })
84            .await
85            .map_err(|_| SetNextTargetSeqNumError::Send(SendError::SessionGone))?;
86        receiver
87            .await
88            .map_err(|_| SetNextTargetSeqNumError::Send(SendError::SessionGone))?
89    }
90}
91
92impl<M> From<InternalSessionRef<M>> for SessionHandle<M> {
93    fn from(session_ref: InternalSessionRef<M>) -> Self {
94        Self {
95            outbound_message_sender: session_ref.outbound_message_sender.clone(),
96            admin_request_sender: session_ref.admin_request_sender.clone(),
97        }
98    }
99}
hotfix/session/session_handle.rs

hotfix/session/
session_handle.rs
