hotfix/session/
session_ref.rs

1use thiserror::Error;
2use tokio::sync::{mpsc, oneshot};
3use tracing::debug;
4
5use crate::config::SessionConfig;
6use crate::message::{OutboundMessage, RawFixMessage};
7use crate::session::Session;
8use crate::session::admin_request::AdminRequest;
9use crate::session::error::{SendError, SendOutcome, SessionCreationError};
10use crate::session::event::{ScheduleResponse, SessionEvent};
11use crate::store::MessageStore;
12use crate::transport::writer::WriterRef;
13use crate::{Application, session};
14
15/// A request to send an outbound message, optionally with confirmation.
16pub(crate) struct OutboundRequest<M> {
17    pub message: M,
18    pub confirm: Option<oneshot::Sender<Result<SendOutcome, SendError>>>,
19}
20
21#[derive(Clone)]
22pub struct InternalSessionRef<Outbound> {
23    pub(crate) event_sender: mpsc::Sender<SessionEvent>,
24    pub(crate) outbound_message_sender: mpsc::Sender<OutboundRequest<Outbound>>,
25    pub(crate) admin_request_sender: mpsc::Sender<AdminRequest>,
26}
27
28impl<Outbound: OutboundMessage> InternalSessionRef<Outbound> {
29    pub fn new(
30        config: SessionConfig,
31        application: impl Application<Outbound = Outbound>,
32        store: impl MessageStore + 'static,
33    ) -> Result<Self, SessionCreationError> {
34        let (event_sender, event_receiver) = mpsc::channel::<SessionEvent>(100);
35        let (outbound_message_sender, outbound_message_receiver) =
36            mpsc::channel::<OutboundRequest<Outbound>>(10);
37        let (admin_request_sender, admin_request_receiver) = mpsc::channel::<AdminRequest>(10);
38        let session = Session::new(config, application, store)?;
39        tokio::spawn(session::run_session(
40            session,
41            event_receiver,
42            outbound_message_receiver,
43            admin_request_receiver,
44        ));
45
46        Ok(Self {
47            event_sender,
48            outbound_message_sender,
49            admin_request_sender,
50        })
51    }
52
53    pub async fn register_writer(&self, writer: WriterRef) -> Result<(), SessionGone> {
54        self.event_sender
55            .send(SessionEvent::Connected(writer))
56            .await?;
57
58        Ok(())
59    }
60
61    pub async fn new_fix_message_received(&self, msg: RawFixMessage) -> Result<(), SessionGone> {
62        self.event_sender
63            .send(SessionEvent::FixMessageReceived(msg))
64            .await?;
65
66        Ok(())
67    }
68
69    pub async fn disconnect(&self, reason: String) -> Result<(), SessionGone> {
70        self.event_sender
71            .send(SessionEvent::Disconnected(reason))
72            .await?;
73
74        Ok(())
75    }
76
77    pub async fn should_reconnect(&self) -> Result<bool, SessionGone> {
78        let (sender, receiver) = oneshot::channel();
79        self.event_sender
80            .send(SessionEvent::ShouldReconnect(sender))
81            .await?;
82        Ok(receiver.await?)
83    }
84
85    pub async fn await_in_schedule(&self) -> Result<ScheduleResponse, SessionGone> {
86        debug!("awaiting in-schedule time");
87        let (sender, receiver) = oneshot::channel::<ScheduleResponse>();
88        self.event_sender
89            .send(SessionEvent::AwaitSchedule(sender))
90            .await?;
91        Ok(receiver.await?)
92    }
93}
94
95#[derive(Debug, Error)]
96#[error("session task terminated")]
97pub struct SessionGone(String);
98
99impl From<mpsc::error::SendError<SessionEvent>> for SessionGone {
100    fn from(err: mpsc::error::SendError<SessionEvent>) -> Self {
101        Self(err.to_string())
102    }
103}
104
105impl From<oneshot::error::RecvError> for SessionGone {
106    fn from(err: oneshot::error::RecvError) -> Self {
107        Self(err.to_string())
108    }
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114    use crate::session::test_utils::create_test_session_ref;
115
116    #[tokio::test]
117    async fn await_in_schedule_returns_in_schedule_when_session_responds_in_schedule() {
118        let (session_ref, mut event_receiver) = create_test_session_ref();
119
120        tokio::spawn(async move {
121            match event_receiver.recv().await {
122                Some(SessionEvent::AwaitSchedule(responder)) => {
123                    let _ = responder.send(ScheduleResponse::InSchedule);
124                }
125                other => panic!("unexpected event: {other:?}"),
126            }
127        });
128
129        let result = session_ref.await_in_schedule().await;
130        assert!(matches!(result, Ok(ScheduleResponse::InSchedule)));
131    }
132
133    #[tokio::test]
134    async fn await_in_schedule_returns_shutdown_when_session_responds_shutdown() {
135        let (session_ref, mut event_receiver) = create_test_session_ref();
136
137        tokio::spawn(async move {
138            match event_receiver.recv().await {
139                Some(SessionEvent::AwaitSchedule(responder)) => {
140                    let _ = responder.send(ScheduleResponse::Shutdown);
141                }
142                other => panic!("unexpected event: {other:?}"),
143            }
144        });
145
146        let result = session_ref.await_in_schedule().await;
147        assert!(matches!(result, Ok(ScheduleResponse::Shutdown)));
148    }
149
150    #[tokio::test]
151    async fn await_in_schedule_returns_err_when_event_channel_closed() {
152        let (session_ref, event_receiver) = create_test_session_ref();
153        drop(event_receiver);
154
155        let result = session_ref.await_in_schedule().await;
156        assert!(matches!(result, Err(SessionGone(_))));
157    }
158}
hotfix/session/session_ref.rs

hotfix/session/
session_ref.rs
