Skip to content

Commit fb726d8

Browse files
committed
Add test case for resending garbled messages
1 parent 9b24905 commit fb726d8

1 file changed

Lines changed: 93 additions & 0 deletions

File tree

crates/hotfix/src/session/outbound.rs

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,3 +117,96 @@ fn log_skipped_admin_messages(begin: u64, end: u64) {
117117
end, "skipped admin message(s) during resend, requesting reset for these"
118118
);
119119
}
120+
121+
#[cfg(test)]
122+
mod tests {
123+
use super::*;
124+
use crate::config::SessionConfig;
125+
use crate::session::ctx::SessionCtx;
126+
use crate::store::Result as StoreResult;
127+
use chrono::{DateTime, Utc};
128+
use hotfix_message::MessageBuilder;
129+
use hotfix_message::dict::Dictionary;
130+
use hotfix_message::message::Config as MessageConfig;
131+
use tokio::sync::mpsc;
132+
133+
#[derive(Clone)]
134+
struct GarbledMessageStore {
135+
messages: Vec<Vec<u8>>,
136+
}
137+
138+
#[async_trait::async_trait]
139+
impl MessageStore for GarbledMessageStore {
140+
async fn add(&mut self, _: u64, _: &[u8]) -> StoreResult<()> {
141+
Ok(())
142+
}
143+
async fn get_slice(&self, _: usize, _: usize) -> StoreResult<Vec<Vec<u8>>> {
144+
Ok(self.messages.clone())
145+
}
146+
fn next_sender_seq_number(&self) -> u64 {
147+
1
148+
}
149+
fn next_target_seq_number(&self) -> u64 {
150+
1
151+
}
152+
async fn increment_sender_seq_number(&mut self) -> StoreResult<()> {
153+
Ok(())
154+
}
155+
async fn increment_target_seq_number(&mut self) -> StoreResult<()> {
156+
Ok(())
157+
}
158+
async fn set_target_seq_number(&mut self, _: u64) -> StoreResult<()> {
159+
Ok(())
160+
}
161+
async fn reset(&mut self) -> StoreResult<()> {
162+
Ok(())
163+
}
164+
fn creation_time(&self) -> DateTime<Utc> {
165+
Utc::now()
166+
}
167+
}
168+
169+
fn create_test_ctx(store: GarbledMessageStore) -> SessionCtx<(), GarbledMessageStore> {
170+
let message_config = MessageConfig::default();
171+
let dictionary = Dictionary::fix44();
172+
let message_builder = MessageBuilder::new(dictionary, message_config).unwrap();
173+
SessionCtx {
174+
config: SessionConfig {
175+
begin_string: "FIX.4.4".to_string(),
176+
sender_comp_id: "SENDER".to_string(),
177+
target_comp_id: "TARGET".to_string(),
178+
data_dictionary_path: None,
179+
connection_host: "localhost".to_string(),
180+
connection_port: 9876,
181+
tls_config: None,
182+
heartbeat_interval: 30,
183+
logon_timeout: 10,
184+
logout_timeout: 2,
185+
reconnect_interval: 30,
186+
reset_on_logon: false,
187+
schedule: None,
188+
},
189+
store,
190+
application: (),
191+
message_builder,
192+
message_config,
193+
}
194+
}
195+
196+
#[tokio::test]
197+
async fn resend_messages_returns_error_for_garbled_stored_message() {
198+
let store = GarbledMessageStore {
199+
messages: vec![b"not a valid FIX message".to_vec()],
200+
};
201+
let mut ctx = create_test_ctx(store);
202+
let (sender, _receiver) = mpsc::channel(10);
203+
let writer = WriterRef::new(sender);
204+
205+
let result = resend_messages(&mut ctx, &writer, 1, 1).await;
206+
207+
assert!(
208+
matches!(result, Err(SessionOperationError::StoredMessageParse(_))),
209+
"expected StoredMessageParse error, got: {result:?}"
210+
);
211+
}
212+
}

0 commit comments

Comments
 (0)