1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
#![feature(impl_trait_in_assoc_type)]
pub mod acceptor;
pub mod application;
pub mod initiator;
pub mod io;
pub mod messages_storage;
mod session;
pub mod session_id;
mod session_state;
pub mod settings;
use std::time::Duration;
use easyfix_messages::{
fields::{FixString, MsgType, UtcTimestamp},
messages::{FixtMessage, Header, Message, Trailer},
};
use settings::Settings;
use tokio::sync::mpsc;
const NO_INBOUND_TIMEOUT_PADDING: Duration = Duration::from_millis(250);
const TEST_REQUEST_THRESHOLD: f32 = 1.2;
use tracing::error;
#[derive(Debug, thiserror::Error)]
pub enum SessionError {
#[error("Never received logon from new connection.")]
LogonNeverReceived,
#[error("Message does not point to any session.")]
UnknownSession,
}
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("I/O error: {0}")]
Io(#[from] std::io::Error),
#[error("Session error: {0}")]
SessionError(SessionError),
}
/// Disconnection reasons.
#[derive(Clone, Copy, Debug)]
pub enum DisconnectReason {
/// Logout requested locally
LocalRequestedLogout,
/// Logout requested remotely
RemoteRequestedLogout,
/// Disconnect forced by User code
UserForcedDisconnect,
/// Received message without MsgSeqNum
MsgSeqNumNotFound,
/// Received message with MsgSeqNum too low
MsgSeqNumTooLow,
/// Invalid logon state
InvalidLogonState,
/// Remote side disconnected
Disconnected,
/// I/O Error
IoError,
}
#[derive(Debug)]
pub(crate) enum SenderMsg {
Msg(Box<FixtMessage>),
Disconnect(DisconnectReason),
}
#[derive(Clone, Debug)]
pub struct Sender {
inner: mpsc::UnboundedSender<SenderMsg>,
}
impl Sender {
/// Create new `Sender` instance.
pub(crate) fn new(writer: mpsc::UnboundedSender<SenderMsg>) -> Sender {
Sender { inner: writer }
}
/// Send FIXT message.
///
/// All header and trailer fields can be also adjusted when handing
/// `FixEvent::AppMsgOut` and `FixEvent::AdmMsgOut`.
///
/// Before serialziation following header fields will be filled:
/// - begin_string (if not empty)
/// - msg_type
/// - sender_comp_id (if not empty)
/// - target_comp_id (if not empty)
/// - sending_time (if eq UtcTimestamp::MIN_UTC)
/// - msg_seq_num (if eq 0)
///
/// The checksum(10) field value is always ignored - it is computed and set
/// after serialziation.
pub fn send_raw(&self, msg: Box<FixtMessage>) -> Result<(), Box<FixtMessage>> {
if let Err(msg) = self.inner.send(SenderMsg::Msg(msg)) {
match msg.0 {
SenderMsg::Msg(msg) => {
error!(
"failed to send {:?}<{}> message, receiver closed or dropped",
msg.msg_type(),
msg.msg_type().as_fix_str()
);
Err(msg)
}
SenderMsg::Disconnect(_) => unreachable!(),
}
} else {
Ok(())
}
}
/// Send FIX message.
///
/// FIXT message will be constructed internally using default values
/// for Header and Trailer.
///
/// All header and trailer fields can be also adjusted when handing
/// `FixEvent::AppMsgOut` and `FixEvent::AdmMsgOut`.
pub fn send(&self, msg: Box<Message>) -> Result<(), Box<FixtMessage>> {
let msg = Box::new(FixtMessage {
header: Box::new(new_header(msg.msg_type())),
body: msg,
trailer: Box::new(new_trailer()),
});
self.send_raw(msg)
}
/// Send disconnect message.
///
/// Output stream will close output queue so no more message can be send
/// after this one.
pub(crate) fn disconnect(&self, reason: DisconnectReason) {
if self.inner.send(SenderMsg::Disconnect(reason)).is_err() {
error!("failed to disconnect, receiver closed or dropped");
}
}
}
pub fn new_header(msg_type: MsgType) -> Header {
// XXX: all required fields overwritten before serialization (if not set)
Header {
begin_string: FixString::new(),
msg_type,
sending_time: UtcTimestamp::MIN_UTC,
..Default::default()
}
}
pub fn new_trailer() -> Trailer {
// XXX: all required fields overwritten before serialization
Trailer::default()
}